flume整合kafka

港控/mmm° 2022-05-27 06:55 199阅读 0赞

在很多实时分析的系统中，经常需要使用Flume+Kafka收集数据，然后Strom或者Spark Stream从Kafka取出数据实时分析。flume能接受多种来源的数据，Kafka具有高吞吐、高负载的特点，将两者结合起来就，这样既可以拥有Kafka的容错能力，和Flume的多种接口。  
前一篇是：log4j+flume的整合内容,感兴趣的小伙伴可以先去看看  
[https://blog.csdn.net/jdfk423/article/details/79728293][https_blog.csdn.net_jdfk423_article_details_79728293]

1.将flume和kafka的整合的jar放入到flume的lib文件下  
整合jar的下载地址:[https://download.csdn.net/download/jdfk423/10346396][https_download.csdn.net_download_jdfk423_10346396]

2.我们来配置flume的flume.conf文件  
因为我目前用这个集群是小伙伴用Ambari搭建的搭建Hadoop集群，安装后各组件的默认安装在/usr/hdp/目录。  
将flume的source的type是netcat，后面会用nc命令模拟数据写入，方便测试。flume的sink是kafka。

# Flume agent config
    
    tier1.sources=source1  
    tier1.channels=channel1  
    tier1.sinks=sink1  
    
    tier1.sources.source1.type=netcat  
    tier1.sources.source1.bind=0.0.0.0
    tier1.sources.source1.port=44444  
    tier1.sources.source1.channels=channel1  
    tier1.sources.source1.interceptors=i1
    tier1.sources.source1.interceptors.i1.type=timestamp
    tier1.sources.source1.interceptors.i1.preserveExisting=true
    
    tier1.channels.channel1.type=memory  
    tier1.channels.channel1.capacity=10000  
    tier1.channels.channel1.transactionCapacity=1000  
    tier1.channels.channel1.keep-alive=30  
    
    
    #type和partition值的类，前面已经将他们的jar导入到了flume的lib中
    #设置sink的类型为kafka
    tier1.sinks.sink1.type=org.apache.flume.plugins.KafkaSink
    #将sink和channel相连接
    tier1.sinks.sink1.channel=channel1  
    #填上装有kafka的机器的ip加上kafka的broker监听的端口
    tier1.sinks.sink1.metadata.broker.list=master:6667
    tier1.sinks.sink1.partitioner.class=org.apache.flume.plugins.SinglePartition
    tier1.sinks.sink1.partition.key=1
    tier1.sinks.sink1.serializer.class=kafka.serializer.StringEncoder
    tier1.sinks.sink1.request.required.acks=0
    tier1.sinks.sink1.max.message.size=1000000
    tier1.sinks.sink1.producer.type=async
    #设置编码
    tier1.sinkssink1.custom.encoding=UTF-8
    #topic的名字
    tier1.sinks.sink1.custom.topic.name=test

这样Flume+Kafka算是整合好了，我们现在先来测试一下  
3.先新建test的topic和新建一个消费者  
切到kafka的bin目录下执行下面的命令新建名为test的topic  
`bin/kafka-topics.sh --create --zookeeper master:2181 --replication-factor 2 --partitions 4 --topic test`  
2181是ZooKeeper 的默认端口。

再新建一个消费者 继续执行  
`bin/kafka-console-consumer.sh --zookeeper master:2181 --topic test --from-beginning`

4.使用nc的命令向flume的那台机器的44444端口写入数据  
新建一个终端执行nc命令(nc ip 端口号)  
`nc master 44444`  
提示连接成功后(Ncat: Connected to master:44444)便开始写数据，写完回车便会发送数据到flume中，flume再到kafka，过几秒中我们便可以在消费者那个终端中看到我们发送的数据，代表flume和kafka整合成功了。  
下面是发送数据和接收成功的截图:  
发送的数据  
![这里写图片描述][70]  
消费者消费的数据  
![这里写图片描述][70 1]

[https_blog.csdn.net_jdfk423_article_details_79728293]: https://blog.csdn.net/jdfk423/article/details/79728293
[https_download.csdn.net_download_jdfk423_10346396]: https://download.csdn.net/download/jdfk423/10346396
[70]: /images/20220527/b618b55c0b174e1f83b9b93afb2a91ff.png
[70 1]: /images/20220527/89ceaad26bf34fe2829645c56195b018.png