如何在达到特定大小（128 Mb）时将Kafka消息提交到HDFS接收器

Question

我的配置：Confluent（5.0.0）Kafka生成一些avro消息。 Connect worker（HDFS连接器接收器）将这些消息以Parquet格式传输到HDFS节点。我将connect worker配置为每5000条消息（flush.size=5000）向HDFS提交消息。这个配置工作正常。

我的问题：是否有任何解决方法提交消息达到128 Mb（或256 Mb），而不是消息数量？

我的HDFS连接器配置文件：

name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
format.class=io.confluent.connect.hdfs.parquet.ParquetFormat
tasks.max=1
topics=some_topic
hdfs.url=hdfs://hdfshost:8020/user/someuser/kafka_hdfs_sink/
flush.size=5000

Answer 1

没有这样的配置 - see this open issue

解决方法是平均知道每个消息对于主题的大小（在单个Kafka分区中，因为这是文件的编写方式），然后相应地设置flush.size以大致达到HDFS块大小的因子。

如果您使用TimeBasedPartitioner，则必须知道邮件数量或邮件达到目标尺寸的时间。