Flume学习应用：Java写日志数据到MongoDB

概述

Windows平台：Java写日志到Flume，Flume最终把日志写到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: empapache-flume-1.6.0-bin）
MongoDB:3.2.3
下载mongodb-win32-x86_64-3.2.3-signed.msi，并安装到合适的目录（e.g. D:developMongoDB）

辅助资源下载

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下。

初始化

数据初始化

为MongoDB指定dbpath。

新建一个目录，作为dbpath
e.g. D:developMongoDBmydata

启动监听服务 - CMD运行

D:developMongoDBinmongod --dbpath=D:developMongoDBmydata

配置Flume

生成flume-env.ps1
在安装目录的conf/下，复制flume-env.ps1.template，重命名为flume-env.ps1
修改log4j.properties
在安装目录的conf/下，修改文件log4j.properties；修改flume.root.logger=INFO,console

实现

Flume到MongoDB

在Flume安装目录的conf下，添加mongo-agent.properties：

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

简单说明：

根据agent.sources.so1的相关配置，在44444端口开启监听 - log4j可以往这个端口写日志
根据agent.sinks.s1的相关配置，so1接收到的数据，将写到localhost:27017的test数据库的log-collection中
27017是MongoDB的默认监听端口

Java到Flume

这是一个基于Maven的一个simple project，你可以从flume-ng-java2mongodb拿到源代码。

文件结构

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

文件内容

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>

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的监听端口

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格式的字符串。

测试

启动MongoDB

D:developMongoDBinmongod --dbpath=D:developMongoDBmydata

启动Flume

# cd F:	empapache-flume-1.6.0-binbin
flume-ng.cmd agent --conf ..conf -f ..confmongo-agent.properties -n agent

说明：第1行命令是提醒确认当前所在路径。

运行LogDemo

查看结果

D:	oolsPowerCmd>D:developMongoDBinmongo
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中的数据，可以看到数据已经写入成功。

附录

参考资料

Flume使用小结
主要参考了环境搭建部分，其他部分不适合。
flume学习01-flume介绍
博主写了一系列，共3篇文章来介绍Flume，这里只贴出第一篇，算是入门文章。后面有些内容没有研究。
Flume 1.5日志采集并存入mongodb的安装搭建
主要参考了Flume配置文件中，MongoSink的配置 - 后来发现博主没有指定db，数据是写到events数据库的events-collection中