Блог

Главная - Вопросы по программированию - Водяной знак Flink

Водяной знак Flink

#apache-flink #flink-streaming

#apache-flink #потоковая передача flink

Вопрос:

В Flink я нашел 2 способа настройки водяного знака,

первый — это

 val env = StreamExecutionEnvironment.getExecutionEnvironment
env.getConfig.setAutoWatermarkInterval(5000)

второй

 env.addSource(
    new FlinkKafkaConsumer[...](...)
).assignTimestampsAndWatermarks(
   WatermarkStrategy.forBoundedOutOfOrderness[...](Duration.ofSeconds(10)).withTimestampAssigner(...)
)

Я хотел бы знать, что в конечном итоге вступит в силу.

Ответ №1:

Между этими двумя нет никакого конфликта — они имеют дело с отдельными проблемами. Все, что указано, вступит в силу.

Первый,

 env.getConfig.setAutoWatermarkInterval(5000)

указывает, как часто вы хотите, чтобы создавались водяные знаки (один водяной знак каждые 5000 мс). Если это не было указано, вместо этого будет использоваться значение по умолчанию 200 мс.

Второй,

 env.addSource(
    new FlinkKafkaConsumer[...](...)
).assignTimestampsAndWatermarks(
   WatermarkStrategy.forBoundedOutOfOrderness[...](Duration.ofSeconds(10)).withTimestampAssigner(...)
)

определяет детали того, как эти водяные знаки должны быть вычислены. Т.е. они должны быть сгенерированы с FlinkKafkaConsumer использованием BoundedOutOfOrderness стратегии с ограниченной задержкой в 10 секунд. WatermarkStrategy Также требуется назначитель метки времени.

Значение по умолчанию отсутствует WatermarkStrategy , поэтому требуется что-то вроде этого второго фрагмента кода, если вы хотите работать со временем события.

Комментарии:

1. Из исходного кода стратегии BoundedOutOfOrderness водяной знак, отправляемый в нисходящий поток, определяется ограниченной задержкой, которую я назначил здесь, которая составляет 10 секунд. Насколько я понимаю, если нижестоящий оператор не определил какую-либо стратегию водяных знаков, будет использоваться 5 секунд?

2. Нет, 5-секундный интервал измеряет что-то совершенно другое. Это определяет, как часто у WatermarkStrategy запрашивается создание водяного знака путем вызова его onPeriodicEmit метода.