本文是對大數(shù)據(jù)組件Flume的一個學(xué)習(xí)總結(jié)，共包括如下章節(jié)的內(nèi)容：

• 簡介
• 核心概念
• 使用場景
• 快速起步
• 小結(jié)

一、簡介

Flume是一種分布式，可靠且可用的服務(wù)，用于有效地收集，聚合和移動大量日志數(shù)據(jù)。它具有基于流數(shù)據(jù)的簡單靈活的架構(gòu)、可靠的可靠性機制和許多故障轉(zhuǎn)移和恢復(fù)機制，以及強大的容錯性。它使用簡單的可擴展數(shù)據(jù)模型，允許在線分析數(shù)據(jù)。很多大數(shù)據(jù)分析系統(tǒng)都通過flume來獲取數(shù)據(jù)的輸入。

Flume最早是Cloudera提供的日志收集系統(tǒng)，后來成為Apache下的項目，目前已經(jīng)是Apache下的頂級項目。

Flume有兩個版本,一個是老的Flume 0.9X版本，統(tǒng)稱為Flume-og；另一個是經(jīng)過重大重構(gòu)后的版本Flume1.X，統(tǒng)稱為Flume-ng。在本文中，我們使用的是最新的Flume-ng版本，版本號是Apache Flume 1.9.0。

Flume的官方網(wǎng)站是http://flume.apache.org。

Flume是用java開發(fā)的，其運行依賴java環(huán)境。Flume可以在windows和linux下運行，本文示例所用的是linux環(huán)境。

二、核心概念

Event：消息的基本單位，由header和body組成。

Agent：一個獨立JVM進程，負(fù)責(zé)將一端外部來源產(chǎn)生的消息轉(zhuǎn) 發(fā)到另一端外部的目的地。在分布式部署中，每個節(jié)點部署一個Agent。多個Agent就組成了Flume的分布式系統(tǒng)。

（一）Agent（代理）

Flume 運行的核心是 Agent。Flume以agent為最小的獨立運行單位。一個agent就是一個JVM。它是一個完整的數(shù)據(jù)收集工具，含有三個核心組件，分別是source、 channel、 sink。通過這些組件， Event 可以從一個地方流向另一個地方，如下圖示例：

上圖是Agent的一個最基本的結(jié)構(gòu)，一個Agent中的source、channel、sink數(shù)量可以有多個，它們可相互組合使用，比如一個source可以對應(yīng)多個channel，一個channel也可對應(yīng)多個sink。它們的關(guān)系在配置文件中配置。后面會舉例介紹。

下面接著介紹Agent的這三個核心組件的含義.

（二）Source（源）

Source是數(shù)據(jù)的收集端，負(fù)責(zé)將數(shù)據(jù)捕獲后進行特殊的格式化，將數(shù)據(jù)封裝到事件（event） 里，然后將事件推入Channel中。 Flume提供了很多內(nèi)置的Source， 支持 Avro， log4j， syslog 和 http post(body為json格式)，可以讓應(yīng)用程序同已有的Source直接打交道，如AvroSource，　　SyslogTcpSource。 如果內(nèi)置的Source無法滿足需要， Flume還支持自定義Source。

（三）Channel（通道）

Channel是連接Source和Sink的組件，大家可以將它看做一個數(shù)據(jù)的緩沖區(qū)（數(shù)據(jù)隊列），它可以將事件暫存到內(nèi)存中也可以持久化到本地磁盤上， 直　到Sink處理完該事件。

Flume提供了多種內(nèi)置的Channel類型，如MemoryChannel和FileChannel，也支持用戶自定義Channel。

（四）Sink（接收器）

Sink從Channel中取出事件，然后將數(shù)據(jù)發(fā)到別處，可以向文件系統(tǒng)、數(shù)據(jù)庫、 hadoop存數(shù)據(jù)， 也可以是其他flume agent的Source。Flume也提供了多種內(nèi)置的Sink。

三、使用場景

Flume在英文中的意思是水道， 但Flume更像可以隨意組裝的消防水管，下面根據(jù)官方文檔，展示幾種Flow。

（一）多個agent順序連接

可以將多個Agent順序連接起來，將最初的數(shù)據(jù)源經(jīng)過收集，存儲到最終的存儲系統(tǒng)中。這是最簡單的情況，一般情況下，應(yīng)該控制這種順序連接的Agent 的數(shù)量，因為數(shù)據(jù)流經(jīng)的路徑變長了，出現(xiàn)故障將影響整個Flow上的Agent收集服務(wù)。

（二）多個Agent的數(shù)據(jù)匯聚到同一個Agent

這種情況應(yīng)用的場景比較多，比如要收集Web網(wǎng)站的用戶行為日志， Web網(wǎng)站為了可用性使用的負(fù)載集群模式，每個節(jié)點都產(chǎn)生用戶行為日志，可以為每個節(jié)點都配置一個Agent來單獨收集日志數(shù)據(jù)，然后多個Agent將數(shù)據(jù)最終匯聚到一個用來存儲數(shù)據(jù)存儲系統(tǒng)，如HDFS上。

（三）多級流

Flume還支持多級流，什么多級流？結(jié)合在云開發(fā)中的應(yīng)用來舉個例子，當(dāng)syslog， java， nginx、 tomcat等混合在一起的日志流開始流入一個agent后，可以agent中將混雜的日志流分開，然后給每種日志建立一個自己的傳輸通道。

（四）load balance功能

上圖Agent1是一個路由節(jié)點，負(fù)責(zé)將Channel暫存的Event均衡到對應(yīng)的多個Sink組件上，而每個Sink組件分別連接到一個獨立的Agent上 ，這樣起到負(fù)荷分擔(dān)的效果。

四、快速起步

下面我們通過一個簡單的實例來演示如何使用Flume，從而加深對Flume基本概念的理解。

（一）版本安裝

從flume官網(wǎng)下載二進制的安裝文件apache-flume-1.9.0-bin.tar.gz，解壓后得到的目錄結(jié)構(gòu)如下：

flume的可執(zhí)行程序（實際是個shell腳本）位于bin目錄下，文件名是flume-ng。我們可以運行flume-ng version測試下當(dāng)前環(huán)境是否正常，如：

$ bin/flume-ng version

Flume 1.9.0
Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
Revision: d4fcab4f501d41597bc616921329a4339f73585e
Compiled by fszabo on Mon Dec 17 20:45:25 CET 2018
From source with checksum 35db629a3bda49d23e9b3690c80737f9

如果能正常執(zhí)行，顯示上面的類似信息，說明當(dāng)前環(huán)境是正常的（比如java環(huán)境）。上面測試并不需要修改flume的任何配置文件。

（二）配置文件

啟動一個flume agent需要指定一個配置文件，該配置文件用于配置該agent的相關(guān)參數(shù)信息，如source,channel,sink等設(shè)置信息。配置文件是普通的鍵值對格式的文本文件。在安裝目錄下的conf目錄下有樣例文件可參考。

下面我們通過一個簡單的例子來說明如何設(shè)置agent的配置文件。

每個flume agent都需要指定一個名稱，配置文件的前三個參數(shù)都是以如下內(nèi)容開始，如：

a1.sources=<list of sources>
a1.channels=<list of channel>
a1.sinks=<list of sinks>

上面的配置信息中，a1就是要啟動的flume agent的名稱，a1.sources的值指定該agent的source組件，一個agent可以配置多個source，所以a1.sources的值是個列表；同樣a1.channels用于指定該agent的channel信息；a1.sinks用于指定該agent的sink信息。舉例：

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

上面信息定義了一個名為a1的flume agent，它有一個名為s1的source（源）、一個名為c1的channel（通道），以及一個名為k1的sink（接收器）。

配置了agent的三個核心組件source,channel,sink后（實際只是指定了一個名稱），下面就要接著配置這三個組件的具體信息，如類型等。

下面我們先配置source信息，例子如下：

a1.sources.s1.type=netcat
a1.sources.s1.bind=0.0.0.0
a1.sources.s1.port=12345
a1.sources.s1.channels=c1

上面的信息設(shè)置源s1的類型（type參數(shù)）為netcat，這是flume內(nèi)置的一種源類型，它的作用是監(jiān)聽來自socket中的數(shù)據(jù)，需要指定監(jiān)聽的ip地址和端口號，bind參數(shù)設(shè)置Ip地址，0.0.0.0表示監(jiān)聽任何地址，port參數(shù)設(shè)置端口號，這里設(shè)置的端口號是12345。對每個源，還需要配置其對應(yīng)的channel，一個源可以對應(yīng)多個channel，所以屬性名是復(fù)數(shù)channels，我們這里只對應(yīng)一個channel，即上面配置的通道c1。

我們再配置channel的信息，例子如下：

a1.channels.c1.type = memory

上面配置c1通道為內(nèi)存通道，通道數(shù)據(jù)既可以保存在內(nèi)存中，也可以在文件中，其它采用默認(rèn)值。

最后配置sink信息，例子如下：

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

上面的配置中type指定sink的類型，這里的logger是flume內(nèi)置的一種sink類型，主要用于調(diào)試，它會使用log4j將所有info級別的日志記錄下來，這些日志來自于配置好的通道。channel 參數(shù)用于指定sink從哪個channel獲取數(shù)據(jù)。一個sink只能從一個channel獲取數(shù)據(jù)。

（三）啟動Agent

配置好要啟動的flume agent所需要的配置文件后，就可以啟動agent了，即啟動一個java進程。

假設(shè)我們上面的配置文件名稱為hello.conf，保存到flume解壓目錄的conf目錄下，該配置文件中完整的內(nèi)容如下：

a1.sources=s1
a1.channels=c1
a1.sinks=k1
a1.sources.s1.type=netcat
a1.sources.s1.bind=0.0.0.0
a1.sources.s1.port=12345
a1.sources.s1.channels=c1
a1.channels.c1.type = memory
a1.sinks.k1.type = logger
a1.sinks.k1.channel= c1

下面我們執(zhí)行flume-ng腳本來啟動agent。命令格式如下：

$bin/flume-ng agent -n a1 -c conf -f conf/hello.conf -Dflume.root.logger=INFO,console

上面的命令，各選項的含義如下：

1）flume-ng是可執(zhí)行腳本，位于bin目錄下

2）agent參數(shù)表示這是啟動一個agent。

3）-n 參數(shù)用于指定本agent的名稱，這里是a1，這樣flume就會根據(jù)這個名稱到配置文件中獲取相關(guān)的配置信息。

4）-c 參數(shù)用于指定log4j的配置文件所在的目錄，這里是conf目錄

5）-f參數(shù)用于指定flume agent配置文件的名稱

6）-Dflume.root.logger參數(shù)的作用是覆蓋conf/log4j.properties中的信息，讓采集的信息在控制臺上輸出。

執(zhí)行上面命令后，會打印很多信息，該flume agent會等待socket客戶端的連接和數(shù)據(jù)輸入，這時我們利用操作系統(tǒng)的nc命令來往該agent發(fā)送信息。我們打開另外一個控制臺界面，輸入如下信息：

$ nc localhost 12345

hello,world
OK

good job
OK

上面我們運行nc命令連接flume,然后輸入兩行信息。

這時我們切換到運行flume agent的控制臺界面，發(fā)現(xiàn)控制臺上輸出了如下信息，這正是nc程序發(fā)出的信息，如下：

2019-03-27 11:26:30,281 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 68 65 6C 6C 6F 2C 77 6F 72 6C 64                hello,world }

2019-03-27 11:26:37,165 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 67 6F 6F 64 20 6A 6F 62                         good job }

上面只是一個簡單的例子，從這個例子中可以看出，使用flume并不是很復(fù)雜，在規(guī)劃好啟動多少agent之后。后面的主要工作是設(shè)置每個agent的配置信息，最基本的就是source,channel和sink信息。

五、小結(jié)

本文是對flume組件的一個簡單入門學(xué)習(xí)，主要是對flume的核心概念進行了介紹，并通過一個簡單的例子演示了如何使用flume。