【轉載】Flume學習之路（三）Flume的配置方式

時間 2019-11-09

標籤轉載 flume 學習之路配置方式欄目日誌分析简体版

原文原文鏈接

討論QQ：1586558083html

目錄web

正文apache

回到頂部

1、單一代理流配置

1.1　官網介紹

http://flume.apache.org/FlumeUserGuide.html#avro-source緩存

經過一個通道未來源和接收器連接。須要列出源，接收器和通道，爲給定的代理，而後指向源和接收器及通道。一個源的實例能夠指定多個通道，但只能指定一個接收器實例。格式以下：服務器

實例解析：一個代理名爲agent_foo，外部經過avro客戶端，而且發送數據經過內存通道給hdfs。在配置文件foo.config的可能看起來像這樣：app

案例說明：這將使事件流從avro-appserver-src-1到hdfs-sink-1經過內存通道mem-channel-1。當代理開始foo.config做爲其配置文件，它會實例化流。curl

配置單個組件tcp

定義流以後，須要設置每一個源，接收器和通道的屬性。能夠分別設定組件的屬性值。ide

「type」屬性必須爲每一個組件設置，以瞭解它須要什麼樣的對象。每一個源，接收器和通道類型有其本身的一套，它所需的性能，以實現預期的功能。全部這些，必須根據須要設置。在前面的例子中，從hdfs-sink-1中的流到HDFS，經過內存通道mem-channel-1的avro-appserver-src-1源。下面是一個例子，顯示了這些組件的配置。oop

1.2　測試示例（一）

經過flume來監控一個目錄，當目錄中有新文件時，將文件內容輸出到控制檯。

#配置一個agent，agent的名稱能夠自定義（如a1）
#指定agent的sources（如s1）、sinks（如k1）、channels（如c1）
#分別指定agent的sources，sinks,channels的名稱 名稱能夠自定義
a1.sources = s1  
a1.sinks = k1  
a1.channels = c1  
   
#描述source
#配置目錄scource
a1.sources.s1.type =spooldir  
a1.sources.s1.spoolDir =/home/hadoop/logs  
a1.sources.s1.fileHeader= true  
a1.sources.s1.channels =c1  
   
#配置sink 
a1.sinks.k1.type = logger  
a1.sinks.k1.channel = c1  
   
#配置channel(內存作緩存)
a1.channels.c1.type = memory

啓動命令

[hadoop@hadoop1 ~]$ flume-ng agent --conf conf --conf-file /home/hadoop/apps/flume/examples/case_spool.properties --name a1 -Dflume.root.logger=INFO,console

將123.log移動到logs目錄

運行結果

1.3　測試案例（二）

案例2：實時模擬從web服務器中讀取數據到hdfs中

此處使用exec source詳細參考http://www.cnblogs.com/qingyunzong/p/8995554.html

裏面的2.3Exec Source介紹

回到頂部

2、單代理多流配置

單個Flume代理能夠包含幾個獨立的流。你能夠在一個配置文件中列出多個源，接收器和通道。這些組件能夠鏈接造成多個流。

能夠鏈接源和接收器到其相應的通道，設置兩個不一樣的流。例如，若是須要設置一個agent_foo代理兩個流，一個從外部Avro客戶端到HDFS，另一個是tail的輸出到Avro接收器，而後在這裏是作一個配置

2.1　官方案例

回到頂部

3、配置多代理流程

設置一個多層的流，須要有一個指向下一跳avro源的第一跳的avro 接收器。這將致使第一Flume代理轉發事件到下一個Flume代理。例如，若是按期發送的文件，每一個事件（1文件）AVRO客戶端使用本地Flume 代理，那麼這個當地的代理能夠轉發到另外一個有存儲的代理。

配置以下

3.1　官方案例

這裏鏈接從weblog-agent的avro-forward-sink 到hdfs-agent的avro-collection-source收集源。最終結果從外部源的appserver最終存儲在HDFS的事件。

3.2　測試案例

case_avro.properties

a1.sources = s1
a1.sinks = k1
a1.channels = c1

a1.sources.s1.type = avro a1.sources.s1.channels = c1 a1.sources.s1.bind = 192.168.123.102 a1.sources.s1.port = 22222

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

a1.sinks.k1.type = logger
a1.sinks.k1.channel = c1

case_avro_sink.properties

a2.sources = s1
a2.sinks = k1
a2.channels = c1
 
a2.sources.s1.type = syslogtcp
a2.sources.s1.channels = c1
a2.sources.s1.host = 192.168.123.102
a2.sources.s1.port = 33333
 
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100
 
a2.sinks.k1.type = avro a2.sinks.k1.hostname = 192.168.123.102 a2.sinks.k1.port = 22222 a2.sinks.k1.channel = c1

說明：case_avro_sink.properties是前面的Agent，case_avro.properties是後面的Agent

#先啓動Avro的Source,監聽端口

[hadoop@hadoop1 ~]$ flume-ng agent --conf conf --conf-file ~/apps/flume/examples/case_avro.properties --name a1 -Dflume.root.logger=DEBUG,console -Dorg.apache.flume.log.printconfig=true -Dorg.apache.flume.log.rawdata=true

#再啓動Avro的Sink

flume-ng agent --conf conf --conf-file ~/apps/flume/examples/case_avro_sink.properties --name a2 -Dflume.root.logger=DEBUG,console -Dorg.apache.flume.log.printconfig=true -Dorg.apache.flume.log.rawdata=true

能夠看到已經創建鏈接

#在Avro Sink上生成測試log

[hadoop@hadoop1 ~]$ echo "hello flume avro sink" | nc 192.168.123.102 33333

查看其它結果

回到頂部

4、多路複用流

Flume支持扇出流從一個源到多個通道。有兩種模式的扇出，複製和複用。在複製流的事件被髮送到全部的配置通道。在複用的狀況下，事件被髮送到合格的渠道只有一個子集。扇出流，須要指定源和扇出通道的規則。這是經過添加一個通道「選擇」，能夠複製或複用。再進一步指定選擇的規則，若是它是一個多路。若是你不指定一個選擇，則默認狀況下它複製。

複用的選擇集的屬性進一步分叉。這須要指定一個事件屬性映射到一組通道。選擇配置屬性中的每一個事件頭檢查。若是指定的值相匹配，那麼該事件被髮送到全部的通道映射到該值。若是沒有匹配，那麼該事件被髮送到設置爲默認配置的通道。

映射容許每一個值通道能夠重疊。默認值能夠包含任意數量的通道。下面的示例中有一個單一的流複用兩條路徑。代理有一個單一的avro源和鏈接道兩個接收器的兩個通道。

4.1　官方案例

「State」做爲Header的選擇檢查。若是值是「CA」，而後將其發送到mem-channel-1，若是它的「AZ」的，那麼jdbc- channel-2，若是它的「NY」那麼發到這兩個。若是「State」頭未設置或不匹配的任何三個，而後去默認的mem-channel-1通道。

4.2　測試案例（一）複製

case_replicate_sink.properties

a1.sources = s1
a1.sinks = k1 k2
a1.channels = c1 c2

a1.sources.s1.type = syslogtcp
a1.sources.s1.channels = c1 c2
a1.sources.s1.host = 192.168.123.102
a1.sources.s1.port = 6666
a1.sources.s1.selector.type = replicating

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100

a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.123.102
a1.sinks.k1.port = 7777
a1.sinks.k1.channel = c1

a1.sinks.k1.type = avro
a1.sinks.k1.hostname = 192.168.123.102
a1.sinks.k1.port = 7777
a1.sinks.k1.channel = c2

case_replicate_s1.properties

a2.sources = s1
a2.sinks = k1
a2.channels = c1
 
a2.sources.s1.type = avro
a2.sources.s1.channels = c1
a2.sources.s1.host = 192.168.123.102
a2.sources.s1.port = 7777
 
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100
 
a2.sinks.k1.type = logger
a2.sinks.k1.channel = c1

case_replicate_s2.properties

a3.sources = s1
a3.sinks = k1
a3.channels = c1

a3.sources.s1.type = avro
a3.sources.s1.channels = c1
a3.sources.s1.host = 192.168.123.102
a3.sources.s1.port = 7777

a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100

a3.sinks.k1.type = logger
a3.sinks.k1.channel = c1

#先啓動Avro的Source,監聽端口

flume-ng agent --conf conf --conf-file ~/apps/flume/examples/case_replicate_s1.properties --name a2 -Dflume.root.logger=DEBUG,console -Dorg.apache.flume.log.printconfig=true -Dorg.apache.flume.log.rawdata=true

flume-ng agent --conf conf --conf-file ~/apps/flume/examples/case_replicate_s2.properties --name a3 -Dflume.root.logger=DEBUG,console -Dorg.apache.flume.log.printconfig=true -Dorg.apache.flume.log.rawdata=true

#再啓動Avro的Sink

flume-ng agent --conf conf --conf-file ~/apps/flume/examples/case_replicate_sink.properties --name a1 -Dflume.root.logger=DEBUG,console -Dorg.apache.flume.log.printconfig=true -Dorg.apache.flume.log.rawdata=true

#生成測試log

echo "hello via channel selector" | nc 192.168.123.102 6666

4.3　測試案例（二）複用

case_multi_sink.properties

#2個channel和2個sink的配置文件
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
 
# Describe/configure the source
a1.sources.r1.type = org.apache.flume.source.http.HTTPSource
a1.sources.r1.port = 5140
a1.sources.r1.host = 0.0.0.0
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.channels = c1 c2
 
a1.sources.r1.selector.header = state
a1.sources.r1.selector.mapping.CZ = c1
a1.sources.r1.selector.mapping.US = c2
a1.sources.r1.selector.default = c1
 
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = 172.25.4.23
a1.sinks.k1.port = 4545
 
a1.sinks.k2.type = avro
a1.sinks.k2.channel = c2
a1.sinks.k2.hostname = 172.25.4.33
a1.sinks.k2.port = 4545
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
 
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100

case_ multi _s1.properties

# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1
 
# Describe/configure the source
a2.sources.r1.type = avro
a2.sources.r1.channels = c1
a2.sources.r1.bind = 172.25.4.23
a2.sources.r1.port = 4545
 
# Describe the sink
a2.sinks.k1.type = logger
 a2.sinks.k1.channel = c1

# Use a channel which buffers events in memory
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100

case_ multi _s2.properties

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c1
 
# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.channels = c1
a3.sources.r1.bind = 172.25.4.33
a3.sources.r1.port = 4545
 
# Describe the sink
a3.sinks.k1.type = logger
 a3.sinks.k1.channel = c1

# Use a channel which buffers events in memory
a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100

#先啓動Avro的Source,監聽端口

flume-ng agent -c . -f case_ multi _s1.conf -n a2 -Dflume.root.logger=INFO,console

flume-ng agent -c . -f case_ multi _s2.conf -n a3 -Dflume.root.logger=INFO,console

#再啓動Avro的Sink

flume-ng agent -c . -f case_multi_sink.conf -n a1-Dflume.root.logger=INFO,console

#根據配置文件生成測試的header 爲state的POST請求

curl -X POST -d '[{ "headers" :{"state" : "CZ"},"body" : "TEST1"}]' http://localhost:5140

curl -X POST -d '[{ "headers" :{"state" : "US"},"body" : "TEST2"}]' http://localhost:5140

curl -X POST -d '[{ "headers" :{"state" : "SH"},"body" : "TEST3"}]' http://localhost:5140

轉載自 https://www.cnblogs.com/qingyunzong/p/8996155.html