摘要：對于一般的采集需求，通過對的簡單配置即可實現。針對特殊場景也具備良好的自定義擴展能力，因此，可以適用于大部分的日常數據采集場景。

文章作者：foochane?

原文鏈接：https://foochane.cn/article/2019062701.html

Flume日志采集框架 安裝和部署 Flume運行機制 采集靜態文件到hdfs 采集動態日志文件到hdfs 兩個agent級聯

在一個完整的離線大數據處理系統中，除了hdfs+mapreduce+hive組成分析系統的核心之外，還需要數據采集、結果數據導出、任務調度等不可或缺的輔助系統，而這些輔助工具在hadoop生態體系中都有便捷的開源框架，如圖所示：

Flume是一個分布式、可靠、和高可用的海量日志采集、聚合和傳輸的系統。Flume可以采集文件，socket數據包、文件、文件夾、kafka等各種形式源數據，又可以將采集到的數據(下沉sink)輸出到HDFS、hbase、hive、kafka等眾多外部存儲系統中。

對于一般的采集需求，通過對flume的簡單配置即可實現。

Flume針對特殊場景也具備良好的自定義擴展能力，因此，flume可以適用于大部分的日常數據采集場景。

Flume分布式系統中最核心的角色是agent，flume采集系統就是由一個個agent所連接起來形成,每一個agent相當于一個數據傳遞員，內部有三個組件：

Source：采集組件，用于跟數據源對接，以獲取數據

Sink：下沉組件，用于往下一級agent傳遞數據或者往最終存儲系統傳遞數據

Channel：傳輸通道組件，用于從source將數據傳遞到sink

單個agent采集數據:

多級agent之間串聯:

1 下載安裝包apache-flume-1.9.0-bin.tar.gz解壓

2 在conf文件夾下的flume-env.sh添加JAVA_HOME

export JAVA_HOME=/usr/local/bigdata/java/jdk1.8.0_211

3 根據采集的需求，添加采集方案配置文件，文件名可以任意取

具體可以看后面的示例

4 啟動flume

測試環境下：

$ bin/flume/-ng agent -c conf/ -f ./dir-hdfs.conf -n agent1 -Dflume.root.logger=INFO,console

命令說明：

-c：指定flume自帶的配置文件目錄，不用自己修改

-f：指定自己的配置文件,這里問當前文件夾下的dir-hdfs.conf

-n：指定自己配置文件中使用那個agent,對應的配置文件中定義的名字。

-Dflume.root.logger：把日志打印在控制臺，類型為INFO，這個只用于測試，后面將打印到日志文件中

生產中，啟動flume，應該把flume啟動在后臺：

nohup bin/flume-ng  agent  -c  ./conf  -f ./dir-hdfs.conf -n  agent1 1>/dev/null 2>&1 &

某服務器的某特定目錄下，會不斷產生新的文件，每當有新文件出現，就需要把文件采集到HDFS中去

在安裝目錄下添加文件dir-hdfs.conf,然后添加配置信息。

先獲取agent，命名為agent1，后面的配置都跟在agent1后面，也可以改為其他值，如agt1,同一個配置文件中可以有多個配置配置方案，啟動agent的時候獲取對應的名字就可以。

根據需求，首先定義以下3大要素

數據源組件

即source ——監控文件目錄 : spooldir
spooldir有如下特性：

監視一個目錄，只要目錄中出現新文件，就會采集文件中的內容

采集完成的文件，會被agent自動添加一個后綴：COMPLETED（可修改）

所監視的目錄中不允許重復出現相同文件名的文件

下沉組件

即sink——HDFS文件系統 : hdfs sink

通道組件

即channel——可用file channel 也可以用內存channel

#定義三大組件的名稱
agent1.sources = source1
agent1.sinks = sink1
agent1.channels = channel1

# 配置source組件
agent1.sources.source1.type = spooldir
agent1.sources.source1.spoolDir = /root/log/
agent1.sources.source1.fileSuffix=.FINISHED
#文件每行的長度，注意這里如果事情文件每行超過這個長度會自動切斷，會導致數據丟失
agent1.sources.source1.deserializer.maxLineLength=5120

# 配置sink組件
agent1.sinks.sink1.type = hdfs
agent1.sinks.sink1.hdfs.path =hdfs://Master:9000/access_log/%y-%m-%d/%H-%M
agent1.sinks.sink1.hdfs.filePrefix = app_log
agent1.sinks.sink1.hdfs.fileSuffix = .log
agent1.sinks.sink1.hdfs.batchSize= 100
agent1.sinks.sink1.hdfs.fileType = DataStream
agent1.sinks.sink1.hdfs.writeFormat =Text

# roll：滾動切換：控制寫文件的切換規則
## 按文件體積（字節）來切 
agent1.sinks.sink1.hdfs.rollSize = 512000
## 按event條數切
agent1.sinks.sink1.hdfs.rollCount = 1000000
## 按時間間隔切換文件
agent1.sinks.sink1.hdfs.rollInterval = 60

## 控制生成目錄的規則
agent1.sinks.sink1.hdfs.round = true
agent1.sinks.sink1.hdfs.roundValue = 10
agent1.sinks.sink1.hdfs.roundUnit = minute

agent1.sinks.sink1.hdfs.useLocalTimeStamp = true

# channel組件配置
agent1.channels.channel1.type = memory
## event條數
agent1.channels.channel1.capacity = 500000
##flume事務控制所需要的緩存容量600條event
agent1.channels.channel1.transactionCapacity = 600

# 綁定source、channel和sink之間的連接
agent1.sources.source1.channels = channel1
agent1.sinks.sink1.channel = channel1

Channel參數解釋：

capacity：默認該通道中最大的可以存儲的event數量

trasactionCapacity：每次最大可以從source中拿到或者送到sink中的event數量

keep-alive：event添加到通道中或者移出的允許時間

$ bin/flume/-ng agent -c conf/ -f dir-hdfs.conf -n agent1 -Dflume.root.logger=INFO,console

比如業務系統使用log4j生成的日志，日志內容不斷增加，需要把追加到日志文件中的數據實時采集到hdfs

配置文件名稱：tail-hdfs.conf
根據需求，首先定義以下3大要素：

采集源，即source——監控文件內容更新 : exec tail -F file

下沉目標，即sink——HDFS文件系統 : hdfs sink

Source和sink之間的傳遞通道——channel，可用file channel 也可以用 內存channel

配置文件內容：

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /root/app_weichat_login.log

# Describe the sink
agent1.sinks.sink1.type = hdfs
agent1.sinks.sink1.hdfs.path =hdfs://Master:9000/app_weichat_login_log/%y-%m-%d/%H-%M
agent1.sinks.sink1.hdfs.filePrefix = weichat_log
agent1.sinks.sink1.hdfs.fileSuffix = .dat
agent1.sinks.sink1.hdfs.batchSize= 100
agent1.sinks.sink1.hdfs.fileType = DataStream
agent1.sinks.sink1.hdfs.writeFormat =Text

agent1.sinks.sink1.hdfs.rollSize = 100
agent1.sinks.sink1.hdfs.rollCount = 1000000
agent1.sinks.sink1.hdfs.rollInterval = 60

agent1.sinks.sink1.hdfs.round = true
agent1.sinks.sink1.hdfs.roundValue = 1
agent1.sinks.sink1.hdfs.roundUnit = minute


agent1.sinks.sink1.hdfs.useLocalTimeStamp = true



# 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

啟動命令：

bin/flume-ng agent -c conf -f conf/tail-hdfs.conf -n a1

從tail命令獲取數據發送到avro端口
另一個節點可配置一個avro源來中繼數據，發送外部存儲

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /root/log/access.log


# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hdp-05
a1.sinks.k1.port = 4141
a1.sinks.k1.batch-size = 2



# 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

從avro端口接收數據，下沉到hdfs

采集配置文件，avro-hdfs.conf

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
##source中的avro組件是一個接收者服務
a1.sources.r1.type = avro
a1.sources.r1.bind = hdp-05
a1.sources.r1.port = 4141


# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/taildata/%y-%m-%d/
a1.sinks.k1.hdfs.filePrefix = tail-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 24
a1.sinks.k1.hdfs.roundUnit = hour
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.rollSize = 0
a1.sinks.k1.hdfs.rollCount = 50
a1.sinks.k1.hdfs.batchSize = 10
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的文件類型，默認是Sequencefile，可用DataStream，則為普通文本
a1.sinks.k1.hdfs.fileType = DataStream

# 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