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的三个部分组成（更加详细的请参考官网文档）。
这里source我采用的是syslog，kafka作为sink，channel选用memory。
由于项目的开发环境使用时CDH 5.15.1，所以添加flume agent的配置很友好（其实还是vim修改配置文件比较顺手）。此处注意项：
代理名称（也是就是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。
