分布式日志收集收集系統：Flume

jopen 9年前發布 | 67K 次閱讀 Flume

Flume是一個分布式、可靠、和高可用的海量日志采集、聚合和傳輸的系統。支持在系統中定制各類數據發送方，用于收集數據；同時，Flume提供對數據進行簡單處理，并寫到各種數據接受方（可定制）的能力。Flume 初始的發行版本目前被統稱為 Flume OG（original generation），屬于 cloudera。但隨著 Flume 功能的擴展，Flume OG 代碼工程臃腫、核心組件設計不合理、核心配置不標準等缺點暴露出來，尤其是在 Flume OG 的最后一個發行版本 0.94.0 中，日志傳輸不穩定的現象尤為嚴重。為了解決這些問題，2011 年 10 月 22 號，cloudera 完成了 Flume-728，對 Flume 進行了里程碑式的改動：重構核心組件、核心配置以及代碼架構，重構后的版本統稱為 Flume NG（next generation）；改動的另一原因是將 Flume 納入 apache 旗下，cloudera Flume 改名為 Apache Flume。IBM 的這篇文章：Flume NG：Flume 發展史上的第一次革命，從基本組件以及用戶體驗的角度闡述 Flume OG 到 Flume NG 發生的革命性變化。

一、Flume OG

Flume OG的設計目標：

可靠性：當節點出現故障時，日志能夠被傳送到其他節點上而不會丟失。Flume提供了三種級別的可靠性保障，從強到弱依次分別為：end-to- end（收到數據agent首先將event寫到磁盤上，當數據傳送成功后，再刪除；如果數據發送失敗，可以重新發送。），Store on failure（這也是scribe采用的策略，當數據接收方crash時，將數據寫到本地，待恢復后，繼續發送），Best effort（數據發送到接收方后，不會進行確認）。
可擴展性：Flume采用了三層架構，分別為agent，collector和storage，每一層均可以水平擴展。其中，所有agent和 collector由master統一管理，這使得系統容易監控和維護，且master允許有多個（使用ZooKeeper進行管理和負載均衡），這就避免了單點故障問題。
可管理性：所有agent和Collector由master統一管理，這使得系統便于維護。多master情況，Flume利用 ZooKeeper和gossip，保證動態配置數據的一致性。用戶可以在master上查看各個數據源或者數據流執行情況，且可以對各個數據源配置和動態加載。Flume提供了web 和shell script command兩種形式對數據流進行管理。
功能可擴展性：用戶可以根據需要添加自己的agent，collector或者storage。此外，Flume自帶了很多組件，包括各種agent（file，syslog等），collector和storage（file，HDFS等）。

Flume OG的架構：

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

對于agent數據流配置就是從哪得到數據，把數據發送到哪個collector。
對于collector是接收agent發過來的數據，把數據發送到指定的目標機器上。

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

如前面提到的，Flume采用了分層架構：分別為Agent，Collector和Storage。Agent用于采集數據，Agent是 Flume中產生數據流的地方。同時，Agent會將產生的數據流傳輸到Collector。Collector用于對數據進行聚合，往往會產生一個更大的流，然后傳輸到Storage。其中，Agent和Collector均由兩部分組成：source和sink，source是數據來源，sink是數據去向。Flume使用兩個組件：Master和Node，Node根據在Master shell或web中動態配置，決定其是作為Agent還是Collector。

1、Agent

Agent的作用是將數據源的數據發送給collector。Flume自帶了很多直接可用的數據源（source），如：

text(“filename”)：將文件filename作為數據源，按行發送
tail(“filename”)：探測filename新產生的數據，按行發送出去
fsyslogTcp(5140)：監聽TCP的5140端口，并且接收到的數據發送出去
tailDir(“dirname”[, fileregex=".*"[, startFromEnd=false[, recurseDepth=0]]])：監聽目錄中的文件末尾，使用正則去選定需要監聽的文件（不包含目錄），recurseDepth為遞歸監聽其下子目錄的深度

更多可參見這位朋友的整理：http://www.cnblogs.com/zhangmiao-chp/archive/2011/05/18/2050465.html

同時提供了很多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失敗后，它會非常執著的再來一遍

更多可參見這位朋友的整理：http://www.cnblogs.com/zhangmiao-chp/archive/2011/05/18/2050472.html

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 NG

對于Flume OG ,可以說他是一個分布式日志收集系統，有Mater概念，依賴于Zookeeper，Agent用于采集數據，Agent是Flume中產生數據流的地方，同時，Agent會將產生的數據流傳輸到Collector。對應的，collector用于對數據進行聚合，往往會產生一個更大的流。而對于 Flume NG，它摒棄了Master和zookeeper，collector也沒有了，web配置臺也沒有了，只剩下source，sink和channel，此時一個Agent的概念包括source、channel和sink，完全由一個分布式系統變成了傳輸工具。不同機器之間的數據傳輸不再是OG那樣由 agent->collector，而是由一個Agent端的sink流向另一個agent的source。

Flume NG中的核心概念：

Client：生產數據，運行在一個獨立的線程。
Source：從Client收集數據，傳遞給Channel。可以接收外部源發送過來的數據。不同的 source，可以接受不同的數據格式。比如有目錄池(spooling directory)數據源，可以監控指定文件夾中的新文件變化，如果目錄中有文件產生，就會立刻讀取其內容。
Channel：是一個存儲地，接收source的輸出，直到有sink消費掉channel中的數據。Channel中的數據直到進入到下一個channel中或者進入終端才會被刪除。當sink寫入失敗后，可以自動重啟，不會造成數據丟失，因此很可靠。
Sink：會消費channel中的數據，然后送給外部源或者其他source。如數據可以寫入到HDFS或者HBase中。
Agent：使用JVM 運行Flume。每臺機器運行一個agent，但是可以在一個agent中包含多個sources和sinks。
Events：Flume NG傳輸的數據的基本單位是event，如果是文本文件，通常是一行記錄，這也是事務的基本單位。

Flume NG相對于Flume OG的主要變化：

sources和sinks 使用channels 進行鏈接
兩個主要channel：in-memory channel，非持久性支持，速度快； JDBC-based channel 持久性支持。
不再區分邏輯和物理node，所有物理節點統稱為agents,每個agents 都能運行0個或多個sources 和sinks
不再需要master節點和對zookeeper的依賴，配置文件簡單化。
插件化，一部分面對用戶，工具或系統開發人員。
使用Thrift、Avro Flume sources 可以從flume0.9.4 發送 events 到flume 1.x

Flume OG節點組成圖：

Flume NG節點組成圖：

對應于 OG 的特點，FLUM NG 的特點是：

NG 只有一種角色的節點：代理節點（agent）。
沒有 collector、master 節點。這是核心組件最核心的變化。
去除了 physical nodes、logical nodes 的概念和相關內容。
agent 節點的組成也發生了變化。

Flume NG 以agent為最小的獨立運行單位。一個agent就是一個JVM。單agent由Source、Sink和Channel三大組件構成。

Flume的數據流由事件(Event)貫穿始終。事件是Flume的基本數據單位，它攜帶日志數據(字節數組形式)并且攜帶有頭信息，這些 Event由Agent外部的Source，比如上圖中的Web Server生成。當Source捕獲事件后會進行特定的格式化，然后Source會把事件推入(單個或多個)Channel中。可以把Channel看作是一個緩沖區，它將保存事件直到Sink處理完該事件。Sink負責持久化日志或者把事件推向另一個Source。值得注意的是，Flume提供了大量內置的Source、Channel和Sink類型。不同類型的Source、Channel和Sink可以自由組合。組合方式基于用戶設置的配置文件，非常靈活。比如：Channel可以把事件暫存在內存里，也可以持久化到本地硬盤上。Sink可以把日志寫入HDFS, HBase，甚至是另外一個Source等等。Flume支持用戶建立多級流，也就是說，多個agent可以協同工作，并且支持Fan-in、Fan- out、Contextual Routing、Backup Routes。如下圖：

Flume 允許多個 agent 連在一起，形成前后相連的多級跳：

1、 source

Flume 支持 Avro，log4j，syslog 和 http post(body為json格式)。可以讓應用程序同已有的Source直接打交道，如AvroSource，SyslogTcpSource。也可以寫一個 Source，以 IPC 或 RPC 的方式接入自己的應用，Avro和 Thrift 都可以(分別有 NettyAvroRpcClient 和 ThriftRpcClient 實現了 RpcClient接口)，其中 Avro 是默認的 RPC 協議。具體代碼級別的 Client 端數據接入，可以參考官方手冊。對現有程序改動最小的使用方式是使用是直接讀取程序原來記錄的日志文件，基本可以實現無縫接入，不需要對現有程序進行任何改動。對于直接讀取文件 Source,有兩種方式：

ExecSource: 以運行 Linux 命令的方式，持續的輸出最新的數據，如 tail -F 文件名指令，在這種方式下，取的文件名必須是指定的。 ExecSource 可以實現對日志的實時收集，但是存在Flume不運行或者指令執行出錯時，將無法收集到日志數據，無法保證日志數據的完整性。
SpoolSource: 監測配置的目錄下新增的文件，并將文件中的數據讀取出來。需要注意兩點：拷貝到 spool 目錄下的文件不可以再打開編輯；spool 目錄下不可包含相應的子目錄。SpoolSource 雖然無法實現實時的收集數據，但是可以使用以分鐘的方式分割文件，趨近于實時。如果應用無法實現以分鐘切割日志文件的話，可以兩種收集方式結合使用。在實際使用的過程中，可以結合 log4j 使用，使用 log4j的時候，將 log4j 的文件分割機制設為1分鐘一次，將文件拷貝到spool的監控目錄。log4j 有一個 TimeRolling 的插件，可以把 log4j 分割文件到 spool 目錄。基本實現了實時的監控。Flume 在傳完文件之后，將會修改文件的后綴，變為 .COMPLETED（后綴也可以在配置文件中靈活指定）

2、Channel

當前有幾個 channel 可供選擇，分別是 Memory Channel, JDBC Channel , File Channel，Psuedo Transaction Channel。比較常見的是前三種 channel。

MemoryChannel 可以實現高速的吞吐，但是無法保證數據的完整性。
MemoryRecoverChannel 在官方文檔的建議上已經建義使用FileChannel來替換。
FileChannel保證數據的完整性與一致性。在具體配置FileChannel時，建議FileChannel設置的目錄和程序日志文件保存的目錄設成不同的磁盤，以便提高效率。

File Channel 是一個持久化的隧道（channel），它持久化所有的事件，并將其存儲到磁盤中。因此，即使 Java 虛擬機當掉，或者操作系統崩潰或重啟，再或者事件沒有在管道中成功地傳遞到下一個代理（agent），這一切都不會造成數據丟失。Memory Channel 是一個不穩定的隧道，其原因是由于它在內存中存儲所有事件。如果 java 進程死掉，任何存儲在內存的事件將會丟失。另外，內存的空間收到 RAM大小的限制,而 File Channel 這方面是它的優勢，只要磁盤空間足夠，它就可以將所有事件數據存儲到磁盤上。

3、sink

Sink在設置存儲數據時，可以向文件系統、數據庫、hadoop存數據，在日志數據較少時，可以將數據存儲在文件系中，并且設定一定的時間間隔保存數據。在日志數據較多時，可以將相應的日志數據存儲到Hadoop中，便于日后進行相應的數據分析。更多sink的內容可以參考官方手冊。

從整體上講，NG 在核心組件上進行了大規模的調整，核心組件的數目由 7 刪減到 4。由于 Flume 的使用涉及到眾多因素，如 avro、 thrift、hdfs、jdbc、zookeeper 等，而這些組件和 Flume 的整合都需要關聯到所有組件。所以核心組件的改革對整個 Flume 的使用影響深遠：