Flume數(shù)據(jù)采集之常見(jiàn)集群配置案例

[TOC]

非集群配置

這種情況非集群配置方式，比較簡(jiǎn)單，可以直接參考我整理的《Flume筆記整理》，其基本結(jié)構(gòu)圖如下：

Flume集群之多個(gè)Agent一個(gè)source

結(jié)構(gòu)說(shuō)明

結(jié)構(gòu)圖如下：

說(shuō)明如下：

即可以把我們的Agent部署在不同的節(jié)點(diǎn)上，上面是兩個(gè)Agent的情況。其中Agent foo可以部署在日志產(chǎn)生的節(jié)點(diǎn)上，
比如，可以是我們web服務(wù)器例如tomcat或者nginx的節(jié)點(diǎn)上，foo的source可以配置為監(jiān)控日志文件數(shù)據(jù)的變化，
channel則可以基于內(nèi)存或基于文件進(jìn)行存儲(chǔ)，而sink即日志落地可以配置為avro，即輸出到下一個(gè)Agent中。

Agent bar可以部署在另一個(gè)節(jié)點(diǎn)上，當(dāng)然跟foo在同一個(gè)節(jié)點(diǎn)也是沒(méi)有問(wèn)題，因?yàn)楸旧鞦lume是可以多個(gè)實(shí)例在同一個(gè)
節(jié)點(diǎn)上運(yùn)行的。bar主要作用是收集來(lái)自不同avro source的節(jié)點(diǎn)的日志數(shù)據(jù)，實(shí)際上，如果我們的web環(huán)境是集群的，
那么web服務(wù)器就會(huì)有多個(gè)節(jié)點(diǎn)，這時(shí)就有多個(gè)web服務(wù)器節(jié)點(diǎn)產(chǎn)生日志，我們需要在這多個(gè)web服務(wù)器上都部署agent，
此時(shí)，bar的source就會(huì)有多個(gè)，后面的案例正是如此，不過(guò)在這個(gè)小節(jié)中，只討論多個(gè)agent一個(gè)source的情況。
而對(duì)于agent bar的數(shù)據(jù)下沉方式，也是可以選擇多種方式，詳細(xì)可以參考官網(wǎng)文檔，這里選擇sink為HDFS。

不過(guò)需要注意的是，在agent foo中，source只有一個(gè)，在后面的案例中，會(huì)配置多個(gè)source，即在這一個(gè)agent中，
可以采集不同的日志文件，后面要討論的多個(gè)source，指的是多個(gè)不同日志文件的來(lái)源，即foo中的多個(gè)source，例如
data-access.log、data-ugctail.log、data-ugchead.log等等。

配置案例

環(huán)境說(shuō)明

如下：

即這里有兩個(gè)節(jié)點(diǎn)：

uplooking01：
其中的日志文件 /home/uplooking/data/data-clean/data-access.log
為web服務(wù)器生成的用戶(hù)訪問(wèn)日志，并且每天會(huì)產(chǎn)生一個(gè)新的日志文件。
在這個(gè)節(jié)點(diǎn)上，我們需要部署一個(gè)Flume的Agent，其source為該日志文件，sink為avro。

uplooking03：
這個(gè)節(jié)點(diǎn)的作用主要是收集來(lái)自不同F(xiàn)lume Agent的日志輸出數(shù)據(jù)，例如上面的agent，然后輸出到HDFS中。

說(shuō)明：在我的環(huán)境中，有uplooking01 uplooking02 uplooking03三個(gè)節(jié)點(diǎn)，并且三個(gè)節(jié)點(diǎn)配置了Hadoop集群。

配置

• uplooking01

#########################################################
##
##主要作用是監(jiān)聽(tīng)文件中的新增數(shù)據(jù)，采集到數(shù)據(jù)之后，輸出到avro
##    注意：Flume agent的運(yùn)行，主要就是配置source channel sink
##  下面的a1就是agent的代號(hào)，source叫r1 channel叫c1 sink叫k1
#########################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#對(duì)于source的配置描述 監(jiān)聽(tīng)文件中的新增數(shù)據(jù) exec
a1.sources.r1.type = exec
a1.sources.r1.command  = tail -F /home/uplooking/data/data-clean/data-access.log

#對(duì)于sink的配置描述 使用avro日志做數(shù)據(jù)的消費(fèi)
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = uplooking03
a1.sinks.k1.port = 44444

#對(duì)于channel的配置描述 使用文件做數(shù)據(jù)的臨時(shí)緩存 這種的安全性要高
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /home/uplooking/data/flume/checkpoint
a1.channels.c1.dataDirs = /home/uplooking/data/flume/data

#通過(guò)channel c1將source r1和sink k1關(guān)聯(lián)起來(lái)
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

• uplooking03

#########################################################
##
##主要作用是監(jiān)聽(tīng)avro，采集到數(shù)據(jù)之后，輸出到hdfs
##    注意：Flume agent的運(yùn)行，主要就是配置source channel sink
##  下面的a1就是agent的代號(hào)，source叫r1 channel叫c1 sink叫k1
#########################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#對(duì)于source的配置描述 監(jiān)聽(tīng)avro
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

#對(duì)于sink的配置描述 使用log日志做數(shù)據(jù)的消費(fèi)
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /input/data-clean/access/%y/%m/%d
a1.sinks.k1.hdfs.filePrefix = flume
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.inUsePrefix = tmpFlume
a1.sinks.k1.hdfs.inUseSuffix = .tmp
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = second
#配置下面兩項(xiàng)后，保存到HDFS中的數(shù)據(jù)才是文本
#否則通過(guò)hdfs dfs -text查看時(shí)，顯示的是經(jīng)過(guò)壓縮的16進(jìn)制
a1.sinks.k1.hdfs.serializer = TEXT
a1.sinks.k1.hdfs.fileType = DataStream

#對(duì)于channel的配置描述 使用內(nèi)存緩沖區(qū)域做數(shù)據(jù)的臨時(shí)緩存
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#通過(guò)channel c1將source r1和sink k1關(guān)聯(lián)起來(lái)
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

測(cè)試

首先要確保會(huì)有日志生成，其輸出為/home/uplooking/data/data-clean/data-access.log。

在uplooking03上啟動(dòng)Flume Agent：

[uplooking@uplooking03 flume]$ flume-ng agent -n a1 -c conf --conf-file conf/flume-source-avro.conf -Dflume.root.logger=INFO,console

在uplooking01上啟動(dòng)Flume Agent：

flume-ng agent -n a1 -c conf --conf-file conf/flume-sink-avro.conf -Dflume.root.logger=INFO,console

一段時(shí)間后，便可以在hdfs中看到寫(xiě)入的日志文件：

[uplooking@uplooking02 ~]$ hdfs dfs -ls /input/data-clean/access/18/04/07
18/04/07 08:52:02 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 26 items
-rw-r--r--   3 uplooking supergroup       1131 2018-04-07 08:50 /input/data-clean/access/18/04/07/flume.1523062248369.log
-rw-r--r--   3 uplooking supergroup       1183 2018-04-07 08:50 /input/data-clean/access/18/04/07/flume.1523062248370.log
-rw-r--r--   3 uplooking supergroup       1176 2018-04-07 08:50 /input/data-clean/access/18/04/07/flume.1523062248371.log
......

查看文件中的數(shù)據(jù)：

[uplooking@uplooking02 ~]$ hdfs dfs -text /input/data-clean/access/18/04/07/flume.1523062248369.log
18/04/07 08:55:07 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
1000    220.194.55.244  null    40604   0       POST /check/init HTTP/1.1       500     null    Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.3      1523062236368
1002    221.8.9.6 80    886a1533-38ca-466c-86e1-0b84022f781b    20201   1       GET /top HTTP/1.0       500     null      Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/59.0.3071.115 Safari/537.3      1523062236869
1002    61.172.249.96   99fb19c4-ec59-4abd-899c-4059dea39ead    0       0       POST /updateById?id=21 HTTP/1.1 408       null    Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7.0; rv:11.0) like Gecko    1523062237370
1003    61.172.249.96   886a1533-38ca-466c-86e1-0b84022f781b    10022   1       GET /tologin HTTP/1.1   null    /update/pass      Mozilla/5.0 (Windows; U; Windows NT 5.1)Gecko/20070309 Firefox/2.0.0.3  1523062237871
1003    125.39.129.67   6839fff8-7b3a-48f5-90cd-0f45c7be1aeb    10022   1       GET /tologin HTTP/1.0   408     null      Mozilla/5.0 (Windows; U; Windows NT 5.1)Gecko/20070309 Firefox/2.0.0.3  1523062238372
1000    61.172.249.96   89019ae0-6140-4e5a-9061-e3af74f3e4a8    10022   1       POST /stat HTTP/1.1     null    /passpword/getById?id=11  Mozilla/4.0 (compatible; MSIE 5.0; WindowsNT)   1523062238873

如果在uplooking03的Flume agent不配置hdfs.serializer=TEXT和hdfs.fileType=DataStream，那么上面查看到的數(shù)據(jù)會(huì)是16進(jìn)制數(shù)據(jù)。

Flume集群之多個(gè)Agent多個(gè)source

結(jié)構(gòu)說(shuō)明

如下：

配置案例

環(huán)境說(shuō)明

在我們的環(huán)境中，如下：

即在我們的環(huán)境中，日志源有三份，分別是data-access.log、data-ugchead.log、data-ugctail.log
不過(guò)在下面的實(shí)際配置中，日志源的agent我們只使用兩個(gè)，uplooking01和uplooking02，它們的sink都
輸出到uplooking03的source中。

配置

uplooking01和uplooking02的配置都是一樣的，如下：

#########################################################
##
##主要作用是監(jiān)聽(tīng)文件中的新增數(shù)據(jù)，采集到數(shù)據(jù)之后，打印在控制臺(tái)
##    注意：Flume agent的運(yùn)行，主要就是配置source channel sink
##  下面的a1就是agent的代號(hào)，source叫r1 channel叫c1 sink叫k1
#########################################################
a1.sources = r1 r2 r3
a1.sinks = k1
a1.channels = c1

#對(duì)于source r1的配置描述 監(jiān)聽(tīng)文件中的新增數(shù)據(jù) exec
a1.sources.r1.type = exec
a1.sources.r1.command  = tail -F /home/uplooking/data/data-clean/data-access.log
a1.sources.r1.interceptors = i1 i2
a1.sources.r1.interceptors.i1.type = static
##靜態(tài)的在header中添加一個(gè)key value，下面就配置了兩個(gè)攔截器，i1和i2
a1.sources.r1.interceptors.i1.key = type
a1.sources.r1.interceptors.i1.value = access
a1.sources.r1.interceptors.i2.type = timestamp
## timestamp的作用：這里配置了的話(huà)，在負(fù)責(zé)集中收集日志的flume agent就不需要配置
## a1.sinks.k1.hdfs.useLocalTimeStamp = true也能通過(guò)這些%y/%m/%d獲取時(shí)間信息
## 這樣一來(lái)的話(huà)，就可以減輕集中收集日志的flume agent的負(fù)擔(dān)，因?yàn)榇藭r(shí)的時(shí)間信息可以直接從source中獲取

#對(duì)于source r2的配置描述 監(jiān)聽(tīng)文件中的新增數(shù)據(jù) exec
a1.sources.r2.type = exec
a1.sources.r2.command  = tail -F /home/uplooking/data/data-clean/data-ugchead.log
a1.sources.r2.interceptors = i1 i2
a1.sources.r2.interceptors.i1.type = static
##靜態(tài)的在header中添加一個(gè)key value，下面就配置了兩個(gè)攔截器，i1和i2
a1.sources.r2.interceptors.i1.key = type
a1.sources.r2.interceptors.i1.value = ugchead
a1.sources.r2.interceptors.i2.type = timestamp

#對(duì)于source r3的配置描述 監(jiān)聽(tīng)文件中的新增數(shù)據(jù) exec
a1.sources.r3.type = exec
a1.sources.r3.command  = tail -F /home/uplooking/data/data-clean/data-ugctail.log
a1.sources.r3.interceptors = i1 i2
a1.sources.r3.interceptors.i1.type = static
##靜態(tài)的在header中添加一個(gè)key value，下面就配置了兩個(gè)攔截器，i1和i2
a1.sources.r3.interceptors.i1.key = type
a1.sources.r3.interceptors.i1.value = ugctail
a1.sources.r3.interceptors.i2.type = timestamp

#對(duì)于sink的配置描述 使用avro日志做數(shù)據(jù)的消費(fèi)
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = uplooking03
a1.sinks.k1.port = 44444

#對(duì)于channel的配置描述 使用文件做數(shù)據(jù)的臨時(shí)緩存 這種的安全性要高
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /home/uplooking/data/flume/checkpoint
a1.channels.c1.dataDirs = /home/uplooking/data/flume/data

#通過(guò)channel c1將source r1 r2 r3和sink k1關(guān)聯(lián)起來(lái)
a1.sources.r1.channels = c1
a1.sources.r2.channels = c1
a1.sources.r3.channels = c1
a1.sinks.k1.channel = c1

uplooking03的配置如下：

#########################################################
##
##主要作用是監(jiān)聽(tīng)avro，采集到數(shù)據(jù)之后，輸出到hdfs
##    注意：Flume agent的運(yùn)行，主要就是配置source channel sink
##  下面的a1就是agent的代號(hào)，source叫r1 channel叫c1 sink叫k1
#########################################################
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#對(duì)于source的配置描述 監(jiān)聽(tīng)avro
a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

#對(duì)于sink的配置描述 使用log日志做數(shù)據(jù)的消費(fèi)
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /input/data-clean/%{type}/%Y/%m/%d
a1.sinks.k1.hdfs.filePrefix = %{type}
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.inUseSuffix = .tmp
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollSize = 10485760
# 如果希望上面配置的日志文件滾動(dòng)策略生效，則必須要配置下面這一項(xiàng)
a1.sinks.k1.hdfs.minBlockReplicas = 1
#配置下面兩項(xiàng)后，保存到HDFS中的數(shù)據(jù)才是文本
#否則通過(guò)hdfs dfs -text查看時(shí)，顯示的是經(jīng)過(guò)壓縮的16進(jìn)制
a1.sinks.k1.hdfs.serializer = TEXT
a1.sinks.k1.hdfs.fileType = DataStream

#對(duì)于channel的配置描述 使用內(nèi)存緩沖區(qū)域做數(shù)據(jù)的臨時(shí)緩存
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#通過(guò)channel c1將source r1和sink k1關(guān)聯(lián)起來(lái)
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

測(cè)試

首先需要保證uplooking01和uplooking02上都能正常地產(chǎn)生日志。

在uplooking03上啟動(dòng)Agent：

[uplooking@uplooking03 flume]$ flume-ng agent -n a1 -c conf --conf-file conf/flume-source-avro.conf -Dflume.root.logger=INFO,console

分別在uplooking01和uplooking02上啟動(dòng)Agent：

flume-ng agent -n a1 -c conf --conf-file conf/flume-sink-avro.conf -Dflume.root.logger=INFO,console

一段時(shí)間后，可以在HDFS中查看相應(yīng)的日志文件：

$ hdfs dfs -ls /input/data-clean
18/04/08 01:34:36 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 3 items
drwxr-xr-x   - uplooking supergroup          0 2018-04-07 22:00 /input/data-clean/access
drwxr-xr-x   - uplooking supergroup          0 2018-04-07 22:00 /input/data-clean/ugchead
drwxr-xr-x   - uplooking supergroup          0 2018-04-07 22:00 /input/data-clean/ugctail

查看某個(gè)日志目錄下的日志文件：

[uplooking@uplooking02 data-clean]$ hdfs dfs -ls /input/data-clean/access/2018/04/07
18/04/08 01:35:27 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 2 items
-rw-r--r--   3 uplooking supergroup    2447752 2018-04-08 01:02 /input/data-clean/access/2018/04/08/access.1523116801502.log
-rw-r--r--   3 uplooking supergroup       5804 2018-04-08 01:02 /input/data-clean/access/2018/04/08/access.1523120538070.log.tmp

可以看到日志文件數(shù)量非常少，那是因?yàn)榍懊嬖谂渲?code>uplooking03的agent時(shí)，日志文件滾動(dòng)的方式為，單個(gè)文件滿(mǎn)10M再進(jìn)行切分日志文件。