一步兩步,學習大資料（五）——flume的介紹、配置以及使用
阿新 • • 發佈：2019-01-04
大資料的業務處理中，資料採集佔據重要的地位，而在網際網路中大量資料產生的來源之一便是網路日誌。flume是分散式的日誌收集系統，它將各個伺服器中的資料收集起來並送到指定的地方去，可以是檔案、可以是hdfs。有關flume架構更加詳細的介紹大家可以參考 安靜的技術控 Flume架構以及應用介紹
今天小編給大家介紹的是，flume的安裝、配置、以及一些簡單的使用，帶大家初始flume
準備工作
centos7
apache-flume-1.8.0-bin.tar.gz
安裝
注：小編選擇的目錄，大家可以根據自己的需要進行選擇。
1. 通過Xftp把.gz包上傳到虛擬機器
2. tar zxvf apache-flume-1.8.0-bin.tar.gz
3. mv apache-flume-1.8.0 /usr/share/flume
4. 配置flume環境變數
開啟 vi /etc/profile
新增#Flume
export FLUME_HOME=/usr/share/flume
export PATH=$PATH:$FLUME_HOME/bin 儲存退出
執行source /etc/profile
執行 flume-ng version 輸出版本資訊表示安裝正確。
具體案例
案例一： NetCat Source：監聽一個指定的網路埠，即只要應用程式向這個端口裡面寫資料，這個source元件就可以獲取到資訊。
flume官網中NetCat Source描述：
Property Name Default Description
channels –
type 配置檔案：
# 指定Agent的元件名稱（a），一個程序
a.sources=r1
a.channels=c1
a.sinks=k1
a.sources.r1.type=netcat
a.sources.r1.bind 啟動flume agent a 服務端：
flume-ng agent -n a1 -c ../conf -f ../conf/netcat.conf -Dflume.root.logger=DEBUG,console
#-Dflume.root.logger=DEBUG,console 設定控制檯列印
#telnet master 8888 2334/hello/1232
案例二 ：NetCat Source：監聽一個指定的網路埠，即只要應用程式向這個端口裡面寫資料，這個source元件就可以獲取到資訊。 其中 Sink：hdfs Channel：file (相比於案例1的兩個變化)
配置檔案
# Name the components on this agent
a.sources = r1
a.sinks = k1
a.channels = c1
# Describe/configure the source
a.sources.r1.type = netcat
a.sources.r1.bind = master
a.sources.r1.port = 8888
# Describe the sink
a.sinks.k1.type = hdfs
#指定hdfs地址中的輸出目錄
a.sinks.k1.hdfs.path = hdfs://master:9000/output
a.sinks.k1.hdfs.writeFormat = Text
a.sinks.k1.hdfs.fileType = DataStream
a.sinks.k1.hdfs.rollInterval = 10
a.sinks.k1.hdfs.rollSize = 0
a.sinks.k1.hdfs.rollCount = 0
a.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
a.sinks.k1.hdfs.useLocalTimeStamp = true
# Use a channel which buffers events in file
a.channels.c1.type = file
a.channels.c1.checkpointDir = /usr/flume/checkpoint
a.channels.c1.dataDirs = /usr/flume/data
# Bind the source and sink to the channel
a.sources.r1.channels = c1
a.sinks.k1.channel = c1啟動flume agent a 服務端：
flume-ng agent -c conf -f flume-hdfs-test01.properties -name a -Dflume.root.logger=INFO,console
在HDFS中檢視flume收集到的日誌資料： ：
telnet master 8888
輸入測試資料，如：123
在hdfs的output目錄中可以看到目錄中多出一個以時間戳命名的檔案，檔案中寫入了你的測試資料（123）案例3：Spooling Directory Source：監聽一個指定的目錄，即只要應用程式向這個指定的目錄中新增新的檔案，source元件就可以獲取到該資訊，並解析該檔案的內容，然後寫入到channle。寫入完成後，標記該檔案已完成或者刪除該檔案。其中 Sink：logger Channel：memory
flume官網中Spooling Directory Source描述：
Property Name Default Description
channels –
type – The component type name, needs to be spooldir.
spoolDir – Spooling Directory Source監聽的目錄
fileSuffix .COMPLETED 檔案內容寫入到channel之後，標記該檔案
deletePolicy never 檔案內容寫入到channel之後的刪除策略: never or immediate
fileHeader false Whether to add a header storing the absolute path filename.
ignorePattern ^$ Regular expression specifying which files to ignore (skip)
interceptors – 指定傳輸中event的head(頭資訊)，常用timestampSpooling Directory Source的兩個注意事項：
①If a file is written to after being placed into the spooling directory, Flume will print an error to its log file and stop processing.
即：拷貝到spool目錄下的檔案不可以再開啟編輯
②If a file name is reused at a later time, Flume will print an error to its log file and stop processing.
即：不能將具有相同檔名字的檔案拷貝到這個目錄下配置檔案：
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
啟動flume agent a1 服務端
flume-ng agent -c conf -f flume-nect-test02.properties -name a1 -Dflume.root.logger=INFO,console
控制檯列印：
從控制檯顯示的結果可以看出event的頭資訊中包含了時間戳資訊。
同時我們檢視一下Spooling Directory中的datafile資訊—-檔案內容寫入到channel之後，該檔案被標記了：flume-hdfs.properties.COMPLETED
案例四：Spooling Directory Source：監聽一個指定的目錄，即只要應用程式向這個指定的目錄中新增新的檔案，source元件就可以獲取到該資訊，並解析該檔案的內容，然後寫入到channle。寫入完成後，標記該檔案已完成或者刪除該檔案。 其中 Sink：hdfs Channel：file (相比於案例三的兩個變化)
配置檔案：
# Name the components on this agent
a.sources = r1
a.sinks = k1
a.channels = c1
# Describe/configure the source
a.sources.r1.type = spooldir
a.sources.r1.spoolDir = /usr/local/datainput
a.sources.r1.fileHeader = true
a.sources.r1.interceptors = i1
a.sources.r1.interceptors.i1.type = timestamp
# Describe the sink
# Describe the sink
a.sinks.k1.type = hdfs
a.sinks.k1.hdfs.path = hdfs://master:9000/output
a.sinks.k1.hdfs.writeFormat = Text
a.sinks.k1.hdfs.fileType = DataStream
a.sinks.k1.hdfs.rollInterval = 10
a.sinks.k1.hdfs.rollSize = 0
a.sinks.k1.hdfs.rollCount = 0
a.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
a.sinks.k1.hdfs.useLocalTimeStamp = true
# Use a channel which buffers events in file
a.channels.c1.type = file
a.channels.c1.checkpointDir = /usr/flume/checkpoint
a.channels.c1.dataDirs = /usr/flume/data
# Bind the source and sink to the channel
a.sources.r1.channels = c1
a.sinks.k1.channel = c1 flume-ng agent -c conf -f flume-spooldir-test01.properties -name a -Dflume.root.logger=INFO,console
在控制檯上可以參看sink的執行進度日誌：
在HDFS的output資料夾中中檢視flume收集到的日誌資料：
案例五，接收json格式資料
配置檔案：
c.sources=r1 r2
c.channels=c1
c.sinks=s1
c.sources.r1.type=spooldir
c.sources.r1.spoolDir=flume
c.sources.r2.type = http
c.sources.r2.port = 8888
c.source.r2.bind = 192.168.13.100
c.sources.r2.channels = c1
c.channels.c1.type=memory
c.channels.c1.capacity=1000
c.channels.c1.transactionCapacity=100
c.sinks.s1.type=hdfs
c.sinks.s1.hdfs.path=/flume/%y-%m-%d
c.sinks.s1.hdfs.rollInterval=0
c.sinks.s1.hdfs.writeFormat=Text
c.sinks.s1.hdfs.fileType=DataStream
c.sinks.s1.hdfs.rollCount=0
c.sinks.s1.hdfs.rollSize=10485760
c.sinks.s1.hdfs.useLocalTimeStamp=true
c.sources.r1.channels=c1
c.sinks.s1.channel=c1啟動agent c
flume-ng agent -c conf -f flume-http-test01.properties -name c -Dflume.root.logger=INFO,console
postman發起json請求：
hdfs中的flume資料夾中檢視時間戳文夾，可以找到有些檔案寫入json請求中的body體