調(diào)優(yōu) | 高吞吐量Flume Agent調(diào)優(yōu)小結(jié)

點擊上方藍(lán)色字體，選擇“設(shè)為星標(biāo)”

前言

所有電商企業(yè)在一年一度的雙11都要迎來大促與大考，前段時間在配合執(zhí)行全鏈路壓測的過程中，發(fā)現(xiàn)平時不太關(guān)注的Flume配置可能存在瓶頸。Flume在筆者負(fù)責(zé)的實時計算平臺里用于收集所有后端訪問日志和埋點日志，其效率和穩(wěn)定性比較重要。除了及時擴容之外，也有必要對Flume進(jìn)行調(diào)優(yōu)。

Flume系統(tǒng)以一個或多個Flume-NG Agent的形式部署，一個Agent對應(yīng)一個JVM進(jìn)程，并且由三個部分組成：Source、Channel和Sink，示意圖如下。

Source

Flume有3種能夠監(jiān)聽文件的Source，分別是Exec Source（配合tail -f命令）、Spooling Directory Source和Taildir Source。Taildir Source顯然是最好用的，在我們的實踐中，需要注意的參數(shù)列舉如下。

• filegroups 如果需要監(jiān)聽的日志文件較多，應(yīng)該將它們分散在不同的目錄下，并配置多個filegroup來并行讀取。注意日志文件的正則表達(dá)式要寫好，防止日志滾動重命名時仍然符合正則表達(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ù)控制從同一個文件中連續(xù)讀取的最大批次數(shù)量，默認(rèn)不限制。如果Flume同時監(jiān)聽多個文件，并且其中某個文件的寫入速度遠(yuǎn)快于其他文件，那么其他文件有可能幾乎無法被讀取，所以強烈建議設(shè)定此參數(shù)。示例：

a1.sources.r1.maxBatchCount = 100

• writePosInterval 該參數(shù)控制向記錄讀取位置的JSON文件（由positionFile參數(shù)指定）寫入inode和偏移量的頻率，默認(rèn)為3000ms。當(dāng)Agent重新啟動時，會從JSON文件中獲取最近記錄的偏移量開始讀取。也就是說，適當(dāng)降低writePosInterval可以減少Agent重啟導(dǎo)致的重復(fù)讀取的數(shù)據(jù)量。

a1.sources.r1.writePosInterval = 1000

Channel

Flume內(nèi)置了多種Channel的實現(xiàn)，比較常用的有Memory Channel、File Channel、JDBC Channel、Kafka Channel等。我們的選擇主要針對Memory Channel和File Channel兩種，對比一下：

• Memory Channel將staging事件數(shù)據(jù)存儲在Agent堆內(nèi)存中，F(xiàn)ile Channel則將它們存儲在指定的文件中；

• 如果Agent失敗，Memory Channel會丟失所有緩存的staging事件，F(xiàn)ile Channel則可以通過額外記錄的checkpoint信息恢復(fù)，保證斷點續(xù)傳；

• Memory Channel能夠容納的數(shù)據(jù)量受堆內(nèi)存的影響，而File Channel不受此限制。

鑒于我們下游業(yè)務(wù)的主要痛點在吞吐量與實時性，且可以容忍數(shù)據(jù)少量丟失，日志服務(wù)器的磁盤壓力也已經(jīng)比較大了，故Memory Channel更加合適。需要注意的參數(shù)如下。

• capacity、transactionCapacity 這兩個參數(shù)分別代表Channel最多能容納的事件數(shù)目，以及每個事務(wù)（即Source的一次put或者Sink的一次take）能夠包含的最多事件數(shù)目。顯然，必須滿足batchSize <= transactionCapacity <= capacity的關(guān)系。適當(dāng)調(diào)大capacity和transactionCapacity可以使得Channel的吞吐量增高，且能夠保證不會出現(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一條事件的超時時間，默認(rèn)為3秒，對于Memory Channel一般不用更改。如果業(yè)務(wù)數(shù)據(jù)是由很多突發(fā)流量組成（也就是說Channel經(jīng)常處于時滿時空的狀態(tài)），那么建議適當(dāng)調(diào)大。示例：

a1.channels.c1.keep-alive = 15

當(dāng)然File Channel也很常用，其參數(shù)就不再贅述，看官可參考官方文檔。

Sink

我們實時數(shù)倉接入層的起點是Kafka，自然要利用Kafka Sink。需要注意的參數(shù)列舉如下。

• kafka.flumeBatchSize

從Channel取出數(shù)據(jù)并發(fā)送到Kafka的批次大小，與Source的batchSize同理。

• kafka.producer.acks

該參數(shù)的含義就留給看官去回想（很基礎(chǔ)的），一般設(shè)為折衷的1即可。設(shè)為-1的可靠性最高，但是相應(yīng)地會影響吞吐量。

• kafka.producer.linger.ms

Kafka Producer檢查批次是否ready的超時時間，超時即發(fā)送（與producer.batch.size共同作用）。一般設(shè)為數(shù)十到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參數(shù)，可以按需配置。

Interceptor

攔截器方面就比較簡單粗暴，在注重吞吐量的場合一定不要使用或者自定義規(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會積極地使用堆外內(nèi)存，如果發(fā)現(xiàn)Flume消耗的總內(nèi)存量過大，可以適當(dāng)限制直接內(nèi)存的用量，如：-XX:MaxDirectMemorySize=4096m。

Flume原生并沒有傳統(tǒng)意義上的“高可用”配置（Sink Group Failover不算）。為了防止Agent進(jìn)程因為各種原因靜默地掛掉，需要用一個“保姆腳本”（nanny script）定期檢測Agent進(jìn)程的狀態(tài)，并及時拉起來。當(dāng)然也可以在下游采用兩級Collector的架構(gòu)增強魯棒性，本文不表。

The End

經(jīng)過上述適當(dāng)?shù)恼{(diào)優(yōu)過程，我們的單個Flume-NG Agent能夠輕松承受高達(dá)5W+ RPS的持續(xù)流量高峰，比較令人滿意了。

版權(quán)聲明：

文章不錯？點個【在看】吧！??