Flume NG 简介及配置实战
2019獨角獸企業重金招聘Python工程師標準>>>
Flume NG 簡介及配置實戰 博客分類： 分布式計算- 1、Flume 的一些核心概念：
- 1.1 數據流模型
- 1.2 高可靠性
- 1.3 可恢復性
- 2、Flume 整體架構介紹
- 2.1 Exec source
- 2.2 Spooling Directory Source
- 3、常用架構、功能配置示例
- 3.1 先來個簡單的：單節點 Flume 配置
- 3.2 單節點 Flume 直接寫入 HDFS
- 3.3 來一個常見架構：多 agent 匯聚寫入 HDFS
- 3.3.1 在各個webserv日志機上配置 Flume Client
- 3.3.2 在匯聚節點配置 Flume server
- 4、可能遇到的問題：
- 4.1 OOM 問題：
- 4.2 JDK 版本不兼容問題：
- 4.3 小文件寫入 HDFS 延時的問題
- 4.4 數據重復寫入、丟失問題
- 4.5 tail 斷點續傳的問題：
- 4.6 在 Flume 中如何修改、丟棄、按預定義規則分類存儲數據？
- 5、Refer：
Flume 作為 cloudera 開發的實時日志收集系統，受到了業界的認可與廣泛應用。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 NG （1.x.x）的主要變化：
- sources和sinks 使用channels 進行鏈接
- 兩個主要channel 。1， ?in-memory channel ?非持久性支持，速度快。2 ， JDBC-based channel 持久性支持。
- 不再區分邏輯和物理node，所有物理節點統稱為 “agents”,每個agents 都能運行0個或多個sources 和sinks
- 不再需要master節點和對zookeeper的依賴，配置文件簡單化。
- 插件化，一部分面對用戶，工具或系統開發人員。
- 使用Thrift、Avro Flume sources 可以從flume0.9.4 發送 events ?到flume 1.x
注：本文所使用的 Flume 版本為 flume-1.4.0-cdh4.7.0，不需要額外的安裝過程，解壓縮即可用。?
1、Flume 的一些核心概念：
?
| Agent | 使用JVM 運行Flume。每臺機器運行一個agent，但是可以在一個agent中包含多個sources和sinks。 |
| Client | 生產數據，運行在一個獨立的線程。 |
| Source | 從Client收集數據，傳遞給Channel。 |
| Sink | 從Channel收集數據，運行在一個獨立線程。 |
| Channel | 連接 sources 和 sinks ，這個有點像一個隊列。 |
| Events | 可以是日志記錄、 avro 對象等。 |
1.1 數據流模型
Flume以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就這些能耐那就大錯特錯了。Flume支持用戶建立多級流，也就是說，多個agent可以協同工作，并且支持Fan-in、Fan-out、Contextual Routing、Backup Routes。如下圖所示：
1.2 高可靠性
作為生產環境運行的軟件，高可靠性是必須的。從單agent來看，Flume使用基于事務的數據傳遞方式來保證事件傳遞的可靠性。Source和Sink被封裝進一個事務。事件被存放在Channel中直到該事件被處理，Channel中的事件才會被移除。這是Flume提供的點到點的可靠機制。
從多級流來看，前一個agent的sink和后一個agent的source同樣有它們的事務來保障數據的可靠性。
1.3 可恢復性
還是靠Channel。推薦使用FileChannel，事件持久化在本地文件系統里(性能較差)。2、Flume 整體架構介紹
Flume架構整體上看就是?source-->channel-->sink?的三層架構（參見最上面的 圖一），類似生成者和消費者的架構，他們之間通過queue（channel）傳輸，解耦。
Source:完成對日志數據的收集，分成 transtion 和 event 打入到channel之中。?
Channel:主要提供一個隊列的功能，對source提供中的數據進行簡單的緩存。?
Sink:取出Channel中的數據，進行相應的存儲文件系統，數據庫，或者提交到遠程服務器。?
對現有程序改動最小的使用方式是使用是直接讀取程序原來記錄的日志文件，基本可以實現無縫接入，不需要對現有程序進行任何改動。?
對于直接讀取文件Source, 主要有兩種方式：?
2.1 Exec?source
可通過寫Unix command的方式組織數據，最常用的就是tail -F [file]。可以實現實時傳輸，但在flume不運行和腳本錯誤時，會丟數據，也不支持斷點續傳功能。因為沒有記錄上次文件讀到的位置，從而沒辦法知道，下次再讀時，從什么地方開始讀。特別是在日志文件一直在增加的時候。flume的source掛了。等flume的source再次開啟的這段時間內，增加的日志內容，就沒辦法被source讀取到了。不過flume有一個execStream的擴展，可以自己寫一個監控日志增加情況，把增加的日志，通過自己寫的工具把增加的內容，傳送給flume的node。再傳送給sink的node。要是能在tail類的source中能支持，在node掛掉這段時間的內容，等下次node開啟后在繼續傳送，那就更完美了。
2.2 Spooling?Directory Source
SpoolSource:是監測配置的目錄下新增的文件，并將文件中的數據讀取出來，可實現準實時。需要注意兩點：1、拷貝到spool目錄下的文件不可以再打開編輯。2、spool目錄下不可包含相應的子目錄。在實際使用的過程中，可以結合log4j使用，使用log4j的時候，將log4j的文件分割機制設為1分鐘一次，將文件拷貝到spool的監控目錄。log4j有一個TimeRolling的插件，可以把log4j分割的文件到spool目錄。基本實現了實時的監控。Flume在傳完文件之后，將會修改文件的后綴，變為.COMPLETED（后綴也可以在配置文件中靈活指定）?ExecSource，SpoolSource對比：ExecSource可以實現對日志的實時收集，但是存在Flume不運行或者指令執行出錯時，將無法收集到日志數據，無法何證日志數據的完整性。SpoolSource雖然無法實現實時的收集數據，但是可以使用以分鐘的方式分割文件，趨近于實時。如果應用無法實現以分鐘切割日志文件的話，可以兩種收集方式結合使用。?
Channel有多種方式：有MemoryChannel, JDBC Channel, MemoryRecoverChannel, FileChannel。MemoryChannel可以實現高速的吞吐，但是無法保證數據的完整性。MemoryRecoverChannel在官方文檔的建議上已經建義使用FileChannel來替換。FileChannel保證數據的完整性與一致性。在具體配置FileChannel時，建議FileChannel設置的目錄和程序日志文件保存的目錄設成不同的磁盤，以便提高效率。?
Sink在設置存儲數據時，可以向文件系統中，數據庫中，hadoop中儲數據，在日志數據較少時，可以將數據存儲在文件系中，并且設定一定的時間間隔保存數據。在日志數據較多時，可以將相應的日志數據存儲到Hadoop中，便于日后進行相應的數據分析。?
3、常用架構、功能配置示例
3.1 先來個簡單的：單節點 Flume 配置
?
?| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 | # example.conf: A single-node Flume configuration # Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1. type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 # Describe the sink a1.sinks.k1. type = logger # Use a channel which buffers events in memory a1.channels.c1. type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 |
然后我們就可以啟動 Flume 了：
?
?| 1 | bin /flume-ng agent --conf conf --conf- file example.conf --name a1 -Dflume.root.logger=INFO,console |
PS：-Dflume.root.logger=INFO,console 僅為 debug 使用，請勿生產環境生搬硬套，否則大量的日志會返回到終端。。。
-c/--conf 后跟配置目錄，-f/--conf-file?后跟具體的配置文件，-n/--name?指定agent的名稱
?
?| 1 2 3 4 5 6 | $ telnet localhost 44444 Trying 127.0.0.1... Connected to localhost.localdomain (127.0.0.1). Escape character is '^]' . Hello world! <ENTER> OK |
?
?| 1 2 3 | 12 /06/19 15:32:19 INFO source .NetcatSource: Source starting 12 /06/19 15:32:19 INFO source .NetcatSource: Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[ /127 .0.0.1:44444] 12 /06/19 15:32:34 INFO sink.LoggerSink: Event: { headers:{} body: 48 65 6C 6C 6F 20 77 6F 72 6C 64 21 0D????????? Hello world!. } |
3.2?單節點 Flume?直接寫入 HDFS
?
?| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 | # Define a memory channel called ch1 on agent1 agent1.channels.ch1. type = memory agent1.channels.ch1.capacity = 100000 agent1.channels.ch1.transactionCapacity = 100000 agent1.channels.ch1.keep-alive = 30 # Define an Avro source called avro-source1 on agent1 and tell it # to bind to 0.0.0.0:41414. Connect it to channel ch1. #agent1.sources.avro-source1.channels = ch1 #agent1.sources.avro-source1.type = avro #agent1.sources.avro-source1.bind = 0.0.0.0 #agent1.sources.avro-source1.port = 41414 #agent1.sources.avro-source1.threads = 5 #define source monitor a file agent1.sources.avro-source1. type = exec agent1.sources.avro-source1.shell = /bin/bash -c agent1.sources.avro-source1. command = tail -n +0 -F /home/storm/tmp/id .txt agent1.sources.avro-source1.channels = ch1 agent1.sources.avro-source1.threads = 5 # Define a logger sink that simply logs all events it receives # and connect it to the other end of the same channel. agent1.sinks.log-sink1.channel = ch1 agent1.sinks.log-sink1. type = hdfs agent1.sinks.log-sink1.hdfs.path = hdfs: //192 .168.1.111:8020 /flumeTest agent1.sinks.log-sink1.hdfs.writeFormat = Text agent1.sinks.log-sink1.hdfs.fileType = DataStream agent1.sinks.log-sink1.hdfs.rollInterval = 0 agent1.sinks.log-sink1.hdfs.rollSize = 1000000 agent1.sinks.log-sink1.hdfs.rollCount = 0 agent1.sinks.log-sink1.hdfs.batchSize = 1000 agent1.sinks.log-sink1.hdfs.txnEventMax = 1000 agent1.sinks.log-sink1.hdfs.callTimeout = 60000 agent1.sinks.log-sink1.hdfs.appendTimeout = 60000 # Finally, now that we've defined all of our components, tell # agent1 which ones we want to activate. agent1.channels = ch1 agent1.sources = avro-source1 agent1.sinks = log-sink1 |
../bin/flume-ng agent --conf ../conf/ -f flume_directHDFS.conf -n agent1 -Dflume.root.logger=INFO,console
PS：實際環境中有這樣的需求，通過在多個agent端tail日志，發送給collector，collector再把數據收集，統一發送給HDFS存儲起來，當HDFS文件大小超過一定的大小或者超過在規定的時間間隔會生成一個文件。
Flume 實現了兩個Trigger，分別為SizeTriger（在調用HDFS輸出流寫的同時，count該流已經寫入的大小總和，若超過一定大小，則創建新的文件和輸出流，寫入操作指向新的輸出流，同時close以前的輸出流）和TimeTriger（開啟定時器，當到達該點時，自動創建新的文件和輸出流，新的寫入重定向到該流中，同時close以前的輸出流）。
3.3 來一個常見架構：多 agent 匯聚寫入 HDFS
?
3.3.1 在各個webserv日志機上配置 Flume Client
?
?| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 | # clientMainAgent clientMainAgent.channels = c1 clientMainAgent.sources? = s1 clientMainAgent.sinks??? = k1 k2 # clientMainAgent sinks group clientMainAgent.sinkgroups = g1 # clientMainAgent Spooling Directory Source clientMainAgent.sources.s1. type = spooldir clientMainAgent.sources.s1.spoolDir? = /dsap/rawdata/ clientMainAgent.sources.s1.fileHeader = true clientMainAgent.sources.s1.deletePolicy =immediate clientMainAgent.sources.s1.batchSize =1000 clientMainAgent.sources.s1.channels =c1 clientMainAgent.sources.s1.deserializer.maxLineLength =1048576 # clientMainAgent FileChannel clientMainAgent.channels.c1. type = file clientMainAgent.channels.c1.checkpointDir = /var/flume/fchannel/spool/checkpoint clientMainAgent.channels.c1.dataDirs = /var/flume/fchannel/spool/data clientMainAgent.channels.c1.capacity = 200000000 clientMainAgent.channels.c1.keep-alive = 30 clientMainAgent.channels.c1.write-timeout = 30 clientMainAgent.channels.c1.checkpoint-timeout=600 # clientMainAgent Sinks # k1 sink clientMainAgent.sinks.k1.channel = c1 clientMainAgent.sinks.k1. type = avro # connect to CollectorMainAgent clientMainAgent.sinks.k1. hostname = flume115 clientMainAgent.sinks.k1.port = 41415 # k2 sink clientMainAgent.sinks.k2.channel = c1 clientMainAgent.sinks.k2. type = avro # connect to CollectorBackupAgent clientMainAgent.sinks.k2. hostname = flume116 clientMainAgent.sinks.k2.port = 41415 # clientMainAgent sinks group clientMainAgent.sinkgroups.g1.sinks = k1 k2 # load_balance type clientMainAgent.sinkgroups.g1.processor. type = load_balance clientMainAgent.sinkgroups.g1.processor.backoff?? = true clientMainAgent.sinkgroups.g1.processor.selector? = random |
../bin/flume-ng agent --conf ../conf/ -f flume_Consolidation.conf -n clientMainAgent -Dflume.root.logger=DEBUG,console
3.3.2 在匯聚節點配置 Flume server
?
?| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 | # collectorMainAgent collectorMainAgent.channels = c2 collectorMainAgent.sources? = s2 collectorMainAgent.sinks??? =k1 k2 # collectorMainAgent AvroSource # collectorMainAgent.sources.s2. type = avro collectorMainAgent.sources.s2.bind = flume115 collectorMainAgent.sources.s2.port = 41415 collectorMainAgent.sources.s2.channels = c2 # collectorMainAgent FileChannel # collectorMainAgent.channels.c2. type = file collectorMainAgent.channels.c2.checkpointDir = /opt/var/flume/fchannel/spool/checkpoint collectorMainAgent.channels.c2.dataDirs = /opt/var/flume/fchannel/spool/data , /work/flume/fchannel/spool/data collectorMainAgent.channels.c2.capacity = 200000000 collectorMainAgent.channels.c2.transactionCapacity=6000 collectorMainAgent.channels.c2.checkpointInterval=60000 # collectorMainAgent hdfsSink collectorMainAgent.sinks.k2. type = hdfs collectorMainAgent.sinks.k2.channel = c2 collectorMainAgent.sinks.k2.hdfs.path = hdfs: //db-cdh-cluster/flume %{ dir } collectorMainAgent.sinks.k2.hdfs.filePrefix =k2_%{ file } collectorMainAgent.sinks.k2.hdfs.inUsePrefix =_ collectorMainAgent.sinks.k2.hdfs.inUseSuffix =.tmp collectorMainAgent.sinks.k2.hdfs.rollSize = 0 collectorMainAgent.sinks.k2.hdfs.rollCount = 0 collectorMainAgent.sinks.k2.hdfs.rollInterval = 240 collectorMainAgent.sinks.k2.hdfs.writeFormat = Text collectorMainAgent.sinks.k2.hdfs.fileType = DataStream collectorMainAgent.sinks.k2.hdfs.batchSize = 6000 collectorMainAgent.sinks.k2.hdfs.callTimeout = 60000 collectorMainAgent.sinks.k1. type = hdfs collectorMainAgent.sinks.k1.channel = c2 collectorMainAgent.sinks.k1.hdfs.path = hdfs: //db-cdh-cluster/flume %{ dir } collectorMainAgent.sinks.k1.hdfs.filePrefix =k1_%{ file } collectorMainAgent.sinks.k1.hdfs.inUsePrefix =_ collectorMainAgent.sinks.k1.hdfs.inUseSuffix =.tmp collectorMainAgent.sinks.k1.hdfs.rollSize = 0 collectorMainAgent.sinks.k1.hdfs.rollCount = 0 collectorMainAgent.sinks.k1.hdfs.rollInterval = 240 collectorMainAgent.sinks.k1.hdfs.writeFormat = Text collectorMainAgent.sinks.k1.hdfs.fileType = DataStream collectorMainAgent.sinks.k1.hdfs.batchSize = 6000 collectorMainAgent.sinks.k1.hdfs.callTimeout = 60000 |
../bin/flume-ng agent --conf ../conf/ -f flume_Consolidation.conf -n collectorMainAgent -Dflume.root.logger=DEBUG,console
4、可能遇到的問題：
4.1 OOM?問題：
?| 1 2 3 4 5 | flume 報錯： java.lang.OutOfMemoryError: GC overhead limit exceeded 或者： java.lang.OutOfMemoryError: Java heap space Exception in thread "SinkRunner-PollingRunner-DefaultSinkProcessor" java.lang.OutOfMemoryError: Java heap space |
Flume 啟動時的最大堆內存大小默認是 20M，線上環境很容易 OOM，因此需要你在 flume-env.sh?中添加 JVM 啟動參數:?
| 1 | JAVA_OPTS= "-Xms8192m -Xmx8192m -Xss256k -Xmn2g -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:-UseGCOverheadLimit" |
然后在啟動 agent 的時候一定要帶上?-c conf 選項，否則?flume-env.sh 里配置的環境變量不會被加載生效。
具體參見：
http://stackoverflow.com/questions/1393486/error-java-lang-outofmemoryerror-gc-overhead-limit-exceeded
http://marc.info/?l=flume-user&m=138933303305433&w=2
4.2 JDK 版本不兼容問題：
?
?| 1 2 3 4 5 6 7 | 2014 - 07 - 07 14 : 44 : 17 , 902 (agent-shutdown-hook) [WARN - org.apache.flume.sink.hdfs.HDFSEventSink.stop(HDFSEventSink.java: 504 )] Exception while closing hdfs: //192.168.1.111:8020/flumeTest/FlumeData. Exception follows. java.lang.UnsupportedOperationException: This is supposed to be overridden by subclasses. ???????? at com.google.protobuf.GeneratedMessage.getUnknownFields(GeneratedMessage.java: 180 ) ???????? at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$GetFileInfoRequestProto.getSerializedSize(ClientNamenodeProtocolProtos.java: 30108 ) ???????? at com.google.protobuf.AbstractMessageLite.toByteString(AbstractMessageLite.java: 49 ) ???????? at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.constructRpcRequest(ProtobufRpcEngine.java: 149 ) ???????? at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java: 193 ) |
把你的 jdk7 換成 jdk6 試試。
4.3 小文件寫入 HDFS 延時的問題
其實上面 3.2 中已有說明，flume 的 sink 已經實現了幾種最主要的持久化觸發器：
比如按大小、按間隔時間、按消息條數等等，針對你的文件過小遲遲沒法寫入 HDFS 持久化的問題，
那是因為你此時還沒有滿足持久化的條件，比如你的行數還沒有達到配置的閾值或者大小還沒達到等等，
可以針對上面 3.2 小節的配置微調下，例如：
?
?| 1 | agent1.sinks.log-sink1.hdfs.rollInterval = 20 |
下面貼一些常見的持久化觸發器：
?
?| 1 2 3 4 5 6 7 8 9 10 11 12 13 | # Number of seconds to wait before rolling current file (in 600 seconds) agent.sinks.sink.hdfs.rollInterval=600 # File size to trigger roll, in bytes (256Mb) agent.sinks.sink.hdfs.rollSize = 268435456 # never roll based on number of events agent.sinks.sink.hdfs.rollCount = 0 # Timeout after which inactive files get closed (in seconds) agent.sinks.sink.hdfs.idleTimeout = 3600 agent.sinks.HDFS.hdfs.batchSize = 1000 |
http://stackoverflow.com/questions/20638498/flume-not-writing-to-hdfs-unless-killed
注意：對于 HDFS 來說應當竭力避免小文件問題，所以請慎重對待你配置的持久化觸發機制。
4.4 數據重復寫入、丟失問題
Flume的HDFSsink在數據寫入/讀出Channel時，都有Transcation的保證。當Transaction失敗時，會回滾，然后重試。但由于HDFS不可修改文件的內容，假設有1萬行數據要寫入HDFS，而在寫入5000行時，網絡出現問題導致寫入失敗，Transaction回滾，然后重寫這10000條記錄成功，就會導致第一次寫入的5000行重復。這些問題是 HDFS 文件系統設計上的特性缺陷，并不能通過簡單的Bugfix來解決。我們只能關閉批量寫入，單條事務保證，或者啟用監控策略，兩端對數。
Memory和exec的方式可能會有數據丟失，file 是 end to end 的可靠性保證的，但是性能較前兩者要差。
end to end、store on failure 方式 ACK 確認時間設置過短（特別是高峰時間）也有可能引發數據的重復寫入。
4.5 tail 斷點續傳的問題：
可以在 tail 傳的時候記錄行號，下次再傳的時候，取上次記錄的位置開始傳輸，類似：
?
?| 1 | agent1.sources.avro-source1. command = /usr/local/bin/tail? -n +$( tail -n1 /home/storm/tmp/n ) --max-unchanged-stats=600 -F? /home/storm/tmp/id .txt | awk 'ARNGIND==1{i=$0;next}{i++; if($0~/文件已截斷/)i=0; print i >> "/home/storm/tmp/n";print $1"---"i}' /home/storm/tmp/n - |
（1）文件被 rotation 的時候，需要同步更新你的斷點記錄“指針”，
（2）需要按文件名來追蹤文件，
（3）flume 掛掉后需要累加斷點續傳“指針”
（4）flume 掛掉后，如果恰好文件被 rotation，那么會有丟數據的風險，
? ? ? ?只能監控盡快拉起或者加邏輯判斷文件大小重置指針。
（5）tail 注意你的版本，請更新?coreutils 包到最新。
4.6 在 Flume 中如何修改、丟棄、按預定義規則分類存儲數據？
這里你需要利用 Flume 提供的攔截器（Interceptor）機制來滿足上述的需求了，具體請參考下面幾個鏈接：
（1）Flume-NG源碼閱讀之Interceptor(原創) ?
http://www.cnblogs.com/lxf20061900/p/3664602.html
（2）Flume-NG自定義攔截器
http://sep10.com/posts/2014/04/15/flume-interceptor/
（3）Flume-ng生產環境實踐（四）實現log格式化interceptor
http://blog.csdn.net/rjhym/article/details/8450728
（4）flume-ng如何根據源文件名輸出到HDFS文件名
http://abloz.com/2013/02/19/flume-ng-output-according-to-the-source-file-name-to-the-hdfs-file-name.html
5、Refer：
（1）scribe、chukwa、kafka、flume日志系統對比 ?
http://www.ttlsa.com/log-system/scribe-chukwa-kafka-flume-log-system-contrast/
（2）關于Flume-ng那些事 ?http://www.ttlsa.com/?s=flume
? ? ? ? ?關于Flume-ng那些事（三）：常見架構測試 ?http://www.ttlsa.com/log-system/about-flume-ng-3/
（3）Flume 1.4.0 User Guide
http://archive.cloudera.com/cdh4/cdh/4/flume-ng-1.4.0-cdh4.7.0/FlumeUserGuide.html
（4）flume日志采集 ?http://blog.csdn.net/sunmeng_007/article/details/9762507
（5）Flume-NG + HDFS + HIVE 日志收集分析
http://eyelublog.wordpress.com/2013/01/13/flume-ng-hdfs-hive-%E6%97%A5%E5%BF%97%E6%94%B6%E9%9B%86%E5%88%86%E6%9E%90/
（6）【Twitter Storm系列】flume-ng+Kafka+Storm+HDFS 實時系統搭建
http://blog.csdn.net/weijonathan/article/details/18301321
（7）Flume-NG + HDFS + PIG 日志收集分析
http://hi.baidu.com/life_to_you/item/a98e2ec3367486dbef183b5e
flume 示例一收集tomcat日志 ?http://my.oschina.net/88sys/blog/71529
flume-ng 多節點集群示例 ?http://my.oschina.net/u/1401580/blog/204052
試用flume-ng 1.1??http://heipark.iteye.com/blog/1617995
（8）Flafka: Apache Flume Meets Apache Kafka for Event Processing
http://blog.cloudera.com/blog/2014/11/flafka-apache-flume-meets-apache-kafka-for-event-processing/
（9）Flume-ng的原理和使用
http://segmentfault.com/blog/javachen/1190000002532284
（10）基于Flume的美團日志收集系統(一)架構和設計
http://tech.meituan.com/mt-log-system-arch.html
（11）基于Flume的美團日志收集系統(二)改進和優化
http://tech.meituan.com/mt-log-system-optimization.html
（12）How-to: Do Real-Time Log Analytics with Apache Kafka, Cloudera Search, and Hue
http://blog.cloudera.com/blog/2015/02/how-to-do-real-time-log-analytics-with-apache-kafka-cloudera-search-and-hue/
（13）Real-time analytics in Apache Flume - Part 1
http://jameskinley.tumblr.com/post/57704266739/real-time-analytics-in-apache-flume-part-1
?
?
http://my.oschina.net/leejun2005/blog/288136
轉載于:https://my.oschina.net/xiaominmin/blog/1597628
總結
以上是生活随笔為你收集整理的Flume NG 简介及配置实战的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 段表和页表【转】
- 下一篇： 利用btrace在线监控java程序状态