2023学年完整公开课版flume整合kafka

Flume整合kafka2知识目标创建主题配置Flume测试01能力目标启动zookeeper和kafka掌握创建整合流程配置文件并启动测试02学习目标3目录01启动zookeeper和kafka02创建主题并启动消费者03配置Flume04测试启动zookeeper和kafka4Flume发送数据到Kafka上主要是通过

KafkaSink

来实现的，主要步骤如下：这里启动一个单节点的Kafka作为测试：#启动ZookeeperzkServer.shstart

#启动kafkabin/kafka-server-start.shconfig/perties创建主题5创建一个主题

flume-kafka，之后Flume收集到的数据都会发到这个主题上：#创建主题bin/kafka-topics.sh--create\--zookeeperhadoop001:2181\--replication-factor1\--partitions1--topicflume-kafka查看创建的主题bin/kafka-topics.sh--zookeeperhadoop001:2181--list启动kafka消费者6启动一个消费者，监听我们刚才创建的

flume-kafka

主题：bin/kafka-console-consumer.sh--bootstrap-serverhadoop001:9092--topicflume-kafka配置Flume7

新建配置文件

perties，文件内容如下。这里我们监听一个名为

kafka.log

的文件，当文件内容有变化时，将新增加的内容发送到Kafka的

flume-kafka

主题上。a1.sources=s1a1.channels=c1a1.sinks=k1

a1.sources.s1.type=execmand=tail-F/tmp/kafka.loga1.sources.s1.channels=c1配置Flume8#设置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.StringEncodera1.sinks.k1.channel=c1

a1.channels.c1.type=memorya1.channels.c1.capacity=10000a1.channels.c1.transactionCapacity=100启动Flume9flume-ngagent\--confconf\--conf-file/usr/app/apache-flume-1.6.0-cdh5.15.2-bin/examples/perties\--namea1-Dflume.root.logger=INFO,console启动Flumagenta1服务端10

命令如下：flume-ngagent-c.-fspoolDirForFileToHdfs.conf-na1-Dflume.root.logger=INFO,console

使用cp命令向SpoolingDirectory中发送数据：cpstudent.txt/LogForFlumeCollect/测试11

向监听的

/tmp/kafka.log

文件中追加内容，查看Kafka消费者