Оконная агрегация: скользящие и сессионные окна в потоке

Прямой ответ

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

Три типа окон покрывают основные задачи.

Кувыркающееся окно делит время на непересекающиеся интервалы:

|----окно 1----|----окно 2----|----окно 3----|
00:00        00:05         00:10         00:15

Скользящее окно задаётся размером и шагом. Размер 15 минут при шаге 5 минут даёт три перекрывающихся окна для каждого момента, поэтому событие участвует в трёх агрегатах.

# Скользящее окно: размер 15 минут, шаг 5 минут
window_size = timedelta(minutes=15)
slide = timedelta(minutes=5)

def windows_for(ts: datetime) -> list[tuple[datetime, datetime]]:
    start = ts.replace(minute=(ts.minute // 5) * 5, second=0, microsecond=0)
    return [(start - i * slide, start - i * slide + window_size) for i in range(3)]

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

Граница окна относится к событию по времени события, а не по времени обработки. Различие принципиально при задержке доставки; о порядке событий — в статье про watermark.

Состояние агрегата и закрытие окна

Агрегат накапливается в состоянии до закрытия окна, поэтому два вопроса определяют реализацию.

Хранение состояния. Промежуточные значения хранятся до закрытия окна. Объём состояния пропорционален числу активных окон, поэтому скользящие окна с малым шагом требуют больше состояния, чем кувыркающиеся.

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

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

Четыре следствия выбора типа окна.

Кувыркающееся окно даёт минимальный объём состояния и однозначную принадлежность события к интервалу. Подходит для периодической отчётности.

Скользящее окно даёт плавность показателя, но событие обрабатывается несколько раз, а объём состояния растёт пропорционально отношению размера к шагу.

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

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

О пакетной обработке того же набора данных — в статье про основы Spark: пакетный расчёт по завершённым интервалам не требует управления состоянием и допустимой задержкой, но даёт результат с задержкой.

Практика: реализация оконной агрегации

1. Выбирайте тип окна по задаче. Периодическая отчётность требует кувыркающегося окна, плавный показатель — скользящего, анализ поведения — сессионного.

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

3. Закрывайте окно по watermark. Закрытие по системному времени приводит к потере событий, пришедших с задержкой.

4. Задавайте допустимую задержку явно. События после закрытия окна либо отбрасываются, либо вызывают пересчёт, и выбор фиксируется.

5. Оценивайте объём состояния. Скользящее окно с малым шагом требует состояния в несколько раз больше кувыркающегося окна того же размера.

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

7. Сохраняйте промежуточные агрегаты. Пересчёт окна при позднем событии требует предыдущего значения.

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

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

Практический сценарий: расхождение оконного агрегата с отчётом

Контекст: показатель числа событий за час в реальном времени расходился с ежедневным отчётом на несколько процентов.

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

Вывод: группировка по времени обработки даёт расхождение при неравномерной задержке доставки. Переход на время события с watermark и допустимой задержкой устраняет расхождение, а сверка с пакетным расчётом подтверждает корректность.

FAQ

Чем скользящее окно отличается от кувыркающегося? Скользящее окно перекрывается с соседними, поэтому событие участвует в нескольких агрегатах. Кувыркающееся окно относится к событию однозначно.

Как определяется сессионное окно? Промежутком между событиями: новый интервал начинается после отсутствия событий дольше заданного промежутка.

Что делать с событиями, пришедшими после закрытия окна? Отбросить либо пересчитать агрегат. Выбор определяется допустимой точностью и стоимостью пересчёта.

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

Источники: Apache Flink watermarks, Kafka Streams DSL, PostgreSQL window functions — ссылки в шапке.