本文在【flume使用(二)：采集遠(yuǎn)程日志數(shù)據(jù)到MySql數(shù)據(jù)庫(kù)】一文基礎(chǔ)之上進(jìn)行測(cè)試操作。本文使用到的：

flume版本、jdk版本、mysql、數(shù)據(jù)庫(kù)表、javaBean、自定義的mysqlSink、以及多agent配置均同【flume使用(二)：采集遠(yuǎn)程日志數(shù)據(jù)到MySql數(shù)據(jù)庫(kù)】

一、需求說(shuō)明

采集遠(yuǎn)程系統(tǒng)上log4j實(shí)時(shí)生成的日志到本機(jī)上進(jìn)行數(shù)據(jù)庫(kù)持久化。

二、設(shè)計(jì)

數(shù)據(jù)流：

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

----[通過(guò)avro協(xié)議傳輸]------->(2)一個(gè)agent(source:avro; channel:memory; sink:avro)

----[通過(guò)avro協(xié)議傳輸]------>(3)一個(gè)agent(source:avro; channel:memory; sink:自定義mysqlSink)

---[mysqlSink進(jìn)行jdbc持久化]----->(4)mysql數(shù)據(jù)庫(kù)

本測(cè)試為了方便一些：

(1)是在windows上直接eclipse執(zhí)行；(可以打包讓運(yùn)行在linux機(jī)器202上)

(2)是linux機(jī)器202；

(3)是linux機(jī)器201；

(4)數(shù)據(jù)庫(kù)在linux機(jī)器201上

三、實(shí)施

1.編寫實(shí)時(shí)輸出log4j日志程序

(1)新建maven測(cè)試項(xiàng)目：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);

}

}

}

注意這里打印的數(shù)據(jù)必須要能讓mysqlSink.java能夠解析。由于log4j輸出數(shù)據(jù)格式問(wèn)題，導(dǎo)致原本根據(jù)“,”拆分的字符不能正確解析，原因是輸出的數(shù)據(jù)如： [?log4j-flume-name-1668,1668?? ] 。這條數(shù)據(jù)前后都有間隔，導(dǎo)致第二個(gè)字段"1688? "不能被轉(zhuǎn)換成整型。

解決方式：修改解析方式：mysqlSink.java中在string轉(zhuǎn)int類型時(shí)，先對(duì)string進(jìn)行去除空格處理

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

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

目前文章flume使用(二)已經(jīng)做了修正

2.編寫采集log4j實(shí)時(shí)傳輸過(guò)來(lái)數(shù)據(jù)的flume的配置文件

此配置文件放到linux機(jī)器202中，而原linux機(jī)器201的conf配置文件不變?nèi)匀皇褂?#xff1a;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機(jī)器202的flume目錄中的bin中

四、測(cè)試都進(jìn)入flume中的bin目錄下執(zhí)行命令

1.啟動(dòng)linux機(jī)器201

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

2.啟動(dòng)linux機(jī)器202(注意啟動(dòng)的配置文件)

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

3.在windows上執(zhí)行編寫好的實(shí)時(shí)輸出log4j的程序

此時(shí)刷新mysql數(shù)據(jù)庫(kù)表，即可看到數(shù)據(jù)正源源不斷的插入進(jìn)來(lái)：

總結(jié)

以上是生活随笔為你收集整理的flume可以实时监控mysql嘛_flume使用（三）：实时log4j日志通过flume输出到MySql数据库...的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。