Watermark: время события и порядок в потоке данных

Прямой ответ

События потока приходят не в порядке возникновения: задержка доставки, повторная отправка и несколько источников нарушают последовательность. Watermark — отметка, означающая «событий с временем раньше этой отметки больше не ожидается».

Два времени определяются независимо.

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

from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class WatermarkGenerator:
    max_seen: datetime | None = None
    out_of_orderness: timedelta = timedelta(seconds=30)

    def on_event(self, event_time: datetime) -> datetime | None:
        if self.max_seen is None or event_time > self.max_seen:
            self.max_seen = event_time
        return self.max_seen - self.out_of_orderness

Генератор отслеживает максимальное виденное время события и вычитает допуск на непорядок. Watermark отстаёт от максимального времени на величину допуска, поэтому события, пришедшие с небольшой задержкой, попадают в свои окна.

Отношение между watermark и окнами определяет момент закрытия окна; об оконной агрегации — в статье про оконную агрегацию потока.

Допуск на непорядок и его последствия

Величина допуска определяет, насколько late-события учитываются, и насколько задерживается результат.

Малый допуск. Watermark продвигается близко к максимальному времени события, результат формируется быстро, но события с задержкой больше допуска не попадают в окна.

Большой допуск. Учитываются события с существенной задержкой, но формирование результата откладывается на величину допуска.

Практическое следствие: допуск выбирается по распределению задержки доставки, а не произвольно. Измерение разницы между временем события и временем поступления показывает, какой процент событий укладывается в заданный допуск.

Три способа обработки событий, пришедших после закрытия окна.

Отбрасывание. Событие игнорируется, результат не меняется. Просто, но даёт неполный результат.

Допустимая задержка. Окно остаётся открытым дольше watermark и пересчитывается при позднем событии. Результат точнее, стоимость — пересчёт.

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

Отдельный аспект — простой потока. При отсутствии событий watermark не продвигается, поэтому окна не закрываются. Таймаут простоя продвигает watermark при длительном отсутствии событий, иначе результат не формируется.

Порядок событий внутри партиции гарантирован, но время события может убывать, поскольку задержка доставки различается; о партициях — в статье про распределение сообщений. О гарантиях порядка обработки — в статье про порядок обработки сообщений.

Практика: работа с временем события и watermark

1. Сохраняйте время события в самом событии. Отметка, добавленная при поступлении, отражает время обработки и не подходит для оконной агрегации.

2. Измеряйте распределение задержки доставки. Допуск на непорядок выбирается по проценту событий, укладывающихся в допуск.

3. Устанавливайте таймаут простоя. При отсутствии событий watermark не продвигается, поэтому окна не закрываются без таймаута.

4. Выбирайте способ обработки поздних событий явно. Отбрасывание, пересчёт либо отдельный поток — решение фиксируется.

5. Проверяйте долю поздних событий. Рост доли указывает на изменение задержки доставки и требует пересмотра допуска.

6. Используйте монотонный генератор watermark. Watermark не должен двигаться назад, поскольку закрытые окна не открываются повторно.

7. Различайте задержку доставки и задержку обработки. Первое влияет на допуск, второе — на своевременность результата.

Рекомендации:

1. Не используйте время обработки для оконной агрегации. Задержка доставки смещает событие в соседний интервал, что даёт расхождение с фактической последовательностью.
2. Учитывайте несколько источников событий. Каждый источник имеет собственную задержку, поэтому допуск определяется самым медленным.
3. Проверяйте поведение при всплеске задержки. Рост задержки доставки увеличивает долю поздних событий, и результат становится неполным без пересмотра допуска.
4. Мониторьте отставание watermark от системного времени. Разница показывает задержку формирования результата.
5. Документируйте выбранный допуск. Значение определяет полноту результата и своевременность, поэтому фиксируется вместе с обоснованием.

Практический сценарий: окна не закрывались в ночные часы

Контекст: показатель за час формировался вовремя днём и отсутствовал ночью, появляясь только утром.

1. Диагностика. Watermark не продвигался в период низкой интенсивности событий. Окна оставались открытыми, поскольку условие закрытия не выполнялось, и результат формировался только после поступления первых утренних событий.
2. Причина. Генератор watermark основывался на времени событий, а при отсутствии событий максимальное время не обновлялось. Таймаут простоя не был установлен.
3. Действия. (а) Установили таймаут простоя, продвигающий watermark при отсутствии событий дольше заданного интервала; (б) проверили, что продвижение по таймауту не приводит к преждевременному закрытию окон при временном затишье; (в) добавили метрику отставания watermark от системного времени для обнаружения остановки; (г) добавили оповещение при отставании сверх ожидаемого; (д) проверили формирование результата в период низкой нагрузки в тестах.
4. Результат. Окна закрываются по таймауту простоя, а результат формируется независимо от интенсивности потока событий.

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

FAQ

Что такое watermark? Отметка времени, означающая, что событий с временем раньше этой отметки больше не ожидается. Используется для закрытия окон.

Зачем нужен допуск на непорядок? События приходят с задержкой, поэтому watermark отстаёт от максимального времени события на величину допуска, позволяя поздним событиям попасть в свои окна.

Что произойдёт при отсутствии событий в потоке? Watermark не продвигается, и окна не закрываются. Таймаут простоя продвигает watermark при длительном отсутствии событий.

Можно ли использовать время обработки вместо времени события? Технически да, но задержка доставки смещает событие в соседний интервал, что даёт расхождение с фактической последовательностью возникновения.

Источники: Apache Flink time concepts, watermark generation, Kafka semantics — ссылки в шапке.