Flume连接KAFKA

Posted 2023-03-04 是秋不是楸

前提工作：虚拟机有jdk环境，hadoop集群，flume安装包，Kafka安装包、

1.解压flume，Kafka安装包

[root@docker ~]# tar -zxvf /h3cu/apache-flume-1.9.0-bin.tar.gz -C /usr/local/src/
[root@docker ~]# tar -zxvf /h3cu/kafka_2.12-2.4.1.tgz -C /usr/local/src/

2.进入flume配置文件夹中

[root@docker ~]# cd /usr/local/src/flume/conf/
#将文件flume-env.sh.template改名为flume-env.sh
[root@docker conf]# mv flume-env.sh.template flume-env.sh
创建一份XXX.conf文件
touch flume.conf
vi flume.confa1.sources=r1

a1.sinks=k1
a1.channels=c1
a1.sources.r1.type=exec
a1.sources.r1.command=tail -F /usr/local/src/logs/web.log #需要采集的文本的路径
a1.sinks.k1.type=org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic=Huang #将数据发送到Kafka主题Huang
a1.sinks.k1.kafka.bootstrap.servers=docker:9092     #Kafka服务器地址为docker:9092
a1.sinks.ka.kafka.flumeBatchSize=20                 #Flume批处理大小为20
a1.sinks.k1.kafka.producer.acks=1                   #Kafka生产者确认为1
a1.sinks.k1.kafka.producer.linger.ms=1              #Kafka生产者延迟为1毫秒
a1.sinks.k1.kafka.producer.compression.type=snappy  #Kafka压缩类型为snappy
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transactionCapacity=100
a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

3.启动flume命令：

[root@docker conf]# flume-ng agent --conf /usr/local/src/flume/conf/ --name a1 --conf-file /usr/local/src/flume/conf/XXX.conf -Dflume.root.logger==INFO,console

4.在这途中可能会遇到jar包冲突的问题

将hadoop的guava-27.0-jre.jar移动到flume下面，并删除flume原来的jar包
[root@docker conf]# cp /usr/local/src/hadoop/share/hadoop/common/lib/guava-27.0-jre.jar /usr/local/src/flume/lib/

1. 配置kafka

[root@docker conf]# cd /usr/local/src/kafka/config/ #进入kafka的配置目录
[root@docker config]# vi server.properties
broker.id=0
listeners=PLAINTEXT://docker:9092
advertised.listeners=PLAINTEXT://docker:9092
zookeeper.connect=docker:2181

2. 启动kafka

kafka-server-start.sh config/server.properties 

3.创建主题

kafka-topics.sh --create --zookeeper docker:2181 --replication-factor 1 --partitions 1 --topic Huang

4.查看主题

kafka-topics.sh --list --zookeeper docker:2181

3. 消费主题，接收flume采集到的数据

kafka-console-consumer.sh --bootstrap-server docker:9092 --topic dblab --from-beginning