基於Flume的美團日誌收集系統二改進和優化

dju alex

·2013-12-09 22:35

**：在《基於flume的美團日誌收集系統(一)架構和設計》中，我們詳述了基於flume的美團日誌收集系統的架構設計，以及為什麼做這樣的設計。在本節中，我們將會講述在實際部署和使用過程中遇到的問題，對flume的功能改進和對系統做的優化。

在flume的使用過程中，遇到的主要問題如下：

a. channel「水土不服」：使用固定大小的memorychannel在日誌高峰時常報佇列大小不夠的異常；使用filechannel又導致io繁忙的問題；

b. hdfssink的效能問題：使用hdfssink向hdfs寫日誌，在高峰時間速度較慢；

c. 系統的管理問題：配置公升級，模組重啟等；

從上面的問題中可以看到，有一些需求是原生flume無法滿足的，因此，基於開源的flume我們增加了許多功能，修改了一些bug，並且進行一些調優。下面將對一些主要的方面做一些說明。

一方面，flume本身提供了http, ganglia的監控服務，而我們目前主要使用zabbix做監控。因此，我們為flume新增了zabbix監控模組，和sa的監控服務無縫融合。

另一方面，淨化flume的metrics。只將我們需要的metrics傳送給zabbix，避免 zabbix server造成壓力。目前我們最為關心的是flume能否及時把應用端傳送過來的日誌寫到hdfs上，對應關注的metrics為：

首先，我們的hdfssink寫到hadoop的檔案採用lzo壓縮儲存。 hdfssink可以讀取hadoop配置檔案中提供的編碼類列表，然後通過配置的方式獲取使用何種壓縮編碼，我們目前使用lzo壓縮資料。採用lzo壓縮而非bz2壓縮，是基於以下測試資料：

event大小(byte)

sink.batch-size

hdfs.batchsize

壓縮格式

總資料大小(g)

耗時(s)

平均events/s

壓縮後大小(g)

544300

10000

bz29.1

2448

6833

1.36

544300

10000

lzo9.1

61227333

3.49

其次，我們的hdfssink增加了建立lzo檔案後自動建立index功能。hadoop提供了對lzo建立索引，使得壓縮檔案是可切分的，這樣hadoop job可以並行處理資料檔案。hdfssink本身lzo壓縮，但寫完lzo檔案並不會建索引，我們在close檔案之後新增了建索引功能。

/**
* rename bucketpath file from .tmp to permanent location.
*/private void renamebucket() throws ioexception, interruptedexception 
final path srcpath = new path(bucketpath);
final path dstpath = new path(targetpath);
callwithtimeout(new callrunner() 
}return null;
}});
}

我們在hdfssink和dualchannel中增加開關，當開關開啟的情況下，hdfssink不再往hdfs上寫資料，並且資料只寫向dualchannel中的filechannel。以此策略來防止hdfs的正常停機維護。

flume本身提供了memorychannel和filechannel。memorychannel處理速度快，但快取大小有限，且沒有持久化；filechannel則剛好相反。我們希望利用兩者的優勢，在sink處理速度夠快，channel沒有快取過多日誌的時候，就使用memorychannel，當sink處理速度跟不上，又需要channel能夠快取下應用端傳送過來的日誌時，就使用filechannel，由此我們開發了dualchannel，能夠智慧型的在兩個channel之間切換。

其具體的邏輯如下：

/***
* puttomemchannel indicate put event to memchannel or filechannel
* takefrommemchannel indicate take event from memchannel or filechannel
* */
private atomicboolean puttomemchannel = new atomicboolean(true);
private atomicboolean takefrommemchannel = new atomicboolean(true);
void doput(event event) 
} else 
}event dotake() 
} else 
}return event;
}

flume提供了nullsink，可以把不需要的日誌通過nullsink直接丟棄，不進行儲存。然而，source需要先將events存放到channel中，nullsink再將events取出扔掉。為了提公升效能，我們把這一步移到了channel裡面做，所以開發了nullchannel。

為支援向storm提供實時資料流，我們增加了kafkasink用來向kafka寫實時資料流。其基本的邏輯如下：

public class kafkasink extends abstractsink implements configurable 
@override
public synchronized void start() 
@override
public synchronized void stop() 
@override
public status process() throws eventdeliveryexception catch (exception e) finally 
return status;}}

scribed在通過scribesource傳送資料報給flume時，大於4096位元組的包，會先傳送乙個dummy包檢查伺服器的反應，而flume的scribesource對於logentry.size()=0的包返回try_later，此時scribed就認為出錯，斷開連線。這樣迴圈反覆嘗試，無法真正傳送資料。現在在scribesource的thrift介面中，對size為0的情況返回ok，保證後續正常傳送資料。

hdfssink的path引數指明了日誌被寫到hdfs的位置，該引數中可以引用格式化的引數，將日誌寫到乙個動態的目錄中。這方便了日誌的管理。例如我們可以將日誌寫到category分類的目錄，並且按天和按小時存放：

lc.sinks.sink_hdfs.hdfs.path = /user/hive/work/orglog.db/%/dt=%y%m%d/hour=%h

hdfss ink中處理每條event時，都要根據配置獲取此event應該寫入的hdfs path和filename，預設的獲取方法是通過正規表示式替換配置中的變數，獲取真實的path和filename。因為此過程是每條event都要做的操作，耗時很長。通過我們的測試，20萬條日誌，這個操作要耗時6-8s左右。

由於我們目前的path和filename有固定的模式，可以通過字串拼接獲得。而後者比正則匹配快幾十倍。拼接定符串的方式，20萬條日誌的操作只需要幾百毫秒。

在我們初始的設計中，所有的日誌都通過乙個channel和乙個hdfssink寫到hdfs上。我們來看一看這樣做有什麼問題。

首先，我們來看一下hdfssink在傳送資料的邏輯：

//從channel中取batchsize大小的events
for (txneventcount = 0; txneventcount < batchsize; txneventcount++) {
｝for (bucketwriter bucketwriter : writers) {
//然後對每乙個bucketwriter呼叫相應的flush方法將資料flush到hdfs上
bucketwriter.flush();
｝

其次，對於我們的日誌來說，基本符合80/20原則。即20%的category產生了系統80%的日誌量。這樣對大部分日誌來說，每20萬條可能只包含幾條日誌，也需要往hdfs上flush一次。

上述的情況會導致hdfssink寫hdfs的效率極差。下圖是單channel的情況下每小時的傳送量和寫hdfs的時間趨勢圖。

鑑於這種實際應用場景，我們把日誌進行了大小歸類，分為big, middle和small三類，這樣可以有效的避免小日誌跟著大日誌一起頻繁的flush，提公升效果明顯。下圖是分佇列後big佇列的每小時的傳送量和寫hdfs的時間趨勢圖。

目前，flume日誌收集系統提供了乙個高可用，高可靠，可擴充套件的分布式服務，已經有效地支援了美團的日誌資料收集工作。

後續，我們將在如下方面繼續研究：

基於Flume的美團日誌收集系統 二 改進和優化

基於Flume的美團日誌收集系統 二 改進和優化

基於Flume的美團日誌收集系統 二 改進和優化

基於Flume的日誌收集系統方案參考

相關推薦

基於Flume的美團日誌收集系統二改進和優化

基於Flume的美團日誌收集系統二改進和優化

基於Flume的美團日誌收集系統二改進和優化