Flume + MongoDB 實戰：Java 客戶端寫入日誌 → Flume Avro Source → MongoDB Sink 全鏈路搭建

架構總覽

Java App (log4j/Logback) 
    → Flume Avro Source (port 44444) 
    → Memory Channel 
    → MongoSink (custom) 
    → MongoDB (test.log collection)

環境準備

Flume 設定 conf/mongo-agent.properties

agent.sources = avro-src
agent.channels = mem-ch
agent.sinks = mongo-sink

# Avro Source
avro-src.type = avro
avro-src.bind = 0.0.0.0
avro-src.port = 44444
avro-src.channels = mem-ch

# Memory Channel
mem-ch.type = memory
mem-ch.capacity = 1000
mem-ch.transactionCapacity = 100

# MongoDB Sink (需自訂實現類)
mongo-sink.type = org.riderzen.flume.sink.MongoSink
mongo-sink.host = localhost
mongo-sink.port = 27017
mongo-sink.model = single
mongo-sink.db = test
mongo-sink.collection = log
mongo-sink.batch = 100
mongo-sink.channel = mem-ch

啟動 Flume Agent

bin/flume-ng agent -n agent -c conf -f conf/mongo-agent.properties -Dflume.root.logger=INFO,console

Java 客戶端 (Maven 專案)

目錄結構

src/main/java/cn/sinobest/flume/client/demo/LogDemo.java
src/main/resources/log4j.properties
pom.xml

log4j.properties → Flume Appender

log4j.rootLogger=INFO, flume
log4j.appender.flume=org.apache.flume.clients.log4jappender.Log4jAppender
log4j.appender.flume.Hostname=localhost
log4j.appender.flume.Port=44444
log4j.appender.flume.UnsafeMode=true
log4j.appender.flume.AvroReflectionEnabled=false

LogDemo.java

package cn.sinobest.flume.client.demo;
import org.apache.log4j.Logger;
public class LogDemo {
    private static final Logger LOG = Logger.getLogger(LogDemo.class);
    public static void main(String[] args) {
        for (int i = 0; i < 100; i++) {
            LOG.info("測試日誌 " + i + " @ " + System.currentTimeMillis());
            try { Thread.sleep(100); } catch (InterruptedException ignored) {}
        }
    }
}

pom.xml 關鍵依賴

<dependencies>
  <dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-sdk</artifactId><version>1.6.0</version></dependency>
  <dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-clients</artifactId><version>1.6.0</version></dependency>
  <dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.17</version></dependency>
</dependencies>

驗證資料落庫

> use test
> db.log.find().pretty()
{
  "_id" : ObjectId("..."),
  "timestamp" : ISODate("2017-01-18T07:30:00.123Z"),
  "message" : "測試日誌 42 @ 1484724600123",
  "level" : "INFO",
  "logger" : "cn.sinobest.flume.client.demo.LogDemo"
}

生產化建議

• Channel 改用 File Channel 或 Kafka Channel 保證不丟資料
• Sink 批次大小 batch 配合 transactionCapacity 調優吞吐
• MongoDB 寫入關注 WriteConcern.ACKNOWLEDGED + 索引優化 {timestamp: 1}
• 監控指標：Channel 填滿率、Sink 處理延遲、MongoDB 寫入 QPS