問題導讀： 1.Flume的存在些什麼問題？ 2.基於開源的Flume美團增加了哪些功能？ 3.Flume系統如何調優？
在《基於Flume的美團日誌收集系統(一)架構和設計》中，我們詳述了基於Flume的美團日誌收集系統的架構設計，以及為什麼做這樣的設計。在本節中，我們將會講述在實際部署和使用過程中遇到的問題，對Flume的功能改進和對系統做的優化。 1 Flume的問題總結 在Flume的使用過程中，遇到的主要問題如下： a. Channel“水土不服”：使用固定大小的MemoryChannel在日誌高峰時常報佇列大小不夠的異常；使用FileChannel又導致IO繁忙的問題； b. HdfsSink的效能問題：使用HdfsSink向Hdfs寫日誌，在高峰時間速度較慢； c. 系統的管理問題：配置升級，模組重啟等； 2 Flume的功能改進和優化點         從上面的問題中可以看到，有一些需求是原生Flume無法滿足的，因此，基於開源的Flume我們增加了許多功能，修改了一些Bug，並且進行一些調優。下面將對一些主要的方面做一些說明。 2.1 增加Zabbix monitor服務        一方面，Flume本身提供了http, ganglia的監控服務，而我們目前主要使用zabbix做監控。因此，我們為Flume添加了zabbix監控模組，和sa的監控服務無縫融合。        另一方面，淨化Flume的metrics。只將我們需要的metrics傳送給zabbix，避免 zabbix server造成壓力。目前我們最為關心的是Flume能否及時把應用端傳送過來的日誌寫到Hdfs上， 對應關注的metrics為：

• Source : 接收的event數和處理的event數
• Channel : Channel中擁堵的event數
• Sink : 已經處理的event數
  

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

event大小(Byte)	sink.batch-size	hdfs.batchSize	壓縮格式	總資料大小(G)	耗時(s)	平均events/s	壓縮後大小(G)
544	300	10000	bz2	9.1	2448	6833	1.36
544	300	10000	lzo	9.1	612	27333	3.49

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

1. /**
   
2.    * Rename bucketPath file from .tmp to permanent location.
   
3.    */
   
4.   private void renameBucket() throws IOException, InterruptedException {
   
5.       if(bucketPath.equals(targetPath)) {
   
6.               return;
   
7.         }
   
8.         final Path srcPath = new Path(bucketPath);
   
9.         final Path dstPath = new Path(targetPath);
   
10.         callWithTimeout(new CallRunner<Object>() {
    
11.               @Override
    
12.               public Object call() throws Exception {
    
13.                 if(fileSystem.exists(srcPath)) { // could block
    
14.                       LOG.info("Renaming " + srcPath + " to " + dstPath);
    
15.                      fileSystem.rename(srcPath, dstPath); // could block
    
16.                       //index the dstPath lzo file
    
17.                       if (codeC != null && ".lzo".equals(codeC.getDefaultExtension()) ) {
    
18.                               LzoIndexer lzoIndexer = new LzoIndexer(new Configuration());
    
19.                               lzoIndexer.index(dstPath);
    
20.                       }
    
21.                 }
    
22.                 return null;
    
23.               }
    
24.     });
    
25. }

複製程式碼
2.3 增加HdfsSink的開關        我們在HdfsSink和DualChannel中增加開關，當開關開啟的情況下，HdfsSink不再往Hdfs上寫資料，並且資料只寫向DualChannel中的FileChannel。以此策略來防止Hdfs的正常停機維護。 2.4 增加DualChannel       Flume本身提供了MemoryChannel和FileChannel。MemoryChannel處理速度快，但快取大小有限，且沒有持久化；FileChannel則剛好相反。我們希望利用兩者的優勢，在Sink處理速度夠快，Channel沒有快取過多日誌的時候，就使用MemoryChannel，當Sink處理速度跟不上，又需要Channel能夠快取下應用端傳送過來的日誌時，就使用FileChannel，由此我們開發了DualChannel，能夠智慧的在兩個Channel之間切換。 其具體的邏輯如下：

1. /***
   
2. * putToMemChannel indicate put event to memChannel or fileChannel
   
3. * takeFromMemChannel indicate take event from memChannel or fileChannel
   
4. * */
   
5. private AtomicBoolean putToMemChannel = new AtomicBoolean(true);
   
6. private AtomicBoolean takeFromMemChannel = new AtomicBoolean(true);
   
7. void doPut(Event event) {
   
8.         if (switchon && putToMemChannel.get()) {
   
9.               //往memChannel中寫資料
   
10.               memTransaction.put(event);
    
11.               if ( memChannel.isFull() || fileChannel.getQueueSize() > 100) {
    
12.                 putToMemChannel.set(false);
    
13.               }
    
14.         } else {
    
15.               //往fileChannel中寫資料
    
16.               fileTransaction.put(event);
    
17.         }
    
18.   }
    
19. Event doTake() {
    
20.     Event event = null;
    
21.     if ( takeFromMemChannel.get() ) {
    
22.         //從memChannel中取資料
    
23.         event = memTransaction.take();
    
24.         if (event == null) {
    
25.             takeFromMemChannel.set(false);
    
26.         } 
    
27.     } else {
    
28.         //從fileChannel中取資料
    
29.         event = fileTransaction.take();
    
30.         if (event == null) {
    
31.             takeFromMemChannel.set(true);
    
32.             putToMemChannel.set(true);
    
33.         } 
    
34.     }
    
35.     return event;
    
36. }

複製程式碼


2.5 增加NullChannel        Flume提供了NullSink，可以把不需要的日誌通過NullSink直接丟棄，不進行儲存。然而，Source需要先將events存放到Channel中，NullSink再將events取出扔掉。為了提升效能，我們把這一步移到了Channel裡面做，所以開發了NullChannel。 2.6 增加KafkaSink        為支援向Storm提供實時資料流，我們增加了KafkaSink用來向Kafka寫實時資料流。其基本的邏輯如下：

1. public class KafkaSink extends AbstractSink implements Configurable {
   
2.         private String zkConnect;
   
3.         private Integer zkTimeout;
   
4.         private Integer batchSize;
   
5.         private Integer queueSize;
   
6.         private String serializerClass;
   
7.         private String producerType;
   
8.         private String topicPrefix;
   
9.         private Producer<String, String> producer;
   
10.         public void configure(Context context) {
    
11.             //讀取配置，並檢查配置
    
12.         }
    
13.         @Override
    
14.         public synchronized void start() {
    
15.             //初始化producer
    
16.         }
    
17.         @Override
    
18.         public synchronized void stop() {
    
19.             //關閉producer
    
20.         }
    
21.         @Override
    
22.         public Status process() throws EventDeliveryException {
    
23.             Status status = Status.READY;
    
24.             Channel channel = getChannel();
    
25.             Transaction tx = channel.getTransaction();
    
26.             try {
    
27.                     tx.begin();
    
28.                     //將日誌按category分佇列存放
    
29.                     Map<String, List<String>> topic2EventList = new HashMap<String, List<String>>();
    
30.                     //從channel中取batchSize大小的日誌，從header中獲取category，生成topic，並存放於上述的Map中；
    
31.                     //將Map中的資料通過producer傳送給kafka 
    
32.                    tx.commit();
    
33.             } catch (Exception e) {
    
34.                     tx.rollback();
    
35.                     throw new EventDeliveryException(e);
    
36.             } finally {
    
37.                 tx.close();
    
38.             }
    
39.             return status;
    
40.         }
    
41. }

複製程式碼


2.7 修復和scribe的相容問題        Scribed在通過ScribeSource傳送資料包給Flume時，大於4096位元組的包，會先發送一個Dummy包檢查伺服器的反應，而Flume的ScribeSource對於logentry.size()=0的包返回TRY_LATER，此時Scribed就認為出錯，斷開連線。這樣迴圈反覆嘗試，無法真正傳送資料。現在在ScribeSource的Thrift介面中，對size為0的情況返回OK，保證後續正常傳送資料。 3. Flume系統調優經驗總結3.1 基礎引數調優經驗

• HdfsSink中預設的serializer會每寫一行在行尾新增一個換行符，我們日誌本身帶有換行符，這樣會導致每條日誌後面多一個空行，修改配置不要自動新增換行符；
  

1. lc.sinks.sink_hdfs.serializer.appendNewline = false

複製程式碼

• 調大MemoryChannel的capacity，儘量利用MemoryChannel快速的處理能力；
  

• 調大HdfsSink的batchSize，增加吞吐量，減少hdfs的flush次數；
• 適當調大HdfsSink的callTimeout，避免不必要的超時錯誤；
  

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

1. lc.sinks.sink_hdfs.hdfs.path = /user/hive/work/orglog.db/%{category}/dt=%Y%m%d/hour=%H

複製程式碼
       HdfsS ink中處理每條event時，都要根據配置獲取此event應該寫入的Hdfs path和filename，預設的獲取方法是通過正則表示式替換配置中的變數，獲取真實的path和filename。因為此過程是每條event都要做的操作，耗時很長。通過我們的測試，20萬條日誌，這個操作要耗時6-8s左右。 由於我們目前的path和filename有固定的模式，可以通過字串拼接獲得。而後者比正則匹配快幾十倍。拼接定符串的方式，20萬條日誌的操作只需要幾百毫秒。 3.3 HdfsSink的b/m/s優化       在我們初始的設計中，所有的日誌都通過一個Channel和一個HdfsSink寫到Hdfs上。我們來看一看這樣做有什麼問題。 首先，我們來看一下HdfsSink在傳送資料的邏輯：

1. //從Channel中取batchSize大小的events
   
2. for (txnEventCount = 0; txnEventCount < batchSize; txnEventCount++) {
   
3.     //對每條日誌根據category append到相應的bucketWriter上；
   
4.     bucketWriter.append(event);
   
5. ｝
   
6. for (BucketWriter bucketWriter : writers) {
   
7.     //然後對每一個bucketWriter呼叫相應的flush方法將資料flush到Hdfs上
   
8.     bucketWriter.flush();
   
9. ｝

複製程式碼

       假設我們的系統中有100個category，batchSize大小設定為20萬。則每20萬條資料，就需要對100個檔案進行append或者flush操作。 其次，對於我們的日誌來說，基本符合80/20原則。即20%的category產生了系統80%的日誌量。這樣對大部分日誌來說，每20萬條可能只包含幾條日誌，也需要往Hdfs上flush一次。 上述的情況會導致HdfsSink寫Hdfs的效率極差。下圖是單Channel的情況下每小時的傳送量和寫hdfs的時間趨勢圖。  
       鑑於這種實際應用場景，我們把日誌進行了大小歸類，分為big, middle和small三類，這樣可以有效的避免小日誌跟著大日誌一起頻繁的flush，提升效果明顯。下圖是分佇列後big佇列的每小時的傳送量和寫hdfs的時間趨勢圖。