使用 flink 从多个代理 kafka 读取

Question

我想从flink读取多个kafka。

我有 3 台用于 kafka 的计算机集群。有以下话题

Topic:myTopic   PartitionCount:3    ReplicationFactor:1 Configs:
Topic: myTopic  Partition: 0    Leader: 2   Replicas: 2 Isr: 2
Topic: myTopic  Partition: 1    Leader: 0   Replicas: 0 Isr: 0
Topic: myTopic  Partition: 2    Leader: 1   Replicas: 1 Isr: 1

我从 Flink 执行以下代码：

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "x.x.x.x:9092,x.x.x.x:9092,x.x.x.x:9092");
properties.setProperty("group.id", "flink");

DataStream<T> stream = env.addSource(new FlinkKafkaConsumer09<>("myTopic", new SimpleStringSchema(), properties)
stream.map(....)
env.execute()

我启动了 3 次相同的工作。

如果我使用一个代理执行此代码，它运行良好，但有 3 个损坏（在 3 台不同的机器上）只有一个分区被读取。

(In this question)提出的解决方案是

为每个集群创建单独的 FlinkKafkaConsumer 实例（这就是您已经在做的），然后合并生成的流

在我的情况下它不起作用。

所以我的问题是：

1. 我错过了什么吗？
2. 如果我们在 Kafka 集群中有一台新计算机，我们是否需要更改 flink 的代码来为新的 borker 添加消费者？或者我们可以在运行时自动处理吗？

Answer 1

您似乎误解了 Kafka 的分布式流的概念。

Kafka 主题由多个分区（在您的情况下为 3 个）组成。每个消费者都可以使用这些分区中的一个或多个。如果您使用相同的 group.id 启动应用程序的 3 个实例，则每个使用者实际上只会从一个代理读取数据 - 它会尝试平均分配负载，因此每个使用者只有一个分区。

我建议阅读更多关于这个主题的内容，尤其是关于Kafka documentation 中消费者群体的概念。

无论如何FlinkKafkaConsumer09 可以在多个并行实例中运行，每个实例将从一个或多个 Kafka 分区中提取数据。您无需担心创建更多消费者实例。一个消费者实例可以从所有分区中提取记录。

我不知道您为什么要开始工作 3 次而不是一次将并行度设置为 3。这将解决您的问题。

DataStream<T> stream =
      env.addSource(new FlinkKafkaConsumer09<>("myTopic", new SimpleStringSchema(), properties))
              .setParallelism(3);