Flume学习之路kafka sink参数介绍

参数名称默认值描述type-必须设置为org.apache.flume.sink.kafka.KafkaSink-Kafka-Sink将连接到的代理列表，以获取主题分区列表。这可以是部分代理列表，但是对于HA，我们建议至少两个。格式是用逗号分隔的主机名：端口列表kafka中将发布消息的主题。如果配置了此参数，则消息将发布到该主题。如果事件标题包含“主题”字段，则事件将发布到该主题，并覆盖此处配置的

怒上王者

2515人浏览 · 2021-03-02 19:47:25

怒上王者 · 2021-03-02 19:47:25 发布

kafka sink参数介绍

参数名称	默认值	描述
type	-	必须设置为org.apache.flume.sink.kafka.KafkaSink
kafka.bootstrap.servers	-	Kafka-Sink将连接到的代理列表，以获取主题分区列表。这可以是部分代理列表，但是对于HA，我们建议至少两个。格式是用逗号分隔的主机名：端口列表
kafka.topic	default-flume-topic	kafka中将发布消息的主题。如果配置了此参数，则消息将发布到该主题。如果事件标题包含“主题”字段，则事件将发布到该主题，并覆盖此处配置的主题。
flumeBatchSize	100	一批中要处理多少条消息。较大的批次可提高吞吐量，同时增加延迟。
kafka.producer.acks	1	在成功考虑一条消息之前，有多少个副本必须确认一条消息。接受的值为0（永远不等待确认），1（仅等待领导者），-1（等待所有副本）将其设置为-1，以避免在某些领导者失败的情况下丢失数据。
useFlumeEventFormat	false	默认情况下，事件直接从事件主体作为字节放入Kafka主题。设置为true可将事件存储为Flume Avro二进制格式。与KafkaSource上的相同属性或Kafka Channel上的parseAsFlumeEvent属性结合使用，将为生产方保留任何Flume标头。
defaultPartitionId	-	为该通道中要发送到的所有事件指定一个Kafka分区ID（整数），除非由partitionIdHeader覆盖。默认情况下，如果未设置此属性，则事件将由Kafka Producer的分区程序分发-包括通过键（如果指定）（或由kafka.partitioner.class指定的分区程序）分发。
partitionIdHeader	-	设置后，接收器将从事件头中获取使用此属性的值命名的字段的值，并将消息发送到主题的指定分区。如果该值表示无效分区，则将引发EventDeliveryException。如果存在标头值，则此设置将覆盖defaultPartitionId。
kafka.producer.security.protocol	PLAINTEXT	如果使用某种级别的安全性写入Kafka，则设置为SASL_PLAINTEXT，SASL_SSL或SSL。有关安全设置的其他信息，请参见下文。
more producer security props		如果使用SASL_PLAINTEXT，SASL_SSL或SSL，请参阅Kafka安全性以获取需要在生产者上设置的其他属性。
Other Kafka Producer Properties	-	这些属性用于配置Kafka Producer。可以使用Kafka支持的任何生产者属性。唯一的要求是在属性名称前添加前缀 kafka.producer。例如：kafka.producer.linger.ms

Note: Kafka Sink使用FlumeEvent标头中的主题和键属性将事件发送到Kafka。如果标题中存在主题，则事件将发送到该特定主题，从而覆盖为接收器配置的主题。如果标题中存在密钥，那么Kafka将使用该密钥在主题分区之间对数据进行分区。具有相同键的事件将被发送到同一分区。如果键为null，则事件将发送到随机分区。

Kafka接收器还提供了key.serializer（org.apache.kafka.common.serialization.StringSerializer）和value.serializer（org.apache.kafka.common.serialization.ByteArraySerializer）的默认值。不建议修改这些参数。

不推荐使用的属性：

参数名称	默认值	描述
brokerList	-	应使用kafka.bootstrap.servers
topic	default-flume-topic	使用 kafka.topic
batchSize	100	使用kafka.flumeBatchSize
requiredAcks	1	使用kafka.producer.acks

样例

a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = mytopic
a1.sinks.k1.kafka.bootstrap.servers = localhost:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.ki.kafka.producer.compression.type = snappy