У меня есть сценарий, в котором мне нужно переместить записи из улья в кафку.Я не мог найти способ, где я мог бы напрямую добавить мойку Кафки для набора данных Flink.Поэтому я использовал обходной путь, при котором я вызываю преобразование карты для набора данных flink, а внутри функции карты я использую команду kafkaProducer.send () для данной записи.
Проблема, с которой я сталкиваюсь, заключается в том, что я неНе существует способа выполнить kafkaProducer.flush () на каждом рабочем узле, поэтому число записей, записанных в kafka, всегда немного меньше, чем количество записей в наборе данных.
Есть ли элегантный способ справиться с этим?В любом случае я могу добавить приемник Кафки в набор данных в Flink?Или способ вызвать kafkaProducer.flush () в качестве финализатора?