flume應(yīng)用案例
1 flume用法
flume的使用非常簡(jiǎn)單，只需書寫一個(gè)配置文件，在配置文件中描述source，channel和sink的具體實(shí)現(xiàn)，然后運(yùn)行一個(gè)agent的實(shí)例赊窥，在運(yùn)行agent實(shí)例的過程中會(huì)讀取配置文件中的內(nèi)容，這樣flume就會(huì)采集數(shù)據(jù)了狸页。
配置文件編寫規(guī)則
1 整體描述agent中的sources、sink芍耘、channel：
#其中a1為agent的名字腹侣，r1為source名稱，k1為sink名稱齿穗，c1為channel名稱
a1.sources = r1
a1.sinks = k1
a1.channels = c1
2 agent中的source傲隶、sink和channel的具體實(shí)現(xiàn)
需要指定source的類型，即source是接收文件還是接受http的還是接受thrift的窃页；對(duì)于sink同理跺株，需要指定結(jié)果是輸出到hdfs還是和base上等复濒。對(duì)于channel需要指定使用內(nèi)存還是文件進(jìn)行緩存。
#描述和配置source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
#描述和配置sink
a1.sinks.k1.type = logger
#描述和配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
3 通過channel將source與sink連接起來
#通過channel將source與sink連接起來
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
4 在shell中啟動(dòng)agent
$ bin/flume-ng agent -n $agent_name -c conf -f conf/flume-conf.properties.template -Dflume.root.logger=DEBUG,console
參數(shù)說明
-n : 制定agent的名稱（與配置文件中的agent名字一樣）
-c : 指定了flume中配置文件的目錄
-f : 制定配置文件
-Dflume.root.logger=DEBUG,console : 設(shè)置日志等級(jí)
2 案例一
案例描述
監(jiān)聽一個(gè)指定的網(wǎng)絡(luò)端口乒省，即只要應(yīng)用程序向這個(gè)端口寫入數(shù)據(jù)的時(shí)候巧颈，這個(gè)source組建就可以獲取到信息。三個(gè)組建的類型如下：
source : NetCat
sink : logger
channel : memory
配置文件
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = 192.168.32.130
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transctionCapacity = 100
#Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動(dòng)flume agent a1服務(wù)端
$flume-ng agent -n a1 -c ../conf -f ../conf/netcat.conf -Dflume.root.logger=DEBUG,console
在另一個(gè)終端下使用netCat發(fā)送數(shù)據(jù)
$nc localhost 44444
然后在下面隨意輸入內(nèi)容進(jìn)行測(cè)試袖扛。
3 案例二
案例描述
監(jiān)聽一個(gè)指定的網(wǎng)絡(luò)端口砸泛，然后將數(shù)據(jù)寫入hdfs上。三個(gè)組建的類型如下：
source : NetCat
sink : hdfs
channel : file
配置文件
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = 192.168.32.130
a1.sources.r1.port = 44444
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://hadoop200:9000/dataoutput
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.rollInterval = 10
a1.sinks.k1.hdfs.rollSize = 0
a1.sinks.k1.hdfs..rollCount = 0
a1.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# Use a channel which buffers events in file
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /usr/flume/checkpoint
a1.channels.c1.dataDirs = /usr/flume/data
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動(dòng)flume agent a1服務(wù)端
$flume-ng agent -n a1 -c ../conf -f ../conf/netcat.conf -Dflume.root.logger=DEBUG,console
在另一個(gè)終端下使用netCat發(fā)送數(shù)據(jù)
$nc localhost 44444
然后在下面隨意輸入內(nèi)容進(jìn)行測(cè)試蛆封，測(cè)試結(jié)果如下圖：
4 案例三
案例描述
監(jiān)聽一個(gè)指定的目錄唇礁，只要應(yīng)用程序向這個(gè)目錄中添加新的文件，source組件就可以獲取到消息惨篱，并解析該文件的內(nèi)容盏筐，然后寫入到channel。寫入完成后砸讳，標(biāo)記該文件已完成或者刪除該文件琢融。三個(gè)組建的類型如下：
source : Spooling Directory
sink : logger
channel : memory
Spooling Directory Source注意事項(xiàng)：
1.正在拷貝的文件不能再進(jìn)行編輯，否則flume就會(huì)在log中報(bào)錯(cuò)并停掉進(jìn)程簿寂。
2.不能將具有相同文件名的文件拷貝到這個(gè)目錄漾抬，否則flume就會(huì)在log中報(bào)錯(cuò)并停掉進(jìn)程。
配置文件
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /usr/local/datainput
a1.sources.r1.fileHeader = true
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = timestamp
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動(dòng)flume agent a1服務(wù)端
$flume-ng agent -n a1 -c ../conf -f ../conf/netcat.conf -Dflume.root.logger=DEBUG,console
測(cè)試
向指定的文件夾/usr/local/datainput添加數(shù)據(jù)觀察控制臺(tái)的變化以及該目錄文件的變化常遂。
5 案例四
案例描述
監(jiān)聽一個(gè)指定的目錄纳令，只要應(yīng)用程序向這個(gè)目錄中添加新的文件，source組件就可以獲取到消息烈钞，并解析該文件的內(nèi)容泊碑，然后寫入到channel坤按。寫入完成后毯欣，標(biāo)記該文件已完成或者刪除該文件。三個(gè)組建的類型如下：
source : Spooling Directory
sink : hdfs
channel : file
配置文件
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /usr/local/datainput
a1.sources.r1.fileHeader = true
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = timestamp
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://hadoop200:8020/dataoutput
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.roolInterval = 10
a1.sinks.k1.hdfs.roolSize = 0
a1.sinks.k1.hdfs.roolCount = 0
a1.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# Use a channel which buffers in file
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /usr/flume/checkpoint
a1.channels.c1.dataDirs = /usr/flume/data
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
啟動(dòng)flume agent a1服務(wù)端
$flume-ng agent -n a1 -c ../conf -f ../conf/netcat.conf -Dflume.root.logger=DEBUG,console
測(cè)試
向指定的文件夾/usr/local/datainput添加數(shù)據(jù)臭脓，然后通過web觀察hdfs的變化以及該目錄文件的變化酗钞。
