004- Flume Source之Kafka

Posted 2021-04-30 BearData

Kafka Source是指Apache Kafka消费者从Kafka主题中读取消息，如果有多个Kafka Source运行，则可以将他们配置在同一个消费组里面。

对于线上业务系统来说，有的时候需要对大量的数据进行统计，如果直接将数据保存到本地文件（例如使用log4j）可能会拖慢线上系统。那么，最好的方式是将大量的数据通过jms（例如：kafka）发送到消息服务器，消息中间件后面再对接flume来完成数据统计等需求。

接下来，我们来介绍一下flume 的kafka source。

一，属性

二，配置

#

agent.sources = a1

#

agent.channels = c1

#

agent.sinks = k1

# define source

agent.sources.a1.type = org.apache.flume.source.kafka.KafkaSource

agent.sources.a1.channels = c1

agent.sources.a1.batchSize = 5000

agent.sources.a1.batchDurationMillis = 2000

agent.sources.a1.kafka.bootstrap.servers = bigdata000:6667,bigdata002:6667

agent.sources.a1.kafka.topics = beardata

agent.sources.a1.kafka.consumer.group.id = beardata

# define channel

agent.channels.c1.type = memory

# define sink

agent.sinks.k1.type = logger

agent.sinks.k1.maxBytesToLog = 1024

agent.sinks.k1.channel = c1

#bin/flume-ng agent --conf conf --conf-file conf/kafka-source.conf --name agent -Dflume.root.logger=INFO,console

三，测试

启动程序，并给kafka的beardata主题发送消息

我们在Flume控制台中可以看到输出了我们订阅的主题的消息

以上就是Flume Kafka Source，下一篇我们将介绍Flume 自定义 Source。

长按二维码，关注BearData