Flume學習應用：Java寫日誌數據到MongoDB

概述

系統環境

• 操作系統：win7 64
• JDK:1.6.0_43

資源下載

• Maven:3.3.3
  下載、安裝、入門參考：1. Maven - 開始和2. 創建一個簡單的Maven項目
• Flume:1.6.0
  下載apache-flume-1.6.0-bin.tar.gz，並解壓到合適的目錄（e.g. F:\temp\apache-flume-1.6.0-bin）
• MongoDB:3.2.3
  下載mongodb-win32-x86_64-3.2.3-signed.msi，並安裝到合適的目錄（e.g. D:\develop\MongoDB）

輔助資源下載

• MongoDB Java Driver:2.13.0
  下載地址：mongo-java-driver-2.13.0.jar
  下載之後放入Flume安裝目錄的lib下。
• flume-ng-mongodb-sink
  Flume需要這個，才能把數據寫到MongoDB中
  下載地址：flume-ng-mongodb-sink
  這是一個基於Maven的源代碼，package之後，將jar包放到Flume安裝目錄的lib下。

初始化

數據初始化

1. 新建一個目錄，作為dbpath
   e.g. D:\develop\MongoDB\mydata
2. 啟動監聽服務 - CMD運行
   

   D:\develop\MongoDB\bin\mongod --dbpath=D:\develop\MongoDB\mydata

配置Flume

1. 生成flume-env.ps1
   在安裝目錄的conf/下，複製flume-env.ps1.template，重命名為flume-env.ps1
2. 修改log4j.properties
   在安裝目錄的conf/下，修改文件log4j.properties；修改flume.root.logger=INFO,console

實現

Flume到MongoDB

agent.sources = so1
agent.channels = c1
agent.sinks = s1

agent.sources.so1.type = avro
agent.sources.so1.bind = 0.0.0.0
agent.sources.so1.port = 44444
agent.sources.so1.channels = c1

agent.channels.c1.type = memory
agent.channels.c1.capacity = 1000
agent.channels.c1.transactionCapacity = 100

agent.sinks.s1.type = org.riderzen.flume.sink.MongoSink
agent.sinks.s1.host = localhost
agent.sinks.s1.port = 27017
agent.sinks.s1.model = single
agent.sinks.s1.db = test
agent.sinks.s1.collection = log
agent.sinks.s1.batch = 100
agent.sinks.s1.channel = c1

簡單說明：

1. 根據agent.sources.so1的相關配置，在44444埠開啟監聽 - log4j可以往這個埠寫日誌
2. 根據agent.sinks.s1的相關配置，so1接收到的數據，將寫到localhost:27017的test資料庫的log-collection中
   27017是MongoDB的預設監聽埠

Java到Flume

文件結構

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

文件內容

1. pom.xml

   <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
       <modelVersion>4.0.0</modelVersion>
       <groupId>cn.sinobest.asj</groupId>
       <artifactId>fluent-client-demo</artifactId>
       <version>0.0.1-SNAPSHOT</version>
       <dependencies>
           <dependency>
               <groupId>log4j</groupId>
               <artifactId>log4j</artifactId>
               <version>1.2.16</version>
           </dependency>
           <dependency>
               <groupId>commons-logging</groupId>
               <artifactId>commons-logging</artifactId>
               <version>1.1.1</version>
           </dependency>
           <dependency>
               <groupId>org.apache.flume.flume-ng-clients</groupId>
               <artifactId>flume-ng-log4jappender</artifactId>
               <version>1.6.0</version>
           </dependency>
       </dependencies>
   </project>

2. log4j.properties
   

   # 配置Log4jAppender，能寫日誌到Flume
   log4j.appender.flumeAvro=org.apache.flume.clients.log4jappender.Log4jAppender
   log4j.appender.flumeAvro.Hostname=localhost
   log4j.appender.flumeAvro.Port=44444
   log4j.appender.flumeAvro.UnsafeMode=true
   log4j.appender.flumeAvro.layout=org.apache.log4j.PatternLayout
   log4j.appender.flumeAvro.layout.ConversionPattern=%m
   # 配置根logger 此配置的日誌級別和appender會被所有的logger繼承
   log4j.rootLogger=INFO, flumeAvro

   註意： <1> org.apache.flume.clients.log4jappender.Log4jAppender負責將數據寫到Flume的具體實現 <2> 44444埠為前文Flume配置agent.sources.so1的監聽埠
3. LogDemo.java
   

   package cn.sinobest.flume.client.demo;
   import org.apache.commons.logging.Log;
   import org.apache.commons.logging.LogFactory;
   public class LogDemo {
       static final Log log = LogFactory.getLog(LogDemo.class);
       
       public static void main(String[] args) {
           log.info("{'name':'Adam', 'age':'26', 'skill':'reading'}");
       }
   }

   註意：確保日誌內容是json格式的字元串。

測試

1. 啟動MongoDB

   D:\develop\MongoDB\bin\mongod --dbpath=D:\develop\MongoDB\mydata

2. 啟動Flume
   

   # cd F:\temp\apache-flume-1.6.0-bin\bin
   flume-ng.cmd agent --conf ..\conf -f ..\conf\mongo-agent.properties -n agent

   說明：第1行命令是提醒確認當前所在路徑。
3. 運行LogDemo
4. 查看結果

   D:\tools\PowerCmd>D:\develop\MongoDB\bin\mongo
   2016-03-08T18:05:33.168+0800 I CONTROL  [main] Hotfix KB2731284 or later update is not installed, will zero-out data files
   MongoDB shell version: 3.2.3
   connecting to: test
   > db.log.find()
   { "_id" : ObjectId("56dea2b18449e36553652bc3"), "name" : "Adam", "age" : "26", "skill" : "reading" }
   > 

   說明： <1> mongo預設連接test資料庫，也可以使用mongo test的方式指定連接test資料庫。 <2> db.log.find()用來查看log-collection中的數據，可以看到數據已經寫入成功。

附錄

參考資料

1. Flume使用小結
   主要參考了環境搭建部分，其他部分不適合。
2. flume學習01-flume介紹
   博主寫了一系列，共3篇文章來介紹Flume，這裡只貼出第一篇，算是入門文章。後面有些內容沒有研究。
3. Flume 1.5日誌採集並存入mongodb的安裝搭建
   主要參考了Flume配置文件中，MongoSink的配置 - 後來發現博主沒有指定db，數據是寫到events資料庫的events-collection中