點選上方藍色字型，選擇“設為星标”
回複”資源“擷取更多資源
前言
所有電商企業在一年一度的雙11都要迎來大促與大考，前段時間在配合執行全鍊路壓測的過程中，發現平時不太關注的Flume配置可能存在瓶頸。Flume在筆者負責的實時計算平台裡用于收集所有後端通路日志和埋點日志，其效率和穩定性比較重要。除了及時擴容之外，也有必要對Flume進行調優。
Flume系統以一個或多個Flume-NG Agent的形式部署，一個Agent對應一個JVM程序，并且由三個部分組成：Source、Channel和Sink，示意圖如下。
Source
Flume有3種能夠監聽檔案的Source，分别是Exec Source（配合tail -f指令）、Spooling Directory Source和Taildir Source。Taildir Source顯然是最好用的，在我們的實踐中，需要注意的參數列舉如下。
- filegroups 如果需要監聽的日志檔案較多，應該将它們分散在不同的目錄下，并配置多個filegroup來并行讀取。注意日志檔案的正規表達式要寫好，防止日志滾動重命名時仍然符合正規表達式造成重複。示例：
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 該參數控制向Channel發送資料的批次大小，預設值為100。如果日志流量很大，适當增加此值可以增大Source的吞吐量，但是不能超過Channel的capacity和transactionCapacity的限制（後文再說）。示例：
a1.sources.r1.batchSize = 1000
- maxBatchCount 該參數控制從同一個檔案中連續讀取的最大批次數量，預設不限制。如果Flume同時監聽多個檔案，并且其中某個檔案的寫入速度遠快于其他檔案，那麼其他檔案有可能幾乎無法被讀取，是以強烈建議設定此參數。示例：
a1.sources.r1.maxBatchCount = 100
- writePosInterval 該參數控制向記錄讀取位置的JSON檔案（由positionFile參數指定）寫入inode和偏移量的頻率，預設為3000ms。當Agent重新啟動時，會從JSON檔案中擷取最近記錄的偏移量開始讀取。也就是說，适當降低writePosInterval可以減少Agent重新開機導緻的重複讀取的資料量。
a1.sources.r1.writePosInterval = 1000
Channel
Flume内置了多種Channel的實作，比較常用的有Memory Channel、File Channel、JDBC Channel、Kafka Channel等。我們的選擇主要針對Memory Channel和File Channel兩種，對比一下：
- Memory Channel将staging事件資料存儲在Agent堆記憶體中，File Channel則将它們存儲在指定的檔案中；
- 如果Agent失敗，Memory Channel會丢失所有緩存的staging事件，File Channel則可以通過額外記錄的checkpoint資訊恢複，保證斷點續傳；
- Memory Channel能夠容納的資料量受堆記憶體的影響，而File Channel不受此限制。
鑒于我們下遊業務的主要痛點在吞吐量與實時性，且可以容忍資料少量丢失，日志伺服器的磁盤壓力也已經比較大了，故Memory Channel更加合适。需要注意的參數如下。
- capacity、transactionCapacity 這兩個參數分别代表Channel最多能容納的事件數目，以及每個事務（即Source的一次put或者Sink的一次take）能夠包含的最多事件數目。顯然，必須滿足batchSize <= transactionCapacity <= capacity的關系。适當調大capacity和transactionCapacity可以使得Channel的吞吐量增高，且能夠保證不會出現The channel is full or unexpected failure的異常。示例：
a1.channels.c1.type = memory a1.channels.c1.transactionCapacity = 5000 a1.channels.c1.capacity = 10000 - byteCapacity
該參數代表Memory Channel中緩存的事件消息的最大總大小，以位元組為機關，預設是Flume Agent最大堆記憶體的80%。此值不建議更改為固定的，而是建議通過改變Agent的JVM參數來影響，後面再提。
- byteCapacityBufferPercentage
Memory Channel中緩存的事件消息頭占byteCapacity的比例，預設是20%。如果事件的header資訊很少，可以适當減小（我們沒有更改）。
- keep-alive
向Channel中put或take一條事件的逾時時間，預設為3秒，對于Memory Channel一般不用更改。如果業務資料是由很多突發流量組成（也就是說Channel經常處于時滿時空的狀态），那麼建議适當調大。示例：
a1.channels.c1.keep-alive = 15
當然File Channel也很常用，其參數就不再贅述，看官可參考官方文檔。
Sink
我們實時數倉接入層的起點是Kafka，自然要利用Kafka Sink。需要注意的參數列舉如下。
- kafka.flumeBatchSize
從Channel取出資料并發送到Kafka的批次大小，與Source的batchSize同理。
- kafka.producer.acks
該參數的含義就留給看官去回想（很基礎的），一般設為折衷的1即可。設為-1的可靠性最高，但是相應地會影響吞吐量。
- kafka.producer.linger.ms
Kafka Producer檢查批次是否ready的逾時時間，逾時即發送（與producer.batch.size共同作用）。一般設為數十到100毫秒，可以在時效性和吞吐量之間取得比較好的平衡。
- 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參數，可以按需配置。
Interceptor
攔截器方面就比較簡單粗暴，在注重吞吐量的場合一定不要使用或者自定義規則複雜的攔截器（比如自帶的Regex Interceptor、Search and Replace Interceptor），最好是不使用任何攔截器，把資料清洗的任務交給下遊去處理（Flink它不香嘛
Agent Process
在flume-env.sh中添加JVM參數，避免預設堆記憶體太小導緻OOM。
export JAVA_OPTS="-Xms8192m -Xmx8192m -Xmn3072m
-XX:+UseParNewGC
-XX:+UseConcMarkSweepGC
-XX:+HeapDumpOnOutOfMemoryError"
另外，Taildir Source會積極地使用堆外記憶體，如果發現Flume消耗的總記憶體量過大，可以适當限制直接記憶體的用量，如：-XX:MaxDirectMemorySize=4096m。
Flume原生并沒有傳統意義上的“高可用”配置（Sink Group Failover不算）。為了防止Agent程序因為各種原因靜默地挂掉，需要用一個“保姆腳本”（nanny script）定期檢測Agent程序的狀态，并及時拉起來。當然也可以在下遊采用兩級Collector的架構增強魯棒性，本文不表。
The End
經過上述适當的調優過程，我們的單個Flume-NG Agent能夠輕松承受高達5W+ RPS的持續流量高峰，比較令人滿意了。
版權聲明：
本文為大資料技術與架構整理，原作者獨家授權。未經原作者允許轉載追究侵權責任。
編輯｜冷眼丶
微信公衆号｜import_bigdata
歡迎點贊+收藏+轉發朋友圈素質三連
文章不錯？點個【在看】吧！ ????