首页 文章

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

提问于
浏览 次
1

我有以下配置：

  • 一个包含2个分区的kafka主题

  • 一个zookeeper实例

  • 一个kafka实例

  • 具有相同组ID的两个消费者

Flink job snippet:

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

Scenario 1:

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

因此，当我使用eclipse运行此代码时，它工作正常 .

For example : 如果我放置一个包含100条记录的文件，flink会向分区1发送少量消息，并将少量消息发送到分区2，因此消费者都会获得很少的消息 .

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

我希望方案1使用方案2中创建的jar .

apache-kafka apache-flink kafka-producer-api

2 回答

  • 0

    如果你没有提供 FlinkKafkaPartitioner 或者没有明确说使用Kafka的那个，那么将使用 FlinkFixedPartitioner ，这意味着来自一个任务的所有事件都将在同一个分区中结束 .

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

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

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

    回复于
  • 0

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

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

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

    回复于

相关问题