Чтение из нескольких брокеров kafka с флинком

Question

Я хочу читать несколько кафка из флинка.

У меня есть кластер из 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 

От Флинка я выполнить следующий код:

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, нам нужно изменить код флинка, чтобы добавить потребителя для нового borker? Или мы можем справиться с этим автоматически во время выполнения?

Answer 1

Возможно, вы неправильно поняли понятие распределенных потоков Кафки.

Тема Кафки состоит из нескольких разделов (3 в вашем случае). Каждый потребитель может потреблять один или несколько из этих разделов. Если вы запустите 3 экземпляра своего приложения с тем же group.id, каждый потребитель действительно прочитает данные только от одного брокера –, он попытается распределить нагрузку равномерно, так что это один раздел для каждого потребителя.

Рекомендую прочитать больше об этой теме, особенно о концепции групп потребителей в Kafka documentation.

В любом случае FlinkKafkaConsumer09 может работать в нескольких параллельных экземплярах, каждый из которых будет извлекать данные из одного или нескольких разделов Kafka. Вам не нужно беспокоиться о создании дополнительных экземпляров потребителя. Один экземпляр потребителя может извлекать записи из всех разделов.

Я понятия не имею, почему вы начинаете работу 3 раза вместо одного с параллелизмом, установленным в 3. Это решит вашу проблему.

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