什麼是Flume

Flume 是 Cloudera 提供的一個高可用的，高可靠的，分散式的海量日誌採集、聚合和傳輸的系統。Flume 基於流式架構，靈活簡單。

Flume 基礎架構

Agent

Agent 是一個 JVM 程序，它以事件的形式將資料從源頭送至目的。 Agent 主要有 3 個部分組成，Source、Channel、Sink。

Source

Source 是負責接收資料到 Flume Agent 的元件。Source 元件可以處理各種型別、各種 格式的日誌資料，包括 avro、thrift、exec、jms、spooling directory、netcat、taildir、 sequence generator、syslog、http、legacy。 Sink

Sink 不斷地輪詢 Channel 中的事件且批量地移除它們，並將這些事件批量寫入到儲存 或索引系統、或者被髮送到另一個 Flume Agent。 Sink 元件目的地包括 hdfs、logger、avro、thrift、ipc、file、HBase、solr、自定 義。

Channel

Channel 是位於 Source 和 Sink 之間的緩衝區。因此，Channel 允許 Source 和 Sink 運 作在不同的速率上。Channel 是執行緒安全的，可以同時處理幾個 Source 的寫入操作和幾個 Sink 的讀取操作。 Flume 自帶兩種 Channel：Memory Channel 和 File Channel。 Memory Channel 是記憶體中的佇列。Memory Channel 在不需要關心資料丟失的情景下適 用。如果需要關心資料丟失，那麼 Memory Channel 就不應該使用，因為程式死亡、機器宕 機或者重啟都會導致資料丟失。 File Channel 將所有事件寫到磁碟。因此在程式關閉或機器宕機的情況下不會丟失資料。

Event

傳輸單元，Flume 資料傳輸的基本單元，以 Event 的形式將資料從源頭送至目的地。 Event 由 Header 和 Body 兩部分組成，Header 用來存放該 event 的一些屬性，為 K-V 結構， Body 用來存放該條資料，形式為位元組陣列。

Flume 事務

 Put事務

•doPut:將批資料先寫入臨時緩衝區putList

•doCommit:檢查channel記憶體佇列是否足夠合併。

•doRollback:channel記憶體佇列空間不足，回滾資料（直接清除putList中的資料）

 Put事務流程：事務開始的時候會呼叫一個doPut方法，doPut

現在這批資料順利的放到putList之後，接下來可以呼叫 doCommit方法，把putList中所有的event放到channel中，成功放完之後就清空putList。

以上是順利的情況下，那如果事務進行的過程中出問題了怎麼解決呢？

第一種問題：資料傳輸到channel過程出問題
在doCommit提交之後，事務在向channel放的過程中，事務容易出問題。比如：sink那邊取資料慢，而source這邊放資料速度快，就容易造成channel中的資料的積壓，這個時候就會造成putList中的資料放不進去。那現在事務出問題了，如何解決呢？

通過呼叫doRollback方法，doRollback方法會進行兩項操作：1、將putList清空； 2、丟擲channelException異常。這個時候source就會捕捉到doRollback丟擲的異常，然後source就會把剛才的一批資料重新採集一下（不一定採集得到），採集完之後重新走事務的流程。這就是事務的回滾。
(putList的資料在向channel傳送之前先檢查一下channel的容量能否放得下，如果放不下，就一個都不放。)

第二種問題：資料採集過程出問題
有這麼種場景，source採集資料時候採用的是tailDir source，而我們因為某種原因將監控的目錄檔案刪除，這個時候就會出現問題，同樣地，出現問題的解決方式是呼叫doRollback方法來對事務進行回滾。

 Take事務

•doTake:將資料取到臨時緩衝區takeList，並將資料傳送到HDFS

•doCommit:如果資料全部發送成功，則清除臨時緩衝區takeList

•doRollback:資料傳送過程中如果出現異常，rollback將臨時緩衝區takeList中的資料歸還給channel記憶體佇列。

Take事務同樣也有takeList，HDFS sink配置也有一個 batch size，這個引數決定sink從channel取資料的時候一次取多少個，所以這batch size 得小於takeList的大小，而takeList的大小取決於transaction capacity的大小，同樣是channel 中的引數。

Take事務流程：事務開始後，doTake方法會將channel中的event剪下到takeList中，當然，後面接的是HDFS Sink的話，在把channel中的event剪下到takeList中的同時也往寫入HDFS的IO緩衝流中放一份event(資料寫入HDFS是先寫入IO緩衝流然後flush到HDFS）。

當takeList中存放了batch size 數量的event之後，就會呼叫doCommit方法，doCommit方法會做兩個操作：1、針對HDFS Sink，手動呼叫IO流的flush方法，將IO流緩衝區的資料寫入到HDFS磁碟中；2、然後直接清空takeList中的資料。

以上是順利的情況下，那如果事務進行的過程中出問題了怎麼解決呢？

什麼時候最容易出問題呢？——flush到HDFS的時候組容易出問題

  如：flush到HDFS的時候，可能由於網路原因超時導致資料傳輸失敗，這個時候同樣地呼叫doRollback方法來進行回滾，回滾的時候，由於takeList中還有備份資料，所以將takeList中的資料原封不動地還給channel，這時候就完成了事務的回滾。

  但是，如果flush到HDFS的時候，資料flush了一半之後出問題了，這意味著已經有一半的資料已經發送到HDFS上面了，現在出了問題，同樣需要呼叫doRollback方法來進行回滾，回滾並沒有“一半”之說，它只會把整個takeList中的資料返回給channel，然後繼續進行資料的讀寫。這樣開啟下一個事務的時候就容易造成資料重複的問題。

  所以，在某種程度上，flume對資料進行採集傳輸的時候，它有可能會造成資料的重複，但是其資料不丟失。

Flume傳輸是否會丟失或重複資料

這個問題需要分情況來看，需要結合具體使用的source、channel和sink來分析。

source：exec source 後面接tail -f，資料有可能丟；TailDir source 不會丟資料的，可以保證資料不丟失。

sink：資料有可能重複，但是不會丟失。

channel： 要想資料不丟失的話，還是要用 File channel，而memory channel 在flume掛掉的時候還是有可能造成資料的丟失的。

 Flume Agent 內部原理

 1）ChannelSelector

ChannelSelector 的作用就是選出 Event 將要被髮往哪個 Channel。其共有兩種型別， 分別是 Replicating（複製）和 Multiplexing（多路複用）。

ReplicatingSelector 會將同一個 Event 發往所有的 Channel，Multiplexing 會根據相 應的原則，將不同的 Event 發往不同的 Channel。

2）SinkProcessor

SinkProcessor 共 有 三 種 類 型 ， 分 別 是 DefaultSinkProcessor 、 LoadBalancingSinkProcessor 和 FailoverSinkProcessor

DefaultSinkProcessor 對應的是單個的Sink，LoadBalancingSinkProcessor 和 FailoverSinkProcessor 對應的是 Sink Group，LoadBalancingSinkProcessor 可以實現負載均衡的功能，FailoverSinkProcessor 可以錯誤恢復的功能。