前言
所有電商企業(yè)在一年一度的雙11都要迎來大促與大考登刺，我司也不例外（所以最近真是前所未有的忙亂）告嘲。前段時(shí)間在配合執(zhí)行全鏈路壓測(cè)的過程中赏殃，發(fā)現(xiàn)平時(shí)不太關(guān)注的Flume配置可能存在瓶頸。Flume在筆者負(fù)責(zé)的實(shí)時(shí)計(jì)算平臺(tái)里用于收集所有后端訪問日志和埋點(diǎn)日志郊酒，其效率和穩(wěn)定性比較重要遇绞。除了及時(shí)擴(kuò)容之外键袱，也有必要對(duì)Flume進(jìn)行調(diào)優(yōu)燎窘。今天在百忙之中擠出一點(diǎn)時(shí)間來寫寫。
Flume系統(tǒng)以一個(gè)或多個(gè)Flume-NG Agent的形式部署蹄咖，一個(gè)Agent對(duì)應(yīng)一個(gè)JVM進(jìn)程褐健，并且由三個(gè)部分組成：Source、Channel和Sink澜汤，示意圖如下蚜迅。
Source
Flume有3種能夠監(jiān)聽文件的Source，分別是Exec Source（配合tail -f命令）俊抵、Spooling Directory Source和Taildir Source谁不。Taildir Source顯然是最好用的，在我們的實(shí)踐中徽诲，需要注意的參數(shù)列舉如下刹帕。
- filegroups
如果需要監(jiān)聽的日志文件較多，應(yīng)該將它們分散在不同的目錄下谎替，并配置多個(gè)filegroup來并行讀取偷溺。注意日志文件的正則表達(dá)式要寫好，防止日志滾動(dòng)重命名時(shí)仍然符合正則表達(dá)式造成重復(fù)钱贯。示例：
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1 f2
a1.sources.r1.filegroups.f1 = /data/logs/ng1/access.log
a1.sources.r1.headers.f1.headerKey1 = ng1
a1.sources.r1.filegroups.f2 = /data/logs/ng2/.*log
a1.sources.r1.headers.f2.headerKey1 = ng2
- batchSize
該參數(shù)控制向Channel發(fā)送數(shù)據(jù)的批次大小挫掏，默認(rèn)值為100。如果日志流量很大秩命，適當(dāng)增加此值可以增大Source的吞吐量尉共，但是不能超過Channel的capacity和transactionCapacity的限制（后文再說）。示例：
a1.sources.r1.batchSize = 1000
- maxBatchCount
該參數(shù)控制從同一個(gè)文件中連續(xù)讀取的最大批次數(shù)量弃锐，默認(rèn)不限制袄友。如果Flume同時(shí)監(jiān)聽多個(gè)文件，并且其中某個(gè)文件的寫入速度遠(yuǎn)快于其他文件拿愧，那么其他文件有可能幾乎無法被讀取杠河，所以強(qiáng)烈建議設(shè)定此參數(shù)。示例：
a1.sources.r1.maxBatchCount = 100
- writePosInterval
該參數(shù)控制向記錄讀取位置的JSON文件（由positionFile參數(shù)指定）寫入inode和偏移量的頻率，默認(rèn)為3000ms券敌。當(dāng)Agent重新啟動(dòng)時(shí)唾戚，會(huì)從JSON文件中獲取最近記錄的偏移量開始讀取。也就是說待诅，適當(dāng)降低writePosInterval可以減少Agent重啟導(dǎo)致的重復(fù)讀取的數(shù)據(jù)量叹坦。
a1.sources.r1.writePosInterval = 1000
Channel
Flume內(nèi)置了多種Channel的實(shí)現(xiàn)，比較常用的有Memory Channel卑雁、File Channel募书、JDBC Channel、Kafka Channel等测蹲。我們的選擇主要針對(duì)Memory Channel和File Channel兩種莹捡，對(duì)比一下：
- Memory Channel將staging事件數(shù)據(jù)存儲(chǔ)在Agent堆內(nèi)存中，F(xiàn)ile Channel則將它們存儲(chǔ)在指定的文件中扣甲；
- 如果Agent失敗篮赢，Memory Channel會(huì)丟失所有緩存的staging事件，F(xiàn)ile Channel則可以通過額外記錄的checkpoint信息恢復(fù)琉挖，保證斷點(diǎn)續(xù)傳启泣；
- Memory Channel能夠容納的數(shù)據(jù)量受堆內(nèi)存的影響，而File Channel不受此限制示辈。
鑒于我們下游業(yè)務(wù)的主要痛點(diǎn)在吞吐量與實(shí)時(shí)性寥茫，且可以容忍數(shù)據(jù)少量丟失，日志服務(wù)器的磁盤壓力也已經(jīng)比較大了矾麻，故Memory Channel更加合適纱耻。需要注意的參數(shù)如下。
- capacity射富、transactionCapacity
這兩個(gè)參數(shù)分別代表Channel最多能容納的事件數(shù)目膝迎，以及每個(gè)事務(wù)（即Source的一次put或者Sink的一次take）能夠包含的最多事件數(shù)目。顯然胰耗，必須滿足batchSize <= transactionCapacity <= capacity的關(guān)系限次。適當(dāng)調(diào)大capacity和transactionCapacity可以使得Channel的吞吐量增高，且能夠保證不會(huì)出現(xiàn)The channel is full or unexpected failure的異常柴灯。示例：
a1.channels.c1.type = memory
a1.channels.c1.transactionCapacity = 5000
a1.channels.c1.capacity = 10000
byteCapacity
該參數(shù)代表Memory Channel中緩存的事件消息的最大總大小卖漫，以字節(jié)為單位，默認(rèn)是Flume Agent最大堆內(nèi)存的80%赠群。此值不建議更改為固定的羊始，而是建議通過改變Agent的JVM參數(shù)來影響，后面再提查描。byteCapacityBufferPercentage
Memory Channel中緩存的事件消息頭占byteCapacity的比例突委，默認(rèn)是20%柏卤。如果事件的header信息很少，可以適當(dāng)減性扔汀（我們沒有更改）缘缚。keep-alive
向Channel中put或take一條事件的超時(shí)時(shí)間，默認(rèn)為3秒敌蚜，對(duì)于Memory Channel一般不用更改桥滨。如果業(yè)務(wù)數(shù)據(jù)是由很多突發(fā)流量組成（也就是說Channel經(jīng)常處于時(shí)滿時(shí)空的狀態(tài)），那么建議適當(dāng)調(diào)大弛车。示例：
a1.channels.c1.keep-alive = 15
當(dāng)然File Channel也很常用齐媒，其參數(shù)就不再贅述，看官可參考官方文檔纷跛。
Sink
我們實(shí)時(shí)數(shù)倉(cāng)接入層的起點(diǎn)是Kafka喻括，自然要利用Kafka Sink。需要注意的參數(shù)列舉如下忽舟。
kafka.flumeBatchSize
從Channel取出數(shù)據(jù)并發(fā)送到Kafka的批次大小双妨，與Source的batchSize同理淮阐。kafka.producer.acks
該參數(shù)的含義就留給看官去回想（很基礎(chǔ)的）叮阅，一般設(shè)為折衷的1即可。設(shè)為-1的可靠性最高泣特，但是相應(yīng)地會(huì)影響吞吐量浩姥。kafka.producer.linger.ms
Kafka Producer檢查批次是否ready的超時(shí)時(shí)間，超時(shí)即發(fā)送（與producer.batch.size共同作用）状您。一般設(shè)為數(shù)十到100毫秒勒叠，可以在時(shí)效性和吞吐量之間取得比較好的平衡。kafka.producer.compression.type
Producer消息壓縮算法膏孟，支持gzip/snappy/lz4眯分，如果希望降低消息的體積可以配置。
示例：
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.flumeBatchSize = 1000
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 50
a1.sinks.k1.kafka.producer.compression.type = snappy
Kafka Sink也支持其他Producer參數(shù)柒桑，可以按需配置弊决。
還有一點(diǎn)需要注意的是，F(xiàn)lume默認(rèn)引用的Kafka Client版本為0.9魁淳，其產(chǎn)生的消息在較高版本的Kafka Broker上沒有時(shí)間戳飘诗，因此非常建議手動(dòng)將$FLUME_HOME/lib目錄下的kafka-client JAR包替換成0.10.2或更高的版本。
Interceptor
攔截器方面就比較簡(jiǎn)單粗暴界逛，在注重吞吐量的場(chǎng)合一定不要使用或者自定義規(guī)則復(fù)雜的攔截器（比如自帶的Regex Interceptor昆稿、Search and Replace Interceptor），最好是不使用任何攔截器息拜，把數(shù)據(jù)清洗的任務(wù)交給下游去處理（Flink它不香嘛
Agent Process
在flume-env.sh中添加JVM參數(shù)溉潭，避免默認(rèn)堆內(nèi)存太小導(dǎo)致OOM净响。
export JAVA_OPTS="-Xms8192m -Xmx8192m -Xmn3072m -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+HeapDumpOnOutOfMemoryError"
另外，Taildir Source會(huì)積極地使用堆外內(nèi)存喳瓣，如果發(fā)現(xiàn)Flume消耗的總內(nèi)存量過大别惦，可以適當(dāng)限制直接內(nèi)存的用量，如：-XX:MaxDirectMemorySize=4096m夫椭。
Flume原生并沒有傳統(tǒng)意義上的“高可用”配置（Sink Group Failover不算）掸掸。為了防止Agent進(jìn)程因?yàn)楦鞣N原因靜默地掛掉，需要用一個(gè)“保姆腳本”（nanny script）定期檢測(cè)Agent進(jìn)程的狀態(tài)蹭秋，并及時(shí)拉起來扰付。當(dāng)然也可以在下游采用兩級(jí)Collector的架構(gòu)增強(qiáng)魯棒性，本文不表仁讨。Cloudera Community上有一個(gè)關(guān)于Flume HA的提問羽莺，參見這里。
The End
經(jīng)過上述適當(dāng)?shù)恼{(diào)優(yōu)過程洞豁，我們的單個(gè)Flume-NG Agent能夠輕松承受高達(dá)5W+ RPS的持續(xù)流量高峰盐固，比較令人滿意了。
民那晚安晚安丈挟。
