Flume技術(shù)原理深入理解

1、Flume技術(shù)原理深入理解技術(shù)創(chuàng)新，變革未來智慧ITFlume簡介flume是一個分布式、可靠、和高可用的海量日志采集、聚合和傳輸?shù)南到y(tǒng)。支持在日志系統(tǒng)中定制各類數(shù)據(jù)發(fā)送方，用于收集數(shù)據(jù);同時，F(xiàn)lume提供對數(shù)據(jù)進行簡單處理，并寫到各種數(shù)據(jù)接受方(比如文本、HDFS、Hbase等)的能力。flume的數(shù)據(jù)流由事件(Event)貫穿始終。事件是Flume的基本數(shù)據(jù)單位，它攜帶日志數(shù)據(jù)(字節(jié)數(shù)組形式)并且攜帶有頭信息，這些Event由Agent外部的Source生成，當Source捕獲事件后會進行特定的格式化，然后Source會把事件推入(單個或多個)Channel中。你可以把Channel看作

2、是一個緩沖區(qū)，它將保存事件直到Sink處理完該事件。Sink負責(zé)持久化日志或者把事件推向另一個Source。flume版本對比 Flume-og采用了多Master的方式。為了保證配置數(shù)據(jù)的一致性，F(xiàn)lume引入了ZooKeeper，用于保存配置數(shù)據(jù)，ZooKeeper本身可保證配置數(shù)據(jù)的一致性和高可用，另外，在配置數(shù)據(jù)發(fā)生變化時，ZooKeeper可以通知Flume Master節(jié)點。Flume Master間使用gossip協(xié)議同步數(shù)據(jù)。 Flume-ng最明顯的改動就是取消了集中管理配置的 Master 和 Zookeeper，變?yōu)橐粋€純粹的傳輸工具。Flume-ng另一個主要的不同點是

3、讀入數(shù)據(jù)和寫出數(shù)據(jù)現(xiàn)在由不同的工作線程處理（稱為 Runner）。 在 Flume-og 中，讀入線程同樣做寫出工作（除了故障重試）。如果寫出慢的話（不是完全失敗），它將阻塞 Flume 接收數(shù)據(jù)的能力。這種異步的設(shè)計使讀入線程可以順暢的工作而無需關(guān)注下游的任何問題。flume的優(yōu)勢1. Flume可以將應(yīng)用產(chǎn)生的數(shù)據(jù)存儲到任何集中存儲器中，比如HDFS,HBase2. 當收集數(shù)據(jù)的速度超過將寫入數(shù)據(jù)的時候，也就是當收集信息遇到峰值時，這時候收集的信息非常大，甚至超過了系統(tǒng)的寫入數(shù)據(jù)能力，這時候，F(xiàn)lume會在數(shù)據(jù)生產(chǎn)者和數(shù)據(jù)收容器間做出調(diào)整，保證其能夠在兩者之間提供平穩(wěn)的數(shù)據(jù).3. 提供上下

4、文路由特征4. Flume的管道是基于事務(wù)，保證了數(shù)據(jù)在傳送和接收時的一致性.5. Flume是可靠的，容錯性高的，可升級的，易管理的,并且可定制的。flume的特征1. Flume可以高效率的將多個網(wǎng)站服務(wù)器中收集的日志信息存入HDFS/HBase中2. 使用Flume，我們可以將從多個服務(wù)器中獲取的數(shù)據(jù)迅速的移交給Hadoop中3. 除了日志信息，F(xiàn)lume同時也可以用來接入收集規(guī)模宏大的社交網(wǎng)絡(luò)節(jié)點事件數(shù)據(jù)，比如facebook,twitter,電商網(wǎng)站如亞馬遜，flipkart等4. 支持各種接入資源數(shù)據(jù)的類型以及接出數(shù)據(jù)類型5. 支持多路徑流量，多管道接入流量，多管道接出流量，上下文

5、路由等6. 可以被水平擴展Flume架構(gòu)Flume 運行的核心是 Agent。Flume以agent為最小的獨立運行單位。一個agent就是一個JVM。它是一個完整的數(shù)據(jù)收集工具，含有三個核心組件，分別是source、 channel、 sink。通過這些組件， Event 可以從一個地方流向另一個地方，如下圖所示。Flume組件Client：Client生產(chǎn)數(shù)據(jù)，運行在一個獨立的線程。Event： 一個數(shù)據(jù)單元，消息頭和消息體組成。（Events可以是日志記錄、 avro 對象等。）Flow： Event從源點到達目的點的遷移的抽象。Agent： 一個獨立的Flume進程，包含組件Sourc

6、e、 Channel、 Sink。（Agent使用JVM 運行Flume。每臺機器運行一個agent，但是可以在一個agent中包含多個sources和sinks。）Source： 數(shù)據(jù)收集組件。（source從Client收集數(shù)據(jù)，傳遞給Channel）Channel： 中轉(zhuǎn)Event的一個臨時存儲，保存由Source組件傳遞過來的Event。（Channel連接 sources 和 sinks ，這個有點像一個隊列。）Sink： 從Channel中讀取并移除Event， 將Event傳遞到FlowPipeline中的下一個Agent（如果有的話）（Sink從Channel收集數(shù)據(jù)，運行在一

7、個獨立線程。）SourceSource是數(shù)據(jù)的收集端，負責(zé)將數(shù)據(jù)捕獲后進行特殊的格式化，將數(shù)據(jù)封裝到事件（event） 里，然后將事件推入Channel中。 Flume提供了很多內(nèi)置的Source， 支持 Avro， log4j， syslog 和 http post(body為json格式)?？梢宰寫?yīng)用程序同已有的Source直接打交道，如AvroSource如果內(nèi)置的Source無法滿足需要， Flume還支持自定義Source。SourceSource具體作用：AvroSource：監(jiān)聽一個avro服務(wù)端口，采集Avro數(shù)據(jù)序列化后的數(shù)據(jù)；Thrift Source：監(jiān)聽一個Thrift

8、 服務(wù)端口，采集Thrift數(shù)據(jù)序列化后的數(shù)據(jù)；Exec Source：基于Unix的command在標準輸出上采集數(shù)據(jù)；JMS Source：Java消息服務(wù)數(shù)據(jù)源，Java消息服務(wù)是一個與具體平臺無關(guān)的API，這是支持jms規(guī)范的數(shù)據(jù)源采集；Spooling Directory Source：通過文件夾里的新增的文件作為數(shù)據(jù)源的采集；【測試header】Kafka Source：從kafka服務(wù)中采集數(shù)據(jù)。NetCat Source： 綁定的端口（tcp、udp），將流經(jīng)端口的每一個文本行數(shù)據(jù)作為Event輸入HTTP Source：監(jiān)聽HTTP POST和 GET產(chǎn)生的數(shù)據(jù)的采集Chan

9、nel Channel是連接Source和Sink的組件，大家可以將它看做一個數(shù)據(jù)的緩沖區(qū)（數(shù)據(jù)隊列），它可以將事件暫存到內(nèi)存中也可以持久化到本地磁盤上， 直到Sink處理完該事件。介紹兩個較為常用的Channel， MemoryChannel和FileChannel。Channel：一個數(shù)據(jù)的存儲池，中間通道。主要作用：接受source傳出的數(shù)據(jù)，向sink指定的目的地傳輸。Channel中的數(shù)據(jù)直到進入到下一個channel中或者進入終端才會被刪除。當sink寫入失敗后，可以自動重寫，不會造成數(shù)據(jù)丟失，因此很可靠。channel的類型很多比如:內(nèi)存中、jdbc數(shù)據(jù)源中、文件形式存儲等。常見

10、采集的數(shù)據(jù)類型：Memory Channel、File Channel、JDBC Channel、Kafka Channel等詳細查看：/FlumeUserGuide.html#flume-channelsChannel具體作用：Memory Channel：使用內(nèi)存作為數(shù)據(jù)的存儲。JDBC Channel：使用jdbc數(shù)據(jù)源來作為數(shù)據(jù)的存儲。Kafka Channel：使用kafka服務(wù)來作為數(shù)據(jù)的存儲。File Channel：使用文件來作為數(shù)據(jù)的存儲。Spillable Memory Channel：使用內(nèi)存和文件作為數(shù)據(jù)的存儲，即：先存在內(nèi)存中，如果內(nèi)存中數(shù)據(jù)達到閥值則flush到文件

11、中。Sink Sink從Channel中取出事件，然后將數(shù)據(jù)發(fā)到別處，可以向文件系統(tǒng)、數(shù)據(jù)庫、 hadoop存數(shù)據(jù)， 也可以是其他agent的Source。在日志數(shù)據(jù)較少時，可以將數(shù)據(jù)存儲在文件系統(tǒng)中，并且設(shè)定一定的時間間隔保存數(shù)據(jù)。Sink：數(shù)據(jù)的最終的目的地。主要作用：接受channel寫入的數(shù)據(jù)以指定的形式表現(xiàn)出來（或存儲或展示）。sink的表現(xiàn)形式很多比如:打印到控制臺、hdfs上、avro服務(wù)中、文件中等。常見采集的數(shù)據(jù)類型：HDFS Sink、Hive Sink、Logger Sink、Avro Sink、Thrift Sink、File Roll Sink、HBaseSink、K

12、afka Sink等詳細查看：/FlumeUserGuide.html#flume-sinksHDFSSink需要有hdfs的配置文件和類庫。一般采取多個sink匯聚到一臺采集機器負責(zé)推送到hdfs。Sink具體作用：HDFS Sink：將數(shù)據(jù)傳輸?shù)絟dfs集群中。Hive Sink：將數(shù)據(jù)傳輸?shù)絟ive的表中。Logger Sink：將數(shù)據(jù)作為日志處理（根據(jù)flume中的設(shè)置的日志的級別顯示）。Avro Sink：數(shù)據(jù)被轉(zhuǎn)換成Avro Event，然后發(fā)送到配置的RPC端口上。Thrift Sink：數(shù)據(jù)被轉(zhuǎn)換成Thrift Event，然后發(fā)送到配置的RPC端口上。IRC Sink：數(shù)據(jù)向

13、指定的IRC服務(wù)和端口中發(fā)送。File Roll Sink：數(shù)據(jù)傳輸?shù)奖镜匚募?。Null Sink：取消數(shù)據(jù)的傳輸，即不發(fā)送到任何目的地。HBaseSink：將數(shù)據(jù)發(fā)往hbase數(shù)據(jù)庫中。MorphlineSolrSink：數(shù)據(jù)發(fā)送到Solr搜索服務(wù)器（集群）。ElasticSearchSink：數(shù)據(jù)發(fā)送到Elastic Search搜索服務(wù)器（集群）。Kafka Sink：將數(shù)據(jù)發(fā)送到kafka服務(wù)中。Flume數(shù)據(jù)流Flume 的核心是把數(shù)據(jù)從數(shù)據(jù)源收集過來，再送到目的地。為了保證輸送一定成功，在送到目的地之前，會先緩存數(shù)據(jù)，待數(shù)據(jù)真正到達目的地后，刪除自己緩存的數(shù)據(jù) Flume 傳輸?shù)?/p>

14、數(shù)據(jù)的基本單位是 Event，如果是文本文件，通常是一行記錄，這也是事務(wù)的基本單位。 Event 從 Source，流向 Channel，再到 Sink，本身為一個 byte 數(shù)組，并可攜帶 headers 信息。 Event 代表著一個數(shù)據(jù)流的最小完整單元，從外部數(shù)據(jù)源來，向外部的目的地去。值得注意的是，F(xiàn)lume提供了大量內(nèi)置的Source、Channel和Sink類型。不同類型的Source,Channel和Sink可以自由組合。組合方式基于用戶設(shè)置的配置文件，非常靈活。比如：Channel可以把事件暫存在內(nèi)存里，也可以持久化到本地硬盤上。Sink可以把日志寫入HDFS, HBase，甚

15、至是另外一個Source等等。Flume支持用戶建立多級流，也就是說，多個agent可以協(xié)同工作，并且支持Fan-in、Fan-out、Contextual Routing、Backup Routes，這也正是Flume強大之處。如下圖所示：Flume可靠性Flume 使用事務(wù)性的方式保證傳送Event整個過程的可靠性。 Sink 必須在Event 被存入Channel 后，或者，已經(jīng)被傳達到下一站agent里，又或者，已經(jīng)被存入外部數(shù)據(jù)目的地之后，才能把 Event 從 Channel 中 remove 掉。這樣數(shù)據(jù)流里的 event 無論是在一個 agent 里還是多個 agent 之間流

16、轉(zhuǎn)，都能保證可靠，因為以上的事務(wù)保證了 event 會被成功存儲起來。比如 Flume支持在本地保存一份文件 channel 作為備份，而memory channel 將event存在內(nèi)存 queue 里，速度快，但丟失的話無法恢復(fù)。Flume的安裝見安裝文檔flume的使用-案例一 NetCat Source NetCat Source：監(jiān)聽一個指定的網(wǎng)絡(luò)端口， 即只要應(yīng)用程序向這個端口里面寫數(shù)據(jù)，這個source組件就可以獲取到信息。NetCat Source 配置文件配置文件：netcat.conf# 指定Agent的組件名稱（a），一個進程a.sources=r1a.channels=

17、c1 a.sinks=k1a.sources.r1.type=netcat a.sources.r1.bind=1a.sources.r1.port=8888a.sources.r1.channels=c1a.channels.c1.type=memory a.channels.c1.capacity=1000 a.channels.c1.transactionCapacity=1000a.sinks.k1.channel=c1 a.sinks.k1.type=logger在conf目錄下啟動flume agent a 服務(wù)端命令：flume-ng agent -n a -c ./conf -

18、f ./netcat.conf -Dflume.root.logger=DEBUG,console進入到 /usr/local/soft/flume/conf復(fù)制一個配置文件模板：cp perties.template netcat.confSpooling Directory Source：監(jiān)聽一個指定的目錄，即只要應(yīng)用程序向這個指定的目錄中添加新的文 件，source組件就可以獲取到該信息，并解析該文件的內(nèi)容，然后寫入到channle。寫入完成后，標記該文件已 完成或者刪除該文件。其中 Sink：logger Channel：memoryflume案例-案二：Spooling Direct

19、ory SourceSpooling Directory SourceSpooling Directory Source的兩個注意事項：If a file is written to after being placed into the spooling directory, Flume will print an error to its log file and stop processing.即：拷貝到spool目錄下的文件不可以再打開編輯If a file name is reused at a later time, Flume will print an error to its

20、 log file and stop processing.即：不能將具有相同文件名字的文件拷貝到這個目錄下a1.sources = r1a1.sinks = k1 a1.channels = c1#Describe/configure the sourcea1.sources.r1.type = spooldir a1.sources.r1.spoolDir = /usr/local/datainput a1.sources.r1.fileHeader = true erceptors = i1 erceptors.i1.t

21、ype = timestamp# Describe the sinka1.sinks.k1.type = logger# Use a channel which buffers events in memory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1

22、啟動命令：flume-ng agent -c ./conf -f ./perties -n a1 -Dflume.root.logger=INFO,console在 conf目錄下新建文件：perties控制臺顯示頁面案例三 sink-hdfs#具體定義sinka.sinks.k1.type = hdfsa.sinks.k1.hdfs.path =hdfs:/master:9000/flume_hdfsa.sinks.k1.hdfs.filePrefix = events-a.sinks.k1.hdfs.minBlockReplicas=1a.sinks.k1.hdfs.fileType =

23、DataStream#a.sinks.k1.hdfs.fileType = CompressedStream#a.sinks.k1.hdfs.codeC = gzip#不按照條數(shù)生成文件a.sinks.k1.hdfs.rollCount = 0a.sinks.k1.hdfs.rollSize = 33554432#每隔N s將臨時文件滾動成一個目標文件a.sinks.k1.hdfs.rollInterval =0a.sinks.k1.hdfs.idleTimeout=0 j代碼中采集日志寫入flume-avro_sourceflume的conf配置a.sources=r1a.channels=

24、c1a.sinks=k1a.sources.r1.type = avroa.sources.r1.bind = 32a.sources.r1.port = 41414a.sources.r1.channels=c1a.channels.c1.type=memorya.channels.c1.capacity=1000a.channels.c1.transactionCapacity=1000a.sinks.k1.channel=c1a.sinks.k1.type=logger程序日志采集-log4j-appenderLog4J Appender：通過log4j直接將web的日志達到flume中

25、的avro_source中。實現(xiàn)步驟：1、pom中添加依賴 org.apache.flume flume-ng-core 1.6.0 org.apache.flume.flume-ng-clients flume-ng-log4jappender 1.6.0 2、在perties配置信息。3、程序打印日志。flume采集到HDFS部分#具體定義sinka.sinks.k1.type = hdfsa.sinks.k1.hdfs.path =hdfs:/master:9000/flume_hdfsa.sinks.k1.hdfs.filePrefix = events-a.sinks.k1.hdfs

26、.minBlockReplicas=1a.sinks.k1.hdfs.fileType = DataStream#a.sinks.k1.hdfs.fileType = CompressedStream#a.sinks.k1.hdfs.codeC = gzip#不按照條數(shù)生成文件a.sinks.k1.hdfs.rollCount = 0a.sinks.k1.hdfs.rollSize = 33554432#每隔N s將臨時文件滾動成一個目標文件a.sinks.k1.hdfs.rollInterval =0a.sinks.k1.hdfs.idleTimeout=0 log4j配置項log4j.rootLogger=INFO,stdout,flumelog4j.appender.stdout = org.apache.log4j.ConsoleAppenderlog4j.appender.stdout.Target = System.outlog4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.a