Flink kafka - Flink 作业不将消息发送到不同的分区答案

【问题标题】：Flink kafka - Flink job not sending messages to different partitionsFlink kafka - Flink 作业不将消息发送到不同的分区
【发布时间】：2019-02-04 01:54:05
【问题描述】：

我有以下配置：

一个带有 2 个分区的 kafka 主题
一个 zookeeper 实例
一个 kafka 实例
具有相同组 ID 的两个消费者

Flink 作业 sn-p：

speStream.addSink(new FlinkKafkaProducer011(kafkaTopicName,new 
SimpleStringSchema(), props));

场景 1：

我在 Eclipse 上编写了一个 flink 作业（Producer），它从文件夹中读取文件并将 msgs 放在 kafka 主题上。

所以当我使用 eclipse 运行这段代码时，它工作正常。

例如：如果我放置一个包含 100 条记录的文件，flink 会向分区 1 发送少量消息，向分区 2 发送少量消息，因此两个消费者都会收到少量消息。

场景 2： 当我创建上述代码的 jar 并在 flink 服务器上运行它时，flink 将所有 msgs 发送到单个分区，因此只有一个消费者获得所有 msgs。

我想要使用场景 2 中创建的 jar 的场景 1。

【问题讨论】：

您确定您没有为消息使用相同的密钥吗？ Kafka 会将具有相同 key 的消息放在同一个分区中。
我正在使用默认分区程序，它将消息以循环方式进行。正如我所说，如果我通过 eclipse 运行它就可以正常工作

标签： apache-kafka apache-flink kafka-producer-api

【解决方案1】：

如果您不提供FlinkKafkaPartitioner 或未明确表示要使用Kafka 的FlinkFixedPartitioner，则将使用FlinkFixedPartitioner，这意味着来自一个任务的所有事件将最终在同一个分区中。

要使用 Kafka 的分区器，请使用此 ctor：

speStream.addSink(new FlinkKafkaProducer011(kafkaTopicName,new SimpleStringSchema(), props), Optional.empty());

从 IDE 和 eclipse 运行的区别可能是因为 Flink 中的并行或分区设置不同。

【讨论】：

【解决方案2】：

对于 Flink-Kafka Producers，添加“null”作为最后一个参数。

speStream.addSink(new FlinkKafkaProducer011(
    kafkaTopicName,
    new SimpleStringSchema(),
    props,
    (FlinkKafkaPartitioner) null)
);

对此的简短解释是，这会关闭 Flink 使用默认分区器FlinkFixedPartitioner。默认情况下关闭此功能将允许 Kafka 在其认为合适的情况下在其分区之间分配数据。如果这没有关闭，那么使用 FlinkKafkaProducer 的接收器的每个并行度/任务槽将只写入每个并行度/任务槽的一个分区。

【讨论】：