大家好，又见面了，我是你们的朋友全栈君。
整合流程
Flume 发送数据到 Kafka 上主要是通过 KafkaSink 来实现的，主要步骤如下：
1. 启动Zookeeper和Kafka
这里启动一个单节点的 Kafka 作为测试：
# 启动Zookeeper
zkServer.sh start
# 启动kafka
bin/kafka-server-start.sh config/server.properties
2. 创建主题
创建一个主题 flume-kafka，之后 Flume 收集到的数据都会发到这个主题上：
# 创建主题
bin/kafka-topics.sh --create \
--zookeeper hadoop001:2181 \
--replication-factor 1 \
--partitions 1 --topic flume-kafka
# 查看创建的主题
bin/kafka-topics.sh --zookeeper hadoop001:2181 --list
3. 启动kafka消费者
启动一个消费者，监听我们刚才创建的 flume-kafka 主题：
# bin/kafka-console-consumer.sh --bootstrap-server hadoop001:9092 --topic flume-kafka
4. 配置Flume
新建配置文件 exec-memory-kafka.properties，文件内容如下。这里我们监听一个名为 kafka.log 的文件，当文件内容有变化时，将新增加的内容发送到 Kafka 的 flume-kafka 主题上。
a1.sources = s1
a1.channels = c1
a1.sinks = k1
a1.sources.s1.type=exec
a1.sources.s1.command=tail -F /tmp/kafka.log
a1.sources.s1.channels=c1
#设置Kafka接收器
a1.sinks.k1.type= org.apache.flume.sink.kafka.KafkaSink
#设置Kafka地址
a1.sinks.k1.brokerList=hadoop001:9092
#设置发送到Kafka上的主题
a1.sinks.k1.topic=flume-kafka
#设置序列化方式
a1.sinks.k1.serializer.class=kafka.serializer.StringEncoder
a1.sinks.k1.channel=c1
a1.channels.c1.type=memory
a1.channels.c1.capacity=10000
a1.channels.c1.transactionCapacity=100
5. 启动Flume
flume-ng agent \
--conf conf \
--conf-file /usr/app/apache-flume-1.6.0-cdh5.15.2-bin/examples/exec-memory-kafka.properties \
--name a1 -Dflume.root.logger=INFO,console
6. 测试
向监听的 /tmp/kafka.log 文件中追加内容，查看 Kafka 消费者的输出：
可以看到 flume-kafka 主题的消费端已经收到了对应的消息：
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请联系我们举报，一经查实，本站将立刻删除。
发布者：全栈程序员-站长，转载请注明出处：https://javaforall.net/152401.html原文链接：https://javaforall.net
