В прошлом уроке мы говорили о batch — накопил данные за окно, прогнал, сохранил. У этой модели есть фундаментальное ограничение: между событием и его попаданием в аналитику проходит часы или сутки. Для отчётов CFO это нормально. Для обнаружения мошенничества, реагирования на сбои инфраструктуры, real-time recommendations — нет. В этих сценариях нужен другой шаблон — streaming.

Streaming обрабатывает события в момент их появления, без накопления в окно. Это другая инженерная модель с другими понятиями. Этот урок про базу — для глубины есть отдельные курсы (Kafka, Flink, Spark Streaming).

Что такое поток событий

Поток — это бесконечная последовательность сообщений, каждое из которых описывает одно событие. Сообщения приходят непрерывно, в произвольном темпе, в порядке, который не гарантируется. У потока нет «конца», в отличие от batch-окна.

В отличие от batch, тут нет «вчерашнего окна». Каждое событие проходит через систему за миллисекунды-секунды. Если в систему пришёл подозрительный заказ — алерт срабатывает в момент, а не утром следующего дня.

Event time vs processing time

Это первое и самое важное различие, которое отличает streaming от batch. У каждого события есть две временные метки:

Event time — когда событие реально произошло в источнике. Например, пользователь нажал кнопку «купить» в мобильном приложении в 14:23:15.

Processing time — когда событие попало в систему обработки. Например, через 30 секунд после клика, в 14:23:45 — мобильное приложение синхронизировалось с бэкендом.

В идеальном мире event time и processing time почти совпадают. В реальности — нет. Между ними бывают задержки от миллисекунд до часов:

• Мобильник был в самолёте — событие пришло через 4 часа после клика.
• Сетевая задержка между регионами — 500 ms между Европой и Сингапуром.
• Очередь Kafka переполнилась — consumer догоняет полчаса.

Большинство аналитических метрик считается по event time: «выручка за час 14:00-15:00» означает события, фактически произошедшие в этом интервале. Это правильный подход, но он требует, чтобы система умела ждать запоздалые события.

Windows в streaming

В streaming тоже есть окна, но они работают иначе, чем в batch. Окно создаётся динамически в процессе обработки, и в нём накапливаются события за определённый период.

Tumbling windows — непересекающиеся фиксированные окна. Например, окна по 1 минуте: [14:00, 14:01), [14:01, 14:02), и т.д. Каждое событие попадает ровно в одно окно по своему event time.

Hopping windows (или sliding windows в некоторых системах) — окна с фиксированной длиной и фиксированным шагом, могут перекрываться. Например, длина 5 минут, шаг 1 минута: [14:00, 14:05), [14:01, 14:06), [14:02, 14:07). Одно событие попадает в несколько окон.

Sliding windows — окна, которые движутся вместе с каждым событием. В Flink это специальный тип, используется для rolling-метрик.

Session windows — окна группируются по периодам активности. Если между событиями одного пользователя прошло меньше N минут — они в одной сессии. Если больше — началась новая сессия. Длина окна заранее неизвестна.

Watermarks

Главный вопрос streaming с event time: когда закрывать окно? Если окно [14:00, 14:05), можно ли публиковать результат в 14:05:00? А вдруг ещё придёт запоздалое событие с event time 14:04:30, которое сейчас в пути?

Эту проблему решают watermarks (водяные знаки). Watermark — это метка вида: «мы видели все события с event time меньше T». Это сигнал процессору: окна, чьё время полностью меньше T, можно безопасно закрывать и публиковать результат.

Watermark обычно отстаёт от текущего processing time на некоторое окно толерантности: например, «watermark = max(event_time) - 5 минут». Это значит: мы готовы ждать запоздалые события до 5 минут, потом публикуем результат.

Текущий event_time max:  14:08:00
Watermark:               14:03:00  (= max - 5 минут)
-> Окна [14:00, 14:02) и [14:02, 14:03) можно закрывать.
-> Событие с event_time 14:04 ещё может прийти — оно попадёт в открытое окно.

Что делать с событиями, пришедшими после закрытия окна? Тут несколько стратегий:

• Drop — игнорировать. Простейший вариант, теряем точность.
• Side output — отправлять в отдельный поток для специальной обработки (логирование, ручной разбор).
• Allowed lateness — держать окно «полузакрытым» ещё какое-то время, переоткрывая для запоздавших.
• Update mode — отправлять корректирующее событие (insert delta) в downstream.

Exactly-once

Третье ключевое понятие streaming — гарантии доставки. Возможны три уровня:

At-most-once. Сообщение обрабатывается 0 или 1 раз. Просто, но небезопасно: при сбое события теряются.

At-least-once. Сообщение обрабатывается минимум 1 раз, но при сбоях может быть обработано повторно. Безопасно от потерь, но возможны дубликаты. Самый частый дефолт в простых streaming-системах.

Exactly-once. Сообщение обрабатывается ровно 1 раз — никаких потерь, никаких дубликатов. Самая сильная гарантия и самая дорогая в реализации.

Exactly-once в streaming реализуется через комбинацию:

• Транзакционная запись в sink (Kafka transactions, two-phase commit в DWH).
• Idempotent producers на стороне отправителя.
• Checkpoints в процессоре — снэпшоты состояния, к которым можно откатиться при сбое.

Flink, Spark Structured Streaming, Kafka Streams все поддерживают exactly-once при определённых конфигурациях.

Простой пример: rolling count в Kafka + consumer

Чтобы было совсем конкретно, представим пайплайн: пользователи кликают на сайте, события улетают в Kafka, простой consumer считает количество кликов по странам за последние 5 минут.

Producer (на сайте, после клика):

producer.send('user-clicks', value={
    'user_id': user_id,
    'country': country,
    'page': page,
    'event_time': datetime.utcnow().isoformat()
})

Consumer (псевдо-Flink логика):

stream = env.from_kafka_topic('user-clicks')

stream \
    .key_by(lambda event: event['country']) \
    .window(TumblingEventTimeWindows.of(Time.minutes(5))) \
    .with_watermark(BoundedOutOfOrderness(Duration.of_minutes(2))) \
    .aggregate(lambda events: {'country': events[0]['country'], 'count': len(events)}) \
    .add_sink(KafkaSink('country-clicks-5min'))

Что тут происходит:

• Группируем события по country.
• Используем tumbling windows по 5 минут по event_time.
• Watermark — толерантность 2 минуты к запоздалым событиям.
• В каждом окне считаем count.
• Результат пишем в другой Kafka-топик.

Этот consumer работает 24/7 и каждые 5 минут публикует свежие агрегаты. Никакого ночного job-а, никакого batch — данные мгновенно отражают реальность.

Когда нужен streaming

Не всё нужно делать в streaming. Хорошие случаи:

• Фрод-детекция. Подозрительный платёж нужно отклонить за секунды.
• Realtime-метрики. Дашборд DevOps с задержкой 30 секунд для обнаружения сбоев.
• Streaming ETL. Простая чистка/обогащение данных «на пути» в DWH.
• Realtime recommendations. Обновить рекомендации в момент взаимодействия.

Плохие случаи для streaming:

• Тяжёлая аналитика по большим окнам. Месячные отчёты лучше делать batch — streaming на месячных окнах сложен и нестабилен.
• Сценарии с late-arriving в дни. Если данные часто опаздывают на сутки, watermark будет огромным, streaming становится бессмысленным.
• Простая аналитика без требований к свежести. Если CFO смотрит отчёт раз в день, не надо ради этого городить streaming.

Попробуй сам

Возьми любой продукт, которым пользуешься. Подумай, какие там процессы должны работать в streaming, а какие — в batch. Push-уведомление о подозрительном входе в аккаунт — streaming. Месячный отчёт по подпискам — batch. Recommendations на главной — может быть и то, и другое. Подумай, что определяет выбор. Обычно это бизнес-требование к латентности и стоимость инфраструктуры.