1.Flume介紹
1.1 架構介紹
image.png
agent本身是一個Java進程,運行在日誌收集節點—所謂日誌收集節點就是服務器節點。
agent裡面包含3個核心的組件:source—->channel—–>sink,類似生產者、倉庫、消費者的架構。
source:source組件是專門用來收集數據的,可以處理各種類型、各種格式的日誌數據,包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy、自定義。
channel:source組件把數據收集來以後,臨時存放在channel中,即channel組件在agent中是專門用來存放臨時數據的——對採集到的數據進行簡單的緩存,可以存放在memory、jdbc、file等等。
sink:sink組件是用於把數據發送到目的地的組件,目的地包括hdfs、logger、avro、thrift、ipc、file、null、Hbase、solr、自定義。
event:將傳輸的數據進行封裝,是flume傳輸數據的基本單位,如果是文本文件,通常是一行記錄,event也是事務的基本單位。event從source,流向channel,再到sink,本身為一個字節數組,並可攜帶headers(頭信息)信息。event代表著一個數據的最小完整單元,從外部數據源來,向外部的目的地去。
2.背景說明
我們的需求是將Java 應用的log信息進行收集,達到日誌採集的目的,agent目前主要有flume、Logstash,技術選型詳情在此就不表了,最終選擇的flume。
由於當時公司內部推行技術組件一直有難度,且也無法藉助行政手段,因此我們在設計時很多時候考慮都是儘量對應用透明,比如我們的flume source使用的是基於log文件的,而未使用應用與flume agent採用長連接的方式(該方式需要修改log4j配置,並且引入我們的jar),比如我們agent進行日誌等級判斷時 需要兼容各種日誌格式,因為我們難以推動各個應用方統一日誌格式……
sre方面,當時並沒有為agent預留內存等資源,所以一旦我們的agent出現資源佔用過多,都會比較敏感。
整個日誌平臺1.0版本的架構圖如下:
image.png
可以看到我們使用kafka將log信息做轉儲,消息消費者主要有HDFS、ES、Queue等。
3.定製開發
從我們的實際情況出發,我們做定製開發部分主要是source和sink部分。同時我們也開發了一套agent自動部署工具。
3.1 source定製
3.1.1 dirSource
基於文件的dirSource在flume高版本里已經去除了,原生的dirSource也存在很多性能和功能上的問題,為了在我們使用的flume1.6版本里繼續使用dirSource,我們就基於1.6實現了一版dirSource。
dirSource特性
基於NIO的WatchEvent進行log文件內容的寫操作監聽,同時有能動態的監聽文件的創建和刪除。我們豐富了這部分的匹配模式,可以實現靈活的文件監聽。
文件的讀取基於RandomAccessFile,按行讀取
將獲取內容進行處理封裝Event,存入Channel。
存在的問題
無論是WatchEvent還是RandomAccessFile在log瘋狂輸出時,CPU佔用會居高不下。
3.1.2 execSource
execSource為flume新版本推出的用來替代dirSource的一種實現方式,主要是通過Java執行shell命令,並且獲取shell命令的輸出信息,如tail、cat等。
我們在原生的execSource基礎上,實現了文件的自動監聽,實現了多命令模式,並且會自動回收長時間無內容產出的命令,優化了原有的線程關閉的操作及進程鉤子等。
execSource特性
基於NIO的WatchEvent進行log文件內容的寫操作監聽,同時有能動態的監聽文件的創建和刪除。我們豐富了這部分的匹配模式,可以實現靈活的文件監聽
多命令模式
自動回收長時間無內容產出的命令
重啟時自動清理無用的shell命令
存在的問題
flume agent進程被kill -9 時,對導致執行的shell命令無法退出,進而導致句柄得不到釋放,積累下來對服務器造成影響。
3.2 sink定製
我們採用的是kafka sink,flume原生的kafka sink使用的是老版本kafka producer client,發送消息時需要手動實現批量與異步,並且是消息發送的實現上存在一些不足,在大數據量時存在明顯的性能瓶頸,並且會由於集合中消息數量太多而報異常,進而丟失消息。
我們定製的版本使用的new kafka producer client ,並且對消息發送做了優化,同時對Channel參數做了大量的壓測,最終確定了最優配置。
kafkaSink特性
使用new kafka producer client ,默認異步批量發送
-
優化了消息體序列化方式
4.壓測
下文描述的壓測都是在建設日誌平臺過程中對flume的相關測試。
測試環境都是mac book pro ,這裡只關注各個測試項的對比信息。
測試一
類型 | 日誌總數 | 生產頻率 | cpu | cpu平均 | mem | 數據丟失 | 用時 |
---|---|---|---|---|---|---|---|
tailDirSource+New kafka sink | 50萬 | 2000/s | 16-27% | 20% | 230M | 幾百條以內 | 280s |
tailDirSource+Old kafka sink | 50萬 | 2000/s | 16-27% | 19% | 230M | 較上一種丟失少 | 280s |
tailDirSource+New kafka sink | 50萬 | 4000/s | 34-60% | 40% | 230M | <2000 | 145s |
tailDirSource+Old kafka sink | 50萬 | 4000/s | 34-57% | 41% | 230M | <200 | 145s |
execSource+Old kafka sink | 50萬 | 4000/s | <8% | 7.5% | 230M | <200 | 145s |
execSource+Old kafka sink+Spillable Memory Channel | 50萬 | 4000/s | 8-10% | 9.5% | 230M | <200 | 145s |
execSource+Old kafka sink+File Channel | 50萬 | 4000/s | 40-55% | 45% | 230M | <200 | 145s |
說明:
類型New kafka sink為:原生sink,使用kafka舊client,只定制了從head中獲取配置參數,拼接字符串
類型Old kafka sink為:深度定製版,使用kafka新client
結論:
flume 資源佔用從kafka發送部分目前沒有太好的優化方案,且舊kafka client數據丟失更加嚴重。
因此flume kafka sink 維持不變,後續可從flume source入手優化
測試二
類型 | 日誌總數 | 生產頻率 | cpu佔用 | cpu平均 | 內存佔用 | 數據丟失 | 用時 | JVM配置 |
---|---|---|---|---|---|---|---|---|
tailDirSource+ kafka api sink | 50萬 | 3100/s | 34-60% | 40% | 230M | <200 | 163s | 512M |
tailDirSource+ kafka sink | 50萬 | 3100/s | 34-57% | 41% | 230M | <200 | 163s | 512M |
execSource+ kafka sink | 50萬 | 3100/s | <8% | 7.5% | 230M | <200 | 163s | 512M |
execSource+ kafka sink+Spillable Memory Channel | 50萬 | 3100/s | 8-10% | 9.5% | 230M | <200 | 163s | 512M |
execSource+ kafka sink+File Channel | 50萬 | 3100/s | 40-55% | 45% | 230M | <200 | 163s | 512M |
execSource+ kafka sink+MemoryChannel | 500萬 | 31074/s | 30-100% | 40% | 1G | <200 | 163s | 1G |
execSource+ kafka sink+MemoryChannel | 250萬 | 15337/s | 15-20% | 18% | 450M | <200 | 163s | 1G |
execSource+ kafka sink+MemoryChannel+FastJSON | 250萬 | 15337/s | 18-22% | 20% | 420M | <200 | 163s | 1G |
custom execSource+ kafka sink+MemoryChannel+FastJSON | 250萬 | 15432/s | 18-25% | 21% | 420M | <200 | 162s | 1G |
custom execSource+ kafka sink+MemoryChannel+FastJSON | 125萬+125萬 | 7661/s+7668/s | 20-26% | 24% | 440M | <500 | 163s | 1G |
測試三
配置說明一
a1.sinks.k1.batch.num.messages = 5000 a1.sinks.k1.block.on.buffer.full = true a1.sinks.k1.buffer.memory = 167108864 a1.channels.c1.type = memory a1.channels.c1.capacity = 100000 a1.channels.c1.transactionCapacity = 1000 flume -Xmx256M -Xms256M
測試結果一
日誌寫數量 | 用時 | 線程數 | QPS | 日誌文件量 | 成功發送到kafka數量 | topic個數 | CPU | 內存 | 序列化方式 | 其他 |
---|---|---|---|---|---|---|---|---|---|---|
500萬 | 74s | 50 | 70000/s | 600m | 280萬(單個topic) | 2 | 未統計 | 300M | fastjson | agent異常 |
配置說明二
a1.sinks.k1.batch.num.messages = 5000 a1.sinks.k1.block.on.buffer.full = true a1.sinks.k1.buffer.memory = 167108864 a1.channels.c1.type = memory a1.channels.c1.capacity = 500000 a1.channels.c1.transactionCapacity = 500 a1.channels.c1.byteCapacity = 536870912 flume -Xmx256M -Xms256M
測試結果二
日誌寫數量 | 用時 | 線程數 | QPS | 日誌文件量 | 成功發送到kafka數量 | topic個數 | CPU | 內存 | 序列化方式 | 其他 |
---|---|---|---|---|---|---|---|---|---|---|
500萬 | 68s | 50 | 74000/s | 600m | 500萬(單個topic) | 2 | 200%以上 | 320M | fastjson | 無異常 |
500萬 | 68s | 50 | 74000/s | 600m | 500萬(單個topic) | 1 | 100%-200% | 320M | fastjson | 無異常 |
500萬 | 68s | 50 | 74000/s | 600m | 500萬(單個topic) | 1 | 小於100% | 280M | StringBuild拼接 | 無異常 |
總結
數據量過大時,sink中kafka client 緩存被存滿,kafka會報異常,設置block=true後,存入緩存會被阻塞,kafka不報異常,但是由於sink從channel中消費的速度遠低於source存入channel的速度,channel會報Unable to put event on required channel,flume停止提供服務。繼續寫入日誌,會重複發送錯誤。
該異常可通過增大channel的byteCapacity參數或者調大JVM的參數值(byteCapacity默認為JVM的80%)來提高報錯的閥值,且減小transactionCapacity 的值來減緩傳輸到sink的數據量。
JVM內存參數在7萬每秒的壓力下,設置為256M較為合適,byteCapacity設置為512M較為合適,當增加channel個數或者增大channel向sink傳輸的數據量時,都會導致sink消費過慢報異常(總結1中異常),單個channel內存消耗在300M左右。
對於數據量較大的應用,建議只發送單個topic。
個人介紹:
高廣超 :多年一線互聯網研發與架構設計經驗,擅長設計與落地高可用、高性能互聯網架構。目前就職於美團網,負責核心業務研發工作。
本文首發在 高廣超的簡書博客 轉載請註明!
閱讀更多 互聯網技術棧 的文章