1、环境配置：

平台版本：CDH 5.15.1

服务器资源：VM虚拟机三台（规格：CPU 2核，内存8G，硬盘300G）

组件版本：flume使用但是5.15.1自带的版本（1.6），kafka是CDH 5.15.1兼容版本。

部署方式：flume agent和kafka 部署在相同节点。

2、flume的配置

整个项目中之所以选用flume，是看中了flume强大灵活的日志收集功能。本文中CDH使用的是flume agent的部署方式，一个agent包含了source，channel和sink的三个部分组成（更加详细的请参考官网文档）。

flume agent基本模型

这里source我采用的是syslog，kafka作为sink，channel选用memory。

由于项目的开发环境使用时CDH 5.15.1，所以添加flume agent的配置很友好（其实还是vim修改配置文件比较顺手）。此处注意项：

代理名称（也是就是agent名称）需要和下面配置文件中的一致。

CDH环境的flume agent修改项

---

flume agent的详细配置如下：

# Please paste flume.conf here. Example:

# Sources, channels, and sinks are defined per

# agent name, in this case 'tier1'.

tier1.sources  = source1

tier1.channels = channel1

tier1.sinks    = sink1

# For each source, channel, and sink, set

# standard properties.

tier1.sources.source1.type    = syslogtcp

tier1.sources.source1.host    = cdh-3

tier1.sources.source1.port    = 5140

tier1.sources.source1.channels = channel1

#tier1.channels.channel1.type  = file

#tier1.channels.channel1.checkpointDir = /home/flume/channel/checkpoint

#tier1.channels.channel1.dataDirs = /home/flume/channel/data

tier1.channels.channel1.type  = memory

tier1.channels.channel1.capacity =10000

tier1.channels.channel1.transactionCapacity=10000

tier1.sinks.sink1.type        = org.apache.flume.sink.kafka.KafkaSink

tier1.sinks.sink1.kafka.topic = rawlog

tier1.sinks.sink1.kafka.bootstrap.servers = cdh-3:9092

tier1.sinks.sink1.kafka.flumeBatchSize = 2000

tier1.sinks.sink1.kafka.producer.acks = 1

tier1.sinks.sink1.channel      = channel1

此处tier1就是之前agent的名称，需保持一致。kafka sink的topic需要提前创建，flume是不会自动创建的。

3、功能测试

修改完flume agent的配置，重新启动下agent。agent会启动并监听5140端口，如果没有请排查下日志，看是否有其他问题。

为了能准确观察到数据进入kafka的状态，这里采用了KafkaOffsetMonitor来监控状态。KafkaOffsetMonitor是一个可以用于监控Kafka的Topic及Consumer消费状况的工具，其配置和使用特别的方便。源项目Github地址为：https://github.com/quantifind/KafkaOffsetMonitor。 

我们采用了如下的启动参数：

java -cp KafkaOffsetMonitor-assembly-0.2.1.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk 0.0.0.0:2181 --port 9999 --refresh 2.seconds --retain 2.days

如何发送syslog报文这里就不再描述了。进行了下简单测试，单节点agent采用如上的配置，基本上可以达到5k以上的入库性能。具体数据如图所示，注意我们只需观察offset和时间轴就可粗略估算。发送了20k条数据，4s后完全进入kafka。

发送前kafka offset是906370 发送完后offset是926369