本文在【flume使用(二)：采集远程日志数据到MySql数据库】一文基础之上进行测试操作。本文使用到的：

flume版本、jdk版本、mysql、数据库表、javaBean、自定义的mysqlSink、以及多agent配置均同【flume使用(二)：采集远程日志数据到MySql数据库】

一、需求说明

采集远程系统上log4j实时生成的日志到本机上进行数据库持久化。

二、设计

数据流：

(1)log4j(配置文件配置flume输出)

----[通过avro协议传输]------->(2)一个agent(source:avro; channel:memory; sink:avro)

----[通过avro协议传输]------>(3)一个agent(source:avro; channel:memory; sink:自定义mysqlSink)

---[mysqlSink进行jdbc持久化]----->(4)mysql数据库

本测试为了方便一些：

(1)是在windows上直接eclipse执行；(可以打包让运行在linux机器202上)

(2)是linux机器202；

(3)是linux机器201；

(4)数据库在linux机器201上

三、实施

1.编写实时输出log4j日志程序

(1)新建maven测试项目：pom.xml

4.0.0

yichao.mym

flume-log4j

0.0.1-SNAPSHOT

log4j

log4j

1.2.16

org.apache.flume

flume-ng-core

1.7.0

org.apache.flume.flume-ng-clients

flume-ng-log4jappender

1.7.0

(2)编写log4j.properties

log4j.rootLogger=INFO, stdout, flume

log4j.logger.per.flume=INFO

### flume ###

log4j.appender.flume=org.apache.flume.clients.log4jappender.Log4jAppender

log4j.appender.flume.layout=org.apache.log4j.PatternLayout

log4j.appender.flume.Hostname=192.168.216.202

log4j.appender.flume.Port=44444

### stdout ###

log4j.appender.stdout=org.apache.log4j.ConsoleAppender

log4j.appender.stdout.Threshold=INFO

log4j.appender.stdout.Target=System.out

log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %c{1} [%p] %m%n

(3)编写日志打印程序

package yichao.mym.base.bean;

import org.apache.log4j.Logger;

public class WriteLog {

protected static final Logger logger = Logger.getLogger(WriteLog.class);

public static void main(String[] args) throws Exception {

int name = 0;

while (true) {

logger.info("log4j-flume-name-"+name+","+(name++));

Thread.sleep(1000);

}

}

}

注意这里打印的数据必须要能让mysqlSink.java能够解析。由于log4j输出数据格式问题，导致原本根据“,”拆分的字符不能正确解析，原因是输出的数据如： [ log4j-flume-name-1668,1668   ] 。这条数据前后都有间隔，导致第二个字段"1688  "不能被转换成整型。

解决方式：修改解析方式：mysqlSink.java中在string转int类型时，先对string进行去除空格处理

处理前：person.setAge(Integer.parseInt(content.substring(content.indexOf(",")+1)));

处理后：person.setAge(Integer.parseInt(content.substring(content.indexOf(",")+1).trim()));

目前文章flume使用(二)已经做了修正

2.编写采集log4j实时传输过来数据的flume的配置文件

此配置文件放到linux机器202中，而原linux机器201的conf配置文件不变仍然使用：avro-mysql.conf

文件名：log4jAvro-avro.conf

agent1.sources=source1

agent1.channels=channel1

agent1.sinks=mysqlSink

# describe/configure source1

agent1.sources.source1.type=avro

agent1.sources.source1.bind=192.168.216.202

agent1.sources.source1.port=44444

agent1.sources.source1.channels=channel1

# use a channel which buffers events in memory

# type:memory or file is to temporary to save buffer data which is sink using

agent1.channels.channel1.type=memory

agent1.channels.channel1.capacity=5000

agent1.channels.channel1.transactionCapacity=1000

agent1.sinks.mysqlSink.type=avro

agent1.sinks.mysqlSink.channel=channel1

agent1.sinks.mysqlSink.hostname=192.168.216.201

agent1.sinks.mysqlSink.port=4545

配置文件同样放在linux机器202的flume目录中的bin中

四、测试都进入flume中的bin目录下执行命令

1.启动linux机器201

./flume-ng agent -c ../conf -f ../conf/avro-mysql.conf -n agent1 -Dflume.root.logger=INFO,console

2.启动linux机器202(注意启动的配置文件)

./flume-ng agent -c ../conf -f ../conf/log4jAvro-avro.conf -n agent1 -Dflume.root.logger=INFO,console

3.在windows上执行编写好的实时输出log4j的程序

此时刷新mysql数据库表，即可看到数据正源源不断的插入进来：