Как запустить задания потока воздуха на основе завершения потоковой передачи для разделов?

Question

У меня есть потоковое задание flink, которое читает из Kafka и записывает в соответствующие разделы файловой системы. Например, задание сконфигурировано для использования приемника, который записывает в /data/date=$ndomdatecasts/hour=$ndomhour rout.

Как определить, что раздел готов к использованию, чтобы соответствующий конвейер воздушного потока мог выполнить некоторую пакетную обработку в дополнение к этому часу?

Answer 1

Вы можете посмотреть на реализацию ContinuousFileMonitoringSource , чтобы увидеть, как он контролирует файловую систему. А затем сделайте что-то похожее на то, что Дэвид Андерсон предложил в своем другом вопросе, заново создав пользовательскую ProcessFunction.