Flink kafka - Flink作业不向不同分区发送消息

Question

我的配置如下：

1. 一个有两个分区的kafka主题
2. 一个动物园管理员实例
3. 卡夫卡一个例子
4. 两个同组身份的消费者

Flink作业片段：

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

场景1:

我在eclipse上写了一个flink作业(生产者)，它从文件夹中读取一个文件，并将msgs放在kafka主题上。

因此，当我使用eclipse运行这段代码时，它工作得很好。

例如， : --如果我放置一个记录为100条的文件，flink会向分区1发送少量的msgs &将少数msgs发送到分区2，因此两个使用者都会得到很少的msg。

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

我希望场景1使用场景2中创建的jar。

Answer 1

如果您没有提供一个FlinkKafkaPartitioner，或者没有明确地说要使用卡夫卡的FlinkFixedPartitioner，就会使用一个FlinkFixedPartitioner，这意味着来自一个任务的所有事件都将在同一个分区中结束。

要使用Kafka的除法器，请使用以下命令：

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

从IDE运行和eclipse运行之间的区别可能是由于在Flink中并行性或分区设置的不同。

Answer 2

对于Flink-Kafka生产者，添加"null“作为最后一个参数。

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

对此的简短解释是，这将关闭Flink，使其无法使用默认的分区程序FlinkFixedPartitioner。这是关闭的，因为默认将允许Kafka在它认为合适的分区之间分发数据。如果没有关闭它，那么用于使用FlinkKafkaProducer的接收器的每个并行/任务槽只会写入每个并行/任务槽的一个分区。