Flume KafkaChannel的使用

Flume提供了Kafka Sink将数据写入Kafka，也提供了KafkaChannel使用Kafka作为Channel存储数据。

使用KafkaChannel既可以避免Memory Channel的内存溢出，又可以提供不错的性能。

另外当我们需要将数据写入Kafka时，可以不需要Sink，直接使用KafkaChannel就可以将数据写入Kafka，省去了Sink的开销。

以下简单介绍下KafkaChannel的使用。Flume的版本是1.9.0。

需要注意的几点

通常，与channel相关的配置值应用于channel配置级别，例如：a1.channel.k1.type =
与Kafka相关的配置值以"kafka"前缀
生产者/消费者特定的属性以kafka.producer或kafka.consumer为前缀
尽可能使用Kafka参数名称。尽量使用Kafka自身的参数配置。

参数

加粗的代表是必需的。

参数	默认值	描述
type	–	组件名称。必须为`org.apache.flume.channel.kafka.KafkaChannel`
kafka.bootstrap.servers	–	kafka的broker list，建议至少两个。格式是用逗号分隔的主机名：端口列表
kafka.topic	flume-channel	channel将写入的kafka topic
kafka.consumer.group.id	flume	供sink消费的消费者组
parseAsFlumeEvent	true	是否转换为flume的事件格式。
pollTimeout	500	调用poll的超时时间
defaultPartitionId	–	channel默认写入的partitionId。如果没有指定，只是用kafka producer默认的分区器。
partitionIdHeader	–	设置后，生产者将从事件头中获取使用该属性值命名的字段的值，并将消息发送到主题的指定分区。如果该值表示无效分区，则该事件将不被接受到通道中。如果存在头值，则此设置将覆盖defaultPartitionId。
kafka.consumer.auto.offset.reset	latest	sink首次消费channel时的策略
kafka.producer.security.protocol	PLAINTEXT	Set to SASL_PLAINTEXT, SASL_SSL or SSL if writing to Kafka using some level of security. See below for additional info on secure setup.
kafka.consumer.security.protocol	PLAINTEXT	Same as kafka.producer.security.protocol but for reading/consuming from Kafka.
more producer/consumer security props		If using SASL_PLAINTEXT, SASL_SSL or SSL refer to Kafka security for additional properties that need to be set on producer/consumer.

以上参数是Flume所提供的。

仅有上述参数，是不够的，Flume支持我们去配置Kafka自身的生产者和消费者的参数。

分别通过kafka.producer.参数名、kafka.consumer.参数名来配置KafkaChannel作为生产者和消费者时候的参数。

上游Source写入KafkaChannel时，用的是生产者的参数；KafkaChannel写入下游Sink时，用的是消费者的参数。

配置案例

myagent.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
myagent.channels.c1.kafka.bootstrap.servers = sdw4:9092,sdw5:9092,sdw6:9092
myagent.channels.c1.useFlumeEventFormat=false
myagent.channels.c1.kafka.topic = hardware-log
# 配置生产者的ack为1
myagent.channels.c1.kafka.producer.acks = 1
# 配置生产者的批大小为100000
myagent.channels.c1.kafka.producer.batch.size = 100000
myagent.channels.c1.kafka.producer.linger.ms = 5000

参考

Flume KafkaChannel文档 http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html#kafka-channel

Kafka生产者的配置：kafka-clients包中的org.apache.kafka.clients.producer.ProducerConfig

Kafka消费者的配置：kafka-clients包中的org.apache.kafka.clients.consumer.ConsumerConfig