Flink: трансляция оператора цепочки

Question

Предположим, что у меня есть поток данных событий , и я хочу передать его оператору (rich) map (map1), который связан с другим оператором (rich) map (map2). Параллельность двух карт одинакова. То, что я хочу, - это чтобы каждый параллельный экземпляр map1 go выводился в один параллельный экземпляр map2 (т. Е. Нет трансляции между двумя картами). Вот что я сделал до сих пор, но я не уверен, что это логически правильно. Это нормально?

val trainedStream = events.broadcast.map(new Mapper1(...)).setParallelism(par)
trainedStream.startNewChain.map(new Mapper2(...)).setParallelism(par)

Дополнительный вопрос: является ли SubtaskIndex (полученный из RuntimeContext.getIndexOfThisSubtask) двух связанных друг с другом подзадач / параллельных экземпляров map1 и map2 одинаковыми? Есть ли способ проверить это?

код в Scala, но то же самое относится и к Java Я думаю

Answer 1

Цепочка происходит автоматически во Flink, когда это возможно. Итак, в вашем примере достаточно просто использовать

val trainedStream = events.broadcast.map(new Mapper1(...)).map(new Mapper2(...))

Тогда я бы установил параллелизм на env.

Кстати, вы уверены, что хотите транслировать события? A Datastream обрабатывается параллельно по умолчанию. очень необычно для широковещательных событий, так как они будут обрабатываться несколько раз в соответствии с параллелизмом.

Дополнительный вопрос: является ли SubtaskIndex (полученный из RuntimeContext.getIndexOfThisSubtask) двух цепочечные подзадачи / параллельные экземпляры map1 и map2 одинаковы? Есть ли способ проверить это?

Индекс подзадачи одинаков для цепочечных операторов, поскольку они находятся в одной задаче (следовательно, они не могут даже иметь разные индексы). Вы можете видеть, что цепочка прошла успешно, если у вас есть задача mapper1 -> mapper2.