Как время контрольной точки мигания связано с размером выравнивания буфера или временем выравнивания?

Question

Мое задание на потоковую съемку имеет время проверки 2-3 секунды (15-20% времени) и 3-4 минуты (8-12% времени) и в среднем 2 минуты. У нас есть два оператора с состоянием. Первый - это потребитель kafka в качестве источника (FlinkKafkaConsumer010), а другой - приемник hdfs (CustomBucketingSink). Эти два состояния составляют 1-1,5 ГБ для точек сохранения и 800 МБ-6 ГБ (в среднем 3 ГБ) для контрольной точки. У нас есть 30se c окна обработки. Продолжительность контрольной точки и минимальная пауза между двумя контрольными точками - 3 минуты. Моя работа потребляет в среднем около 3 миллионов записей в минуту и ​​около 20 миллионов записей в минуту в пиковое время. Процессора и памяти более чем достаточно.

Вот мои сомнения:

1) Даже когда немногие размеры состояний контрольных точек меньше (на 70-80% меньше) по сравнению с другими состояние контрольной точки, это занимает минуты (15-20% времени) по сравнению с другим, которое занимает 5-10 секунд.

2) Размер выравнивания буфера иногда увеличивается до 7-8 ГБ по сравнению со средним значением 800 МБ-1 ГБ. но время контрольной точки не зависит от этого. Я предполагаю, что это займет больше времени, так как должно ждать барьер контрольной точки.

3) Будет ли затронуто время контрпунктирования, если мы увеличим размер падающего окна. Я считаю, что это не должно влиять ни на время сохранения, ни на время проверки.

4) Немногие из подзадач, которые впадают в hdfs, занимают 2-3 минуты (время 5-10%). Так что пока 98% подзадач выполняются за 30-50 секунд. 1-2 (95% времени, это только одна) подзадачи занимает 2-3 минуты. Что задерживает все время контрольных точек. Проблема не в узле, на котором выполняются эти подзадачи, потому что это иногда происходит с каким-то узлом, а иногда и с другим узлом.

5) Мы получаем одно исключение раз в 6-8 часов, которое перезапускает задание , TimerException {java .nio.channels.ClosedByInterruptException} в org. apache .flink.streaming.runtime.tasks.SystemProcessingTimeService $ TriggerTask.run (SystemProcessingTimeService. java: 288)

* 1014 6) чтобы минимизировать время буфера выравнивания.

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

8) Каждый раз, когда мы перезапускаем задание, все подзадачи занимают единое время в течение 2-3 дней на всех узлах. но затем 1-2 подзадачи занимают 2-3 минуты по сравнению с другими, которые занимают 15-30 секунд. Я могу ошибаться в этом поведении, но, насколько я заметил, это тоже случай.

Answer 1

Обратите внимание, что windows с состоянием, и если вы не выполняете инкрементное агрегирование, более длинное windows имеет больше состояний, что, в свою очередь, повлияет на размеры и длительность контрольных точек.

Было бы полезно узнать, какие состояние бэкенда, которое вы используете, и то, используете ли вы инкрементную контрольную точку или нет.

Я бы начал с попытки найти причину медленной подзадачи (ей), вызывающей обратное давление, которое, в свою очередь, вызывает болезненные контрольная точка. Например, может быть перекос данных или истощение ресурсов. Некоторые распространенные причины включают недостаточную пропускную способность ЦП, сети или диска или ограничения скорости AWS (или другого API). Например, может показаться, что у вас много процессорного времени, но одна горячая клавиша может слишком сильно нагружать один поток и, таким образом, сдерживать весь кластер.

Если вы найдете способ исправить дисбаланс в раковине, тогда проблемы выравнивания контрольной точки должны успокоиться. (Обратите внимание, что если вы можете допускать повторяющиеся результаты, вы можете отключить выравнивание контрольных точек, выбрав CheckpointingMode.AT_LEAST_ONCE.)