最近在做flume的實(shí)時(shí)日志收集包归，用flume默認(rèn)的配置后，發(fā)現(xiàn)不是完全實(shí)時(shí)的铅歼，于是看了一下公壤，原來(lái)是memeryChannel的transactionCapacity在作怪换可，因?yàn)樗J(rèn)是100，也就是說(shuō)收集端的sink會(huì)在收集到了100條以后再去提交事務(wù)（即發(fā)送到下一個(gè)目的地）厦幅，于是我修改了transactionCapacity到10沾鳄，想看看是不是會(huì)更加實(shí)時(shí)一點(diǎn)，結(jié)果發(fā)現(xiàn)收集日志的agent啟動(dòng)的時(shí)候報(bào)錯(cuò)了确憨。
16/04/29 09:36:15 ERROR sink.AbstractRpcSink: Rpc Sink avro-sink: Unable to get event from channel memoryChannel. Exception follows.
org.apache.flume.ChannelException: Take list for MemoryTransaction, capacity 10 full, consider committing more frequently, increasing capacity, or increasing thread count
at org.apache.flume.channel.MemoryChannel$MemoryTransaction.doTake(MemoryChannel.[Java](http://lib.csdn.net/base/javase "Java SE知識(shí)庫(kù)"):96)
at org.apache.flume.channel.BasicTransactionSemantics.take(BasicTransactionSemantics.java:113)
at org.apache.flume.channel.BasicChannelSemantics.take(BasicChannelSemantics.java:95)
at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:354)
at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
at java.lang.Thread.run(Thread.java:745)
于是很納悶译荞，為什么默認(rèn)值100可以，而設(shè)置10就會(huì)說(shuō)小了呢休弃，于是查閱資料吞歼，發(fā)現(xiàn)原來(lái)是sink的batchsize參數(shù)在作怪，下面塔猾，我就來(lái)理一理這個(gè)來(lái)龍去脈浆熔，這個(gè)sink的batchsize是什么意思呢，就是sink會(huì)一次從channel中取多少個(gè)event去發(fā)送桥帆，而這個(gè)發(fā)送是要最終以事務(wù)的形式去發(fā)送的医增，因此這個(gè)batchsize的event會(huì)傳送到一個(gè)事務(wù)的緩存隊(duì)列中（takeList），這是一個(gè)雙向隊(duì)列老虫，這個(gè)隊(duì)列可以在事務(wù)失敗時(shí)進(jìn)行回滾（也就是把取出來(lái)的數(shù)據(jù)吐memeryChannel的queue中）叶骨，它的初始大小就是transactionCapacity定義的大小，源碼中有： takeList = new LinkedBlockingDeque<Event>(transCapacity); 源碼來(lái)自https://segmentfault.com/a/1190000003586635的分享祈匙。
再看這個(gè)錯(cuò)誤拋出的地方：
if(takeList.remainingCapacity() == 0) {
throw new ChannelException("Take list for MemoryTransaction, capacity " +
takeList.size() + " full, consider committing more frequently, " +
"increasing capacity, or increasing thread count");
}
在上面的情況中忽刽，sink一次取100個(gè)events，塞到takelist中夺欲，在塞了10個(gè)后跪帝，就會(huì)引發(fā)上述異常，因此些阅，這個(gè)錯(cuò)誤的解決辦法就是：在sink中伞剑，channel的transactionCapacity參數(shù)不能小于sink的batchsize。
=================================================================================
Flume-ng出現(xiàn)HDFS IO error,Callable timed out異常
摘自http://blog.csdn.net/wsscy2004/article/details/22179361
目前解決方案：
- 根據(jù)http://blog.csdn.net/yangbutao/article/details/8845025市埋，修改flume到hdfs客戶端超時(shí)時(shí)間黎泣。
- 修改hdfs.callTimeout從默認(rèn)的10秒改成:40秒
三、記Flume-NG一些注意事項(xiàng)
一缤谎、關(guān)于Source：
1抒倚、spool-source：適合靜態(tài)文件，即文件本身不是動(dòng)態(tài)變化的坷澡；
2托呕、avro source可以適當(dāng)提高線程數(shù)量來(lái)提高此source性能；
3、ThriftSource在使用時(shí)有個(gè)問(wèn)題需要注意项郊，使用批量操作時(shí)出現(xiàn)異常并不會(huì)打印異常內(nèi)容而是"Thrift source %s could not append events to the channel."馅扣，這是因?yàn)樵创a中在出現(xiàn)異常時(shí)，它并未捕獲異常而是獲取組件名稱呆抑，這是源碼中的一個(gè)bug岂嗓，也可以說(shuō)明thrift很少有人用汁展，否則這個(gè)問(wèn)題也不會(huì)存在在很多版本中鹊碍；
4、如果一個(gè)source對(duì)應(yīng)多個(gè)channel食绿，默認(rèn)就是每個(gè)channel是同樣的一份數(shù)據(jù)侈咕，會(huì)把這批數(shù)據(jù)復(fù)制N份發(fā)送到N個(gè)channel中，所以如果某個(gè)channel滿了會(huì)影響整體的速度的哦器紧；
5耀销、ExecSource官方文檔已經(jīng)說(shuō)明是異步的，可能會(huì)丟數(shù)據(jù)哦铲汪，盡量使用tail -F熊尉，注意是大寫(xiě)的；
二掌腰、關(guān)于Channel：
1狰住、采集節(jié)點(diǎn)建議使用新的復(fù)合類(lèi)型的SpillableMemoryChannel，匯總節(jié)點(diǎn)建議采用memory channel齿梁，具體還要看實(shí)際的數(shù)據(jù)量催植，一般每分鐘數(shù)據(jù)量超過(guò)120MB大小的flume agent都建議用memory channel(自己測(cè)的file channel處理速率大概是2M/s，不同機(jī)器勺择、不同環(huán)境可能不同创南，這里只提供參考)，因?yàn)橐坏┐薬gent的channel出現(xiàn)溢出情況省核，將會(huì)導(dǎo)致大多數(shù)時(shí)間處于file channel(SpillableMemoryChannel本身是file channel的一個(gè)子類(lèi)稿辙，而且復(fù)合channel會(huì)保證一定的event的順序的使得讀完內(nèi)存中的數(shù)據(jù)后，再需要把溢出的拿走气忠，可能這時(shí)內(nèi)存已滿又會(huì)溢出邓深。。笔刹。)芥备，性能大大降低，匯總一旦成為這樣后果可想而知舌菜；
2萌壳、調(diào)整memory 占用物理內(nèi)存空間，需要兩個(gè)參數(shù)byteCapacityBufferPercentage(默認(rèn)是20)和byteCapacity(默認(rèn)是JVM最大可用內(nèi)存的0.8)來(lái)控制，計(jì)算公式是：byteCapacity = (int)((context.getLong("byteCapacity", defaultByteCapacity).longValue() * (1 - byteCapacityBufferPercentage * .01 )) /byteCapacitySlotSize)袱瓮，很明顯可以調(diào)節(jié)這兩個(gè)參數(shù)來(lái)控制缤骨，至于byteCapacitySlotSize默認(rèn)是100，將物理內(nèi)存轉(zhuǎn)換成槽(slot)數(shù)尺借，這樣易于管理绊起，但是可能會(huì)浪費(fèi)空間，至少我是這樣想的燎斩。虱歪。。栅表；
3笋鄙、還有一個(gè)有用的參數(shù)"keep-alive"這個(gè)參數(shù)用來(lái)控制channel滿時(shí)影響source的發(fā)送，channel空時(shí)影響sink的消費(fèi)怪瓶，就是等待時(shí)間萧落，默認(rèn)是3s，超過(guò)這個(gè)時(shí)間就甩異常洗贰，一般不需配置找岖，但是有些情況很有用，比如你得場(chǎng)景是每分鐘開(kāi)頭集中發(fā)一次數(shù)據(jù)敛滋，這時(shí)每分鐘的開(kāi)頭量可能比較大许布，后面會(huì)越來(lái)越小，這時(shí)你可以調(diào)大這個(gè)參數(shù)矛缨，不至于出現(xiàn)channel滿了得情況爹脾；
三、關(guān)于Sink：
1箕昭、avro sink的batch-size可以設(shè)置大一點(diǎn)灵妨，默認(rèn)是100，增大會(huì)減少RPC次數(shù)落竹，提高性能泌霍；
2、內(nèi)置hdfs sink的解析時(shí)間戳來(lái)設(shè)置目錄或者文件前綴非常損耗性能述召，因?yàn)槭腔谡齽t來(lái)匹配的朱转，可以通過(guò)修改源碼來(lái)替換解析時(shí)間功能來(lái)極大提升性能，稍后我會(huì)寫(xiě)一篇文章來(lái)專門(mén)說(shuō)明這個(gè)問(wèn)題积暖；
3藤为、RollingFileSink文件名不能自定義，而且不能定時(shí)滾動(dòng)文件夺刑，只能按時(shí)間間隔滾動(dòng)缅疟，可以自己定義sink分别，來(lái)做定時(shí)寫(xiě)文件；
4存淫、hdfs sink的文件名中的時(shí)間戳部分不能省去耘斩，可增加前綴、后綴以及正在寫(xiě)的文件的前后綴等信息桅咆；"hdfs.idleTimeout"這個(gè)參數(shù)很有意義括授，指的是正在寫(xiě)的hdfs文件多長(zhǎng)時(shí)間不更新就關(guān)閉文件，建議都配置上岩饼，比如你設(shè)置了解析時(shí)間戳存不同的目錄荚虚、文件名，而且rollInterval=0忌愚、rollCount=0曲管、rollSize=1000000却邓，如果這個(gè)時(shí)間內(nèi)的數(shù)據(jù)量達(dá)不到rollSize的要求而且后續(xù)的寫(xiě)入新的文件中了硕糊，就是一直打開(kāi)，類(lèi)似情景不注意的話可能很多腊徙；"hdfs.callTimeout"這個(gè)參數(shù)指的是每個(gè)hdfs操作(讀简十、寫(xiě)、打開(kāi)撬腾、關(guān)閉等)規(guī)定的最長(zhǎng)操作時(shí)間螟蝙，每個(gè)操作都會(huì)放入"hdfs.threadsPoolSize"指定的線程池中得一個(gè)線程來(lái)操作；
5民傻、關(guān)于HBase sink(非異步hbase sink：AsyncHBaseSink)胰默，rowkey不能自定義，而且一個(gè)serializer只能寫(xiě)一列漓踢，一個(gè)serializer按正則匹配多個(gè)列牵署，性能可能存在問(wèn)題，建議自己根據(jù)需求寫(xiě)一個(gè)hbase sink喧半；
6奴迅、avro sink可以配置failover和loadbalance，所用的組件和sinkgroup中的是一樣的挺据，而且也可以在此配置壓縮選項(xiàng)取具，需要在avro source中配置解壓縮；
四扁耐、關(guān)于SinkGroup：
1暇检、不管是loadbalance或者是failover的多個(gè)sink需要共用一個(gè)channel；
2婉称、loadbalance的多個(gè)sink如果都是直接輸出到同一種設(shè)備块仆，比如都是hdfs心墅，性能并不會(huì)有明顯增加，因?yàn)閟inkgroup是單線程的它的process方法會(huì)輪流調(diào)用每個(gè)sink去channel中take數(shù)據(jù)榨乎，并確保處理正確怎燥，使得是順序操作的，但是如果是發(fā)送到下一級(jí)的flume agent就不一樣了蜜暑，take操作是順序的铐姚，但是下一級(jí)agent的寫(xiě)入操作是并行的，所以肯定是快的肛捍；
3隐绵、其實(shí)用loadbalance在一定意義上可以起到failover的作用，生產(chǎn)環(huán)境量大建議loadbalance拙毫；
五依许、關(guān)于監(jiān)控monitor：
1、監(jiān)控我這邊做得還是比較少的缀蹄，但是目前已知的有以下幾種吧：cloudera manager（前提是你得安裝CDH版本）峭跳、ganglia(這個(gè)天生就是支持的)、http(其實(shí)就是將統(tǒng)計(jì)信息jmx信息缺前，封裝成json串蛀醉，使用jetty展示在瀏覽器中而已)、再一個(gè)就是自己實(shí)現(xiàn)收集監(jiān)控信息衅码，自己做(可以收集http的信息或者自己實(shí)現(xiàn)相應(yīng)的接口實(shí)現(xiàn)自己的邏輯拯刁，具體可以參考我以前的博客)；
2逝段、簡(jiǎn)單說(shuō)一下cloudera manager這種監(jiān)控垛玻，最近在使用，確實(shí)很強(qiáng)大奶躯，可以查看實(shí)時(shí)的channel進(jìn)出數(shù)據(jù)速率帚桩、channel實(shí)時(shí)容量、sink的出速率巫糙、source的入速率等等朗儒，圖形化的東西確實(shí)很豐富很直觀，可以提供很多flume agent整體運(yùn)行情況的信息和潛在的一些信息参淹；
六醉锄、關(guān)于flume啟動(dòng)：
1、flume組件啟動(dòng)順序：channels——>sinks——>sources浙值，關(guān)閉順序：sources——>sinks——>channels恳不；
2、自動(dòng)加載配置文件功能开呐，會(huì)先關(guān)閉所有組件烟勋，再重啟所有組件规求；
3、關(guān)于AbstractConfigurationProvider中的Map<Class<? extends Channel>, Map<String, Channel>> channelCache這個(gè)對(duì)象卵惦，始終存儲(chǔ)著agent中得所有channel對(duì)象阻肿，因?yàn)樵趧?dòng)態(tài)加載時(shí)，channel中可能還有未消費(fèi)完的數(shù)據(jù)沮尿，但是需要對(duì)channel重新配置丛塌，所以用以來(lái)緩存channel對(duì)象的所有數(shù)據(jù)及配置信息；
4畜疾、通過(guò)在啟動(dòng)命令中添加 "no-reload-conf"參數(shù)為true來(lái)取消自動(dòng)加載配置文件功能赴邻；
七、關(guān)于interceptor：
八啡捶、關(guān)于自定義組件：sink姥敛、source、channel：
1瞎暑、channel不建議自定義哦彤敛，這個(gè)要求比較高，其他倆都是框架式的開(kāi)發(fā)金顿，往指定的方法填充自己配置臊泌、啟動(dòng)鲤桥、關(guān)閉揍拆、業(yè)務(wù)邏輯即可，以后有機(jī)會(huì)單獨(dú)寫(xiě)一篇文章來(lái)介紹茶凳；
2嫂拴、關(guān)于自定義組件請(qǐng)相信github，上面好多好多好多贮喧，可以直接用的自定義組件....筒狠；
九、關(guān)于Flume-NG集群網(wǎng)絡(luò)拓?fù)浞桨福?/p>
1箱沦、在每臺(tái)采集節(jié)點(diǎn)上部署一個(gè)flume agent辩恼，然后做一到多個(gè)匯總flume agent(loadbalance)，采集只負(fù)責(zé)收集數(shù)據(jù)發(fā)往匯總谓形，匯總可以寫(xiě)HDFS灶伊、HBase、spark寒跳、本地文件聘萨、kafka等等，這樣一般修改會(huì)只在匯總童太，agent少米辐，維護(hù)工作少胸完；
2、采集節(jié)點(diǎn)沒(méi)有部署flume agent翘贮，可能發(fā)往mongo赊窥、redis等，這時(shí)你需要自定義source或者使用sdk來(lái)將其中的數(shù)據(jù)取出并發(fā)往flume agent狸页，這樣agent就又可以充當(dāng)“采集節(jié)點(diǎn)”或者匯總節(jié)點(diǎn)了誓琼，但是這樣在前面相當(dāng)于加了一層控制，就又多了一層風(fēng)險(xiǎn)肴捉；
3腹侣、由于能力有限，其它未知齿穗，上面兩種傲隶，第一種好些，這里看看美團(tuán)的架構(gòu)———— 傳送門(mén)窃页；
四跺株、
java.lang.NullPointerException: Expected timestamp in the Flume event headers, but it was null
sink是hdfs，然后使用目錄自動(dòng)生成功能脖卖。出現(xiàn)如題的錯(cuò)誤乒省，看官網(wǎng)文檔說(shuō)的是需要在每個(gè)文件記錄行的開(kāi)頭需要有時(shí)間戳，但是時(shí)間戳的格式可能比較難調(diào)節(jié)畦木，所以亦可設(shè)置hdfs.useLocalTimeStamp這個(gè)參數(shù)袖扛，比如以每個(gè)小時(shí)作為一個(gè)文件夾，那么配置應(yīng)該是這樣：
a1.sinks.k1.hdfs.path = hdfs://ubuntu:9000/flume/events/%y-%m-%d/%H
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 1
a1.sinks.k1.hdfs.roundUnit = hour
a1.sinks.k1.hdfs.useLocalTimeStamp = true
或者修改hdfs.timeZone這個(gè)參數(shù)使之可以和我們上傳的log文件的日期格式一樣應(yīng)該就可以了十籍，沒(méi)有測(cè)試過(guò)蛆封。
五、flume學(xué)習(xí)（三）：flume將log4j日志數(shù)據(jù)寫(xiě)入到hdfs
本次我們把log4j的日志直接采集輸出到hdfs中去勾栗。需要修改flume.conf中sink的配置：
1. tier1.sources=source1
2. tier1.channels=channel1
3. tier1.sinks=sink1
5. tier1.sources.source1.type=avro
6. tier1.sources.source1.bind=0.0.0.0
7. tier1.sources.source1.port=44444
8. tier1.sources.source1.channels=channel1
10. tier1.channels.channel1.type=memory
11. tier1.channels.channel1.capacity=10000
12. tier1.channels.channel1.transactionCapacity=1000
13. tier1.channels.channel1.keep-alive=30
15. tier1.sinks.sink1.type=hdfs
16. tier1.sinks.sink1.channel=channel1
17. tier1.sinks.sink1.hdfs.path=hdfs://master68:8020/flume/events
18. tier1.sinks.sink1.hdfs.fileType=DataStream
19. tier1.sinks.sink1.hdfs.writeFormat=Text
20. tier1.sinks.sink1.hdfs.rollInterval=0
21. tier1.sinks.sink1.hdfs.rollSize=10240
22. tier1.sinks.sink1.hdfs.rollCount=0
23. tier1.sinks.sink1.hdfs.idleTimeout=60
六惨篱、【Flume】【源碼分析】flume中sink到hdfs，文件系統(tǒng)頻繁產(chǎn)生文件围俘，文件滾動(dòng)配置不起作用砸讳？
解決方案
原因：flume配置問(wèn)題或者說(shuō)代碼問(wèn)題 ，文件滾動(dòng)的判斷條件存在漏洞
增加配置參數(shù) 界牡，即可按照參數(shù)滾動(dòng)文件
flume1.sinks.sink1.hdfs.minBlockReplicas=1
參考：http://blog.csdn.net/simonchi/article/details/43231891
七簿寂、查看最終配置
（來(lái)源http://www.it610.com/article/2107322.htm）
最終配置文件示例
# flume1 which ones we want to activate.
flume1.channels = ch1
flume1.sources = src1
flume1.sinks = sink1
# Define a memory channel called ch1 on flume1
flume1.channels.ch1.type = memory
flume1.channels.ch1.capacity = 100000
flume1.channels.ch1.transactionCapacity = 1000
flume1.channels.ch1.keep-alive = 30
# Define an Avro source called src1 on flume1 and tell it
# to bind to 0.0.0.0:8888\. Connect it to channel ch1.
flume1.sources.src1.channels = ch1
flume1.sources.src1.type = avro
flume1.sources.src1.bind = 0.0.0.0
flume1.sources.src1.port = 8888
flume1.sources.src1.threads = 5
flume1.sinks.sink1.type = hdfs
flume1.sinks.sink1.channel = ch1
flume1.sinks.sink1.hdfs.path =hdfs:``//master:9000/ysg/flume/ysg/%Y%m
flume1.sinks.sink1.hdfs.filePrefix = ysg
flume1.sinks.sink1.hdfs.fileSuffix = .log
flume1.sinks.sink1.hdfs.inUseSuffix = .tmp
flume1.sinks.sink1.hdfs.maxOpenFiles = 5000
flume1.sinks.sink1.hdfs.batchSize= 1
flume1.sinks.sink1.hdfs.fileType = DataStream
flume1.sinks.sink1.hdfs.writeFormat =Text
#flume1.sinks.sink1.hdfs.rollSize =64*1024*1024
flume1.sinks.sink1.hdfs.rollSize = 67108864
flume1.sinks.sink1.hdfs.rollCount = 0
flume1.sinks.sink1.hdfs.rollInterval = 0
flume1.sinks.sink1.hdfs.minBlockReplicas=1
flume1.sinks.sink1.hdfs.useLocalTimeStamp = ``true
flume1.sinks.sink1.hdfs.connect-timeout=80000
flume1.sinks.sink1.hdfs.callTimeout=120000
flume1.sinks.sink1.hdfs.idleTimeout = 60
八、 3.1 基礎(chǔ)參數(shù)調(diào)優(yōu)經(jīng)驗(yàn) --去掉 每寫(xiě)一行在行尾添加一個(gè)換行符 情況
- HdfsSink中默認(rèn)的serializer會(huì)每寫(xiě)一行在行尾添加一個(gè)換行符欢揖，我們?nèi)罩颈旧韼в袚Q行符陶耍，這樣會(huì)導(dǎo)致每條日志后面多一個(gè)空行，修改配置不要自動(dòng)添加換行符她混；
lc.sinks.sink_hdfs.serializer.appendNewline = ``false
調(diào)大MemoryChannel的capacity烈钞，盡量利用MemoryChannel快速的處理能力泊碑；
調(diào)大HdfsSink的batchSize，增加吞吐量毯欣，減少hdfs的flush次數(shù)馒过；
適當(dāng)調(diào)大HdfsSink的callTimeout，避免不必要的超時(shí)錯(cuò)誤酗钞；
