基于flume的美團(tuán)日志收集系統(tǒng)-資料下載頁
2025-06-18 15:52本頁面
【正文】 //讀取配置，并檢查配置 } @Override public synchronized void start() { //初始化producer } @Override public synchronized void stop() { //關(guān)閉producer } @Override public Status process() throws EventDeliveryException { Status status = 。 Channel channel = getChannel()。 Transaction tx = ()。 try { ()。 //將日志按category分隊(duì)列存放 MapString, ListString topic2EventList = new HashMapString, ListString()。 //從channel中取batchSize大小的日志，從header中獲取category，生成topic，并存放于上述的Map中； //將Map中的數(shù)據(jù)通過producer發(fā)送給kafka ()。 } catch (Exception e) { ()。 throw new EventDeliveryException(e)。 } finally { ()。 } return status。 }} 修復(fù)和scribe的兼容問題Scribed在通過ScribeSource發(fā)送數(shù)據(jù)包給Flume時(shí)，大于4096字節(jié)的包，會(huì)先發(fā)送一個(gè)Dummy包檢查服務(wù)器的反應(yīng)，()=0的包返回TRY_LATER，此時(shí)Scribed就認(rèn)為出錯(cuò)，斷開連接。這樣循環(huán)反復(fù)嘗試，無法真正發(fā)送數(shù)據(jù)?，F(xiàn)在在ScribeSource的Thrift接口中，對(duì)size為0的情況返回OK，保證后續(xù)正常發(fā)送數(shù)據(jù)。3. 基礎(chǔ)參數(shù)調(diào)優(yōu)經(jīng)驗(yàn)l HdfsSink中默認(rèn)的serializer會(huì)每寫一行在行尾添加一個(gè)換行符，我們?nèi)罩颈旧韼в袚Q行符，這樣會(huì)導(dǎo)致每條日志后面多一個(gè)空行，修改配置不要自動(dòng)添加換行符； = falsel 調(diào)大MemoryChannel的capacity，盡量利用MemoryChannel快速的處理能力；l 調(diào)大HdfsSink的batchSize，增加吞吐量，減少hdfs的flush次數(shù)；l 適當(dāng)調(diào)大HdfsSink的callTimeout，避免不必要的超時(shí)錯(cuò)誤； HdfsSink獲取Filename的優(yōu)化HdfsSink的path參數(shù)指明了日志被寫到Hdfs的位置，該參數(shù)中可以引用格式化的參數(shù)，將日志寫到一個(gè)動(dòng)態(tài)的目錄中。這方便了日志的管理。例如我們可以將日志寫到category分類的目錄，并且按天和按小時(shí)存放： = /user/hive/work/{category}/dt=%Y%m%d/hour=%HHdfsS ink中處理每條event時(shí)，都要根據(jù)配置獲取此event應(yīng)該寫入的Hdfs path和filename，默認(rèn)的獲取方法是通過正則表達(dá)式替換配置中的變量，獲取真實(shí)的path和filename。因?yàn)榇诉^程是每條event都要做的操作，耗時(shí)很長。通過我們的測(cè)試，20萬條日志，這個(gè)操作要耗時(shí)68s左右。由于我們目前的path和filename有固定的模式，可以通過字符串拼接獲得。而后者比正則匹配快幾十倍。拼接定符串的方式，20萬條日志的操作只需要幾百毫秒。 HdfsSink的b/m/s優(yōu)化在我們初始的設(shè)計(jì)中，所有的日志都通過一個(gè)Channel和一個(gè)HdfsSink寫到Hdfs上。我們來看一看這樣做有什么問題。首先，我們來看一下HdfsSink在發(fā)送數(shù)據(jù)的邏輯：//從Channel中取batchSize大小的eventsfor (txnEventCount = 0。 txnEventCount batchSize。 txnEventCount++) { //對(duì)每條日志根據(jù)category append到相應(yīng)的bucketWriter上； (event)。｝for (BucketWriter bucketWriter : writers) { //然后對(duì)每一個(gè)bucketWriter調(diào)用相應(yīng)的flush方法將數(shù)據(jù)flush到Hdfs上 ()。｝假設(shè)我們的系統(tǒng)中有100個(gè)category，batchSize大小設(shè)置為20萬。則每20萬條數(shù)據(jù)，就需要對(duì)100個(gè)文件進(jìn)行append或者flush操作。其次，對(duì)于我們的日志來說，基本符合80/20原則。即20%的category產(chǎn)生了系統(tǒng)80%的日志量。這樣對(duì)大部分日志來說，每20萬條可能只包含幾條日志，也需要往Hdfs上flush一次。上述的情況會(huì)導(dǎo)致HdfsSink寫Hdfs的效率極差。下圖是單Channel的情況下每小時(shí)的發(fā)送量和寫hdfs的時(shí)間趨勢(shì)圖。鑒于這種實(shí)際應(yīng)用場(chǎng)景，我們把日志進(jìn)行了大小歸類，分為big, middle和small三類，這樣可以有效的避免小日志跟著大日志一起頻繁的flush，提升效果明顯。下圖是分隊(duì)列后big隊(duì)列的每小時(shí)的發(fā)送量和寫hdfs的時(shí)間趨勢(shì)圖。13
職業(yè)教育相關(guān)推薦
鄂ICP備17016276號(hào)-1