flume集成kafka

一、安装flume

请参考：flume安装

二、安装kafka

请参考：kafka安装

三、flume发送数据到kafka

3.1 配置

vim /usr/local/flume/mytest/3-kafka.properties

a3.sources = s1
a3.channels = c1
a3.sinks = k1

a3.sources.s1.type=exec
a3.sources.s1.command=tail -F /tmp/temp.log
a3.sources.s1.channels=c1

a3.channels.c1.type=memory
a3.channels.c1.capacity=10000
a3.channels.c1.transactionCapacity=100
#设置Kafka接收器
a3.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink
#设置Kafka的broker地址和端口号
a3.sinks.k1.brokerList=kafka:9092
#设置Kafka的Topic
a3.sinks.k1.topic=test
#设置序列化方式
a3.sinks.k1.serializer.class=kafka.serializer.StringEncoder
a3.sinks.k1.channel=c1

把kafka主机名配置到flume对应的服务器上

echo -e '192.168.200.91 kafka' >> /etc/hosts && tail -1 /etc/hosts

3.2 启动

bin/flume-ng agent --conf conf --conf-file mytest/3-kafka.properties --name a3 -Dflume.root.logger=INFO,console

3.3 测试

模拟日志

echo `date +%F\ %T` >> /tmp/temp.log

查看kafka消费数据

cd /usr/local/kafka
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

常见问题

乱码问题

如果kafka和flume不在同一台服务器上，且服务器使用了不同的字符集
如：
[root@flume ~]# echo $LANG
zh_CN.GB18030
[root@kafka kafka]# echo $LANG
en_US.UTF-8
这种情况下，建议统一字符编码