Недавно мы говорили про непрерывный мониторинг Flink-приложений и подробно рассмотрели метрики состояния и пропускной способности. В продолжение этой важной для разработчиков и дата-инженеров темы, сегодня рассмотрим, как идентифицировать временную задержку обработки данных.
Пользовательские метрики задержки в потоковых приложениях
Для потоковых приложений, которые обрабатывают события в режиме, близком к реальному времени, важно отслеживать водяные знаки в операторах, чувствительных ко времени операторах, например, оконных функциях, наблюдая за разницей между текущей временной меткой и водяным знаком. Если эта разница аномально велика, возможно, идет обработка более старых событий в рамках восстановления после сбоя или источник данных не отправлял watermark в течение длительного времени, став недоступным или бездействующим. В этом случае нужно найти обходной путь или исправить программный код.
Чтобы получить временную задержку события из встроенных метрик Flink, можно сравнить метку времени сбора значения метрики currentOutputWatermark с фактическим значением и установить соответствующее предупреждение. Это возможно сделать это на разных уровнях. К примеру, чтобы собрать информацию о том, как время события продвигается по графу задания, события можно сгруппировать по различным операторам задания. А если нужно проверить перекос во времени события среди подзадач, следует нанести на граф каждую подзадачу конкретного оператора.
Однако, недостатком работы со встроенными метриками является то, что во многих сторонних системах мониторинга, например, Prometheus, временные метки имеют точность только до секунды. Для сценариев реального времени с очень низкой задержкой этого недостаточно. Кроме того, такое определение задержки события также включает время, которое события тратят на ожидание во внешних системах вне Flink. К примеру, в топике Kafka или перед записью в него. Если нужно более детальное разрешение, чтобы смотреть только на время, затраченное на обработку во Flink, или отслеживать другую разновидность задержки обработки типа исключения задержки из асинхронного ввода-вывода, обработку поздних событий и пр., потребуется собственная индивидуальная метрика, отражающая нужное определение.
Для этого подойдет пользовательская метрика eventTimeLag, которая обновляется каждый раз при открытии окна. Использование гистограммы (histogram), которая измеряет статистические распределения, и может возвращать процентили, пригодится для управления уровнем доступности сервиса. Впрочем, перед тем как создавать собственную метрику задержки, следует посмотреть встроенные latency-маркеры Flink, которые больше похожи на инструмент отладки и специфичны по своему определению. Их мы рассмотрим далее.
Встроенные метрики Apache Flink
Flink позволяет отслеживать задержку записей, проходящих через систему. Эта функция отключена по умолчанию. Чтобы включить отслеживание задержки, надо установить для параметра latencyTrackingInterval положительное число в конфигурации Flink или в конфигурации исполнителя ExecutionConfig. В latencyTrackingInterval источники будут периодически создавать специальную запись, называемую LatencyMarker. Маркер содержит временную метку времени, когда запись была отправлена в источники. Маркеры задержки не могут превзойти обычные пользовательские записи, поэтому, если записи стоят в очереди перед оператором, это увеличит задержку, отслеживаемую маркером. Важно, что маркеры задержки не учитывают время, которое пользовательские записи проводят в операторах, поскольку они обходят их. В частности, маркеры не учитывают время, которое записи проводят, например, в оконных буферах. Если операторы не могут принимать новые записи, они стоят в очереди, что будет отражено в задержке, измеренной с помощью маркеров. LatencyMarkers используются для получения распределения задержки между источниками топологии и каждым нижестоящим оператором. Эти распределения представлены в виде метрик гистограммы. Степень детализации этих распределений можно контролировать в конфигурации Flink.
Для подзадачи с самой высокой степенью детализации Flink будет получать распределение задержки между каждой исходной подзадачей и каждой нисходящей подзадачей, что приводит к квадратичному с точки зрения параллелизма количеству гистограмм. В настоящее время Flink предполагает, что часы всех машин в кластере синхронизированы. Рекомендуется настроить службу автоматической синхронизации часов, например, NTP, чтобы избежать ложных результатов задержки. Важно помнить, что включение метрик задержки может значительно повлиять на производительность кластера, в частности, на степень детализации подзадач. Поэтому рекомендуется делать это только при отладке.
Flink также позволяет отслеживать задержку доступа к состоянию с ключом для стандартных или настраиваемых бэкендов состояния, которые расширяются от AbstractStateBackend. Эта функция отключена по умолчанию. Чтобы включить ее, следует установить для параметра state.backend.latency-track.keyed-state-enabled значение true в конфигурации Flink. Как только отслеживание задержки доступа к состоянию с ключом включено, Flink будет запрашивать задержку доступа к состоянию через каждые N обращений. Значение N определяется в конфигурации state.backend.latency-track.sample-interval и по умолчанию равно 100. Меньшее значение даст более точные результаты, но может снизить производительность, поскольку выборка данных выполняется чаще.
Поскольку типом этой метрики задержки является гистограмма, state.backend.latency-track.history-size будет контролировать максимальное количество записанных значений в истории, что по умолчанию равно 128. Большее значение этой конфигурации потребует больше памяти, но даст более точный результат.
В заключение отметим несколько показателей, специфических для коннекторов к внешним системам. К примеру, при чтении данных из Apache Kafka или AWS Kinesis, метрики records-lag-max и millisBehindLatest соответственно указывают, насколько далеко потребитель или группа потребителей отстают от первого сообщения в очереди. Метрика records-lag-max показывает максимальную задержку с точки зрения количества записей для любого раздела. Увеличение значения с течением времени — явное свидетельство, что группа потребителей не поспевает за продюсерами. Метрика millisBehindLatest показывает количество миллисекунд, в течение которых потребитель отстает от головы потока (первого сообщения в очереди). Для любого потребителя и сегмента Kinesis это указывает, насколько они отстают от текущего времени. Метрика конкретного сегмента может быть указана по имени потока и идентификатору сегмента. Значение 0 указывает на то, что обработка записей завершена, и в данный момент нет новых записей для обработки. Значение -1 указывает на то, что для метрики пока нет отчетного значения. Для удобства Flink перенаправляет эти метрики коннектора в свою систему метрик. Читайте в нашей следующей статье про мониторинг системных метрик JVM и RocksDB во Flink-приложениях.
Больше подробностей про использование возможностей Apache Flink для потоковой обработки событий в распределенных приложениях аналитики больших данных вы узнаете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
