Flume全解析！全！全！全！

俺老孫又回來了

自從接受了人民教師這個設定

莫名感覺萌萌的呢~~

所以俺孫老師今天想跟你們講一個牛逼哄哄的系統喲

Flume是一個分散式、可靠、和高可用的海量日誌聚合的系統，支持在系統中定製各類數據發送方，用於收集數據；同時，Flume提供對數據進行簡單處理，並寫到各種數據接受方（可定製）的能力。

那麼他是為了什麼設計的呢？？？

可靠性

當節點出現故障時，日誌能夠被傳送到其他節點上而不會丟失。Flume提供了三種級別的可靠性保障，從強到弱依次分別為：end-to-end（收到數據agent首先將event寫到磁碟上，當數據傳送成功後，再刪除；如果數據發送失敗，可以重新發送。），Store on failure（這也是scribe採用的策略，當數據接收方crash時，將數據寫到本地，待恢復後，繼續發送），Best effort（數據發送到接收方後，不會進行確認）。

可擴展性

Flume採用了三層架構，分別為agent，collector和storage，每一層均可以水平擴展。其中，所有agent和collector由master統一管理，這使得系統容易監控和維護，且master允許有多個（使用ZooKeeper進行管理和負載均衡），這就避免了單點故障問題。

可管理性

所有agent和colletor由master統一管理，這使得系統便於維護。多master情況，Flume利用ZooKeeper和gossip，保證動態配置數據的一致性。用戶可以在master上查看各個數據源或者數據流執行情況，且可以對各個數據源配置和動態載入。Flume提供了web和shell script command兩種形式對數據流進行管理。

功能可擴展性

用戶可以根據需要添加自己的agent，collector或者storage。此外，Flume自帶了很多組件，包括各種agent（file， syslog等），collector和storage（file，HDFS等）。

Flume的邏輯架構

正如前面提到的，Flume採用了分層架構：分別為agent，collector和storage。其中，agent和collector均由兩部分組成：source和sink，source是數據來源，sink是數據去向。

Flume使用兩個組件：Master和Node，Node根據在Master

(1) agent

agent的作用是將數據源的數據發送給collector。

Flume自帶了很多直接可用的數據源（source），如：

• text(「filename」)：將文件filename作為數據源，按行發送

• tail(「filename」)：探測filename新產生的數據，按行發送出去

• fsyslogTcp(5140)：監聽TCP的5140埠，並且接收到的數據發送出去

• tailDir("dirname"[, fileregex=".*"[, startFromEnd=false[, recurseDepth=0]]])：監聽目錄中的文件末尾，使用正則去選定需要監聽的文件（不包含目錄），recurseDepth為遞歸監聽其下子目錄的深度

同時提供了很多sink，如：

• console[("format")] ：直接將將數據顯示在consolr上

• text(「txtfile」)：將數據寫到文件txtfile中

• dfs(「dfsfile」)：將數據寫到HDFS上的dfsfile文件中

• syslogTcp(「host」,port)：將數據通過TCP傳遞給host節點

• agentSink[("machine"[,port])]：等價於agentE2ESink，如果省略，machine參數，默認使用flume.collector.event.host與flume.collector.event.port作為默認collecotr

• agentDFOSink[("machine" [,port])]：本地熱備agent，agent發現collector節點故障後，不斷檢查collector的存活狀態以便重新發送event，在此間產生的數據將緩存到本地磁碟中

• agentBESink[("machine"[,port])]：不負責的agent，如果collector故障，將不做任何處理，它發送的數據也將被直接丟棄

• agentE2EChain：指定多個collector提高可用性。 當向主collector發送event失效後，轉向第二個collector發送，當所有的collector失敗後，它會非常執著的再來一遍

(2) collector

collector的作用是將多個agent的數據匯總後，載入到storage中。

它的source和sink與agent類似。

數據源（source），如：

• collectorSource[(port)]：Collector source，監聽埠匯聚數據

• autoCollectorSource：通過master協調物理節點自動匯聚數據

• logicalSource：邏輯source，由master分配埠並監聽rpcSink

sink，如：

• collectorSink( "fsdir","fsfileprefix",rollmillis)：collectorSink，數據通過collector匯聚之後發送到hdfs, fsdir 是hdfs目錄，fsfileprefix為文件前綴碼

• customdfs("hdfspath"[, "format"])：自定義格式dfs

(3) storage

storage是存儲系統，可以是一個普通file，也可以是HDFS，HIVE，HBase，分散式存儲等。

(4) Master

Master是管理協調agent和collector的配置等信息，是flume集群的控制器。

在Flume中，最重要的抽象是data flow（數據流），data flow描述了數據從產生，傳輸、處理並最終寫入目標的一條路徑。

● 對於agent數據流配置就是從哪得到數據，把數據發送到哪個collector。

● 對於collector是接收agent發過來的數據，把數據發送到指定的目標機器上。

註：Flume框架對hadoop和zookeeper的依賴只是在jar包上，並不要求flume啟動時必須將hadoop和zookeeper服務也啟動。

Flume分散式環境部署

實驗場景

• 操作系統版本：RedHat 5.6

• Hadoop版本：0.20.2

• Jdk版本：jdk1.6.0_26

• 安裝flume版本：flume-distribution-0.9.4-bin

部署flume在集群上，按照如下步驟：

1. 在集群上的每台機器上安裝flume

2. 選擇一個或多個節點當做master

3. 修改靜態配置文件

4. 在至少一台機器上啟動一個master ，所有節點啟動flume node

5. 動態配置

需要在集群的每台機器上部署Flume。

注意：flume集群整個集群的網路環境要保證穩定，可靠，否則會出現一些莫名錯誤（比如：agent端發送不了數據到collector）。

1.Flume環境安裝

$wget http://cloud.github.com/downloads/cloudera/flume/flume-distribution-0.9.4-bin.tar.gz $tar -xzvf flume-distribution-0.9.4-bin.tar.gz $cp -rf flume-distribution-0.9.4-bin /usr/local/flume $vi /etc/profile #添加環境配置 export FLUME_HOME=/usr/local/flume export PATH=.:$PATH::$FLUME_HOME/bin $source /etc/profile $flume #驗證安裝

2.選擇一個或多個節點當做master

對於master的選擇情況，可以在集群上定義一個master，也可以為了提高可用性選擇多個節點做為master。

• 單點master模式：容易管理，但在系統的容錯和擴展性有缺陷

• 多點master模式：通常是運行3/5個master，能很好的容錯

Flume master數量的選擇原則

分散式的master能夠繼續正常工作不會崩潰的前提是正常工作的master數量超過總master數量的一半。

Flume master 的作用主要有兩個：

• 跟蹤各節點的配置情況，通知節點配置的改變；

• 跟蹤來自flow的結尾操控在可靠模式下（E2E）的信息，以至於讓flow的源頭知道什麼時候停止傳輸event。

3.修改靜態配置文件

site-specific設置對於flume節點和master通過在每一個集群節點的conf/flume-site.xml是可配置的，如果這個文件不存在，設置的屬性默認的在conf/flume­-conf.xml中，在接下來的例子中，在flume的節點上設置master名，讓節點自己去尋找叫「master」的flume

<?xml version="1.0"?> <?xml-stylesheet type="text/xsl" href="configuration.xsl"?> <configuration> <property> <name>flume.master.servers</name> <value>master</value> </property> </configuration>

在多master的情況下需要如下配置：

<property> <name>flume.master.servers</name> <value>hadoopmaster.com,hadoopedge.com,datanode4.com</value> <description>A comma-separated list of hostnames, one for each machine in the Flume Master.</description> </property> <property> <name>flume.master.store</name> <value>zookeeper</value> <description>How the Flume Master stores node configurations. Must be either zookeeper or memory.</description> </property> <property> <name>flume.master.serverid</name> <value>2</value> <description>The unique identifier for a machine in a Flume Master ensemble. Must be different on every master instance.</description> </property>

注意：flume.master.serverid 屬性的配置主要是針對master，集群上Master節點的flume.master.serverid 必須是不能相同的，該屬性的值以0開始。

當使用agent角色時，你可以通過添加下面的配置文件在flume-conf.xml中，來設置默認的collector主機：

<property> <name>flume.collector.event.host</name> <value>collector</value> <description>This is the host name of the default "remote" collector.</description> </property> <property> <name>flume.collector.port</name> <value>35853</value> <description>This default tcp port that the collector listens to in order to receive events it is collecting.</description> </property>

4.啟動集群

集群上節點啟動：

1. 在命令行輸入：flume master 啟動master節點

2. 在命令行輸入：flume node –n nodeName 啟動其他節點，nodeName最好根據集群邏輯的劃分來取名子，這樣在 master進行配置的時候比較清晰。

名字規則自己定義，方便記憶和動態配置即可（後續會有介紹動態配置）

5.基於flume shell的動態配置

假設我們目前部署的Flume集群結構如下：

我們想將A-F所在的機器的系統日誌收集到HDFS中，怎麼樣在flume

1）

$flume shell >connect localhost >help >exec map 192.168.0.1 agentA >exec map 192.168.0.2 agentB >exec map 192.168.0.3 agentC >exec map 192.168.0.4 agentD >exec map 192.168.0.5 agentE >exec map 192.168.0.6 agentF >getnodestatus 192.168.0.1 --> IDLE 192.168.0.2 --> IDLE 192.168.0.3 --> IDLE 192.168.0.4 --> IDLE 192.168.0.5 --> IDLE 192.168.0.6 --> IDLE agentA --> IDLE agentB --> IDLE agentC --> IDLE agentD --> IDLE agentE --> IDLE agentF --> IDLE >exec map 192.168.0.11 collector

這裡你也可以打開web master界面查看。

2）

啟動Collector的監聽埠

>exec config collector collectorSource(35853) collectorSink("","")#collector節點監聽35853埠過來的數據，這一部非常重要

登陸到collector伺服器進行埠檢測

$netstat -nalp|grep 35853

如果在master中未進行上述配置，在collector上檢測不到此打開埠

3）

設置各節點的source和sink

>exec config collector collectorSource(35853) collectorSink("hdfs://namenode/flume/","syslog") >exec config agentA tail("/tmp/log/message") agentBESink("192.168.0.11") #經過實驗，好像一個邏輯節點，最多只能有一個source和sink.>...>exec config agentF tail("/tmp/log/message") agentBESink("192.168.0.11")

這時的配置情況可從master web中一目了然，此時已經可以達到我們最初的目的了。

以上通過flume shell進行的動態配置，在flume master web中都可以進行，在此我們就不做進一步說明啦。

高級動態配置

高級配置其實就是在上述簡單配置中增加了以下幾個特性來保證系統更好的運行：

• 多Master(Master節點的高可用)

• Collector Chain（Collector的高可用）

多Master的情況在上面已經有過介紹，包括用途和master個數等。下面來簡單看一下Collector Chain，其實也很簡單，就是在動態配置時，使用agent*Chain來指定多個Collector來保證其日誌傳輸的可用性。看一下一般正式環境中flume的邏輯圖：

這裡agentA和agentB指向collectorA，如果CollectorA crach了，根據配置的可靠性級別agent會有相應的動作，我們很可能為了保障高效傳輸而沒有選擇E2E（即使是這種方式，Agent本地日誌累積過多依然是一個問題），一般會配置多個Collector，形成collector

>exec config agentC tail("/tmp/log/message") agentE2EChain("collectorB:35853","collectorA:35853") >exec config agentD tail("/tmp/log/message") agentE2EChain("collectorB:35853","collectorC:35853")

這樣collectorB在出問題時：

問題和總結

上述節點有如下幾類：master、agent、collector、storage，針對每類節點我們看一下高可用和有沒有可能引起性能瓶頸問題。

首先，storage層的失敗和collector層的失敗是一樣的，只要數據放不到最終的位置，就認為節點是失敗的。我們一定會根據收集數據的可靠性設定合適的傳輸模式，而且會根據我們的配置，自己控制collector接收數據的情況，collector的性能影響的是整個flume集群的數據吞吐量，所以collector最好單獨部署，所以一般不用考慮高可用問題。

然後，agent層的失敗，Flume數據安全級別的配置主要Agent的配置上，Agent提供三種級別發送數據到collector：E2E、DFO、BF，在些不贅述。看一下一位大牛的總結：

agent節點監控日誌文件夾下的所有文件，每一個agent最多監聽1024個文件，每一個文件在agent的都會有一個類似游標的東西，記錄監聽文件讀取的位置，這樣每次文件有新的記錄產生，那麼游標就會讀取增量記錄，根據agent配置發送到collector的安全層級屬性有E2E,DFO。如果是E2E的情況那麼agent節點會首先把文件寫入到agent節點的文件夾下，然後發送給collector，如果最終數據最終成功存儲到storage層，那麼agent刪除之前寫入的文件，如果沒有收到成功的信息，那麼就保留信息。 如果agent節點出現問題，那麼相當於所有的記錄信息都消失了，如果直接重新啟動，agent會認為日誌文件夾下的所有文件都是沒有監聽過的，沒有文件記錄的標示，所以會重新讀取文件，這樣，日誌就會有重複，具體恢復辦法如下： 將agent節點上監聽的日誌文件夾下已經發送的日誌文件移出，處理完，故障重新啟動agent即可。 註：在agent節點失敗的情況下，按照失敗的時間點，將時間點之前的數據文件移出，將flume.agent.logdir配置的文件夾清空，重新啟動agent。

最後，master失敗，master宕機，整個集群將不能工作，在重新啟動集群，將agent監聽的日誌文件夾下的所有文件移出，然後重新啟動master即可。在多master節點情況下，只要集群上正常工作的master大於總master數量的一半，集群就能正常工作，那麼只要恢復其中宕機的master即可。

問題總結

1.Flume在agent端採集數據的時候默認會在/tmp/flume-{user}下生成臨時的目錄用於存放agent自己截取的日誌文件，如果文件過大導致磁碟寫滿那麼agent端會報出 Error closing logicalNode a2-18 sink: No space left on device，所以在配置agent端的時候需要注意 <property> <name>flume.agent.logdir</name> <value>/data/tmp/flume-${user.name}/agent</value> </property> 屬性，只要保證flume在7*24小時運行過程agent端不會使該路徑flume.agent.logdir磁碟寫滿即可。

2. Flume在啟動時候會去尋找hadoop-core-*.jar的文件，需要修改標準版的hadoop核心jar包的名字 將hadoop-*-core.jar改成hadoop-core-*.jar。

3.Flume集群中的flume必須版本一致。否則會出現莫名其妙的錯誤。

4.Flume集群收集的日誌發送到hdfs上建立文件夾的時間依據是根據event的時間，在源代碼上是Clock.unixTime()，所以如果想要根據日誌生成的時間來生成文件的話，需要對 com.cloudera.flume.core.EventImpl 類的構造函數 public EventImpl(byte[] s, long timestamp, Priority pri, long nanoTime, String host, Map<String, byte[]> fields)重新寫，解析數組s的內容取出時間，賦給timestamp。注意：flume的框架會構造s內容是空的數組，用來發送類似簡單驗證的event，所以需要注意s內容為空的時候timestamp的問題。

5.如果collector和agent不在一個網段的話會發生閃斷的現象，這樣的話，就會造成agent端不能傳送數據個collector所以，在部署agent和collector最好在一個網段。

6.如果在啟動master時出現：「試著啟動hostname，但是hostname不在master列表裡的錯誤「，這是需要檢查是否主機地址和hostname配置的正確與否。

7.在源端，有一個比較大的缺陷，在tail類的source，不支持，斷點續傳功能。因為重啟node後沒有記錄上次文件讀到的位置，從而沒辦法知道，下次再讀時，從什麼地方開始讀。 特別是在日誌文件一直在增加的時候。flume的source node掛了。等flume的source再次開啟的這段時間內，增加的日誌內容，就沒辦法被source讀取到了。 不過flume有一個execStream的擴展，可以自己寫一個監控日誌增加情況，把增加的日誌，通過自己寫的工具把增加的內容，傳送給flume的node。再傳送給sink的node。

今天課程比較長了

但是也比較厲害了

大家一定要收藏起來不斷啃一啃哦