Stream processing: Flink, Spark Streaming, ksqlDB

В уроке 06 мы разобрали Kafka, log-based транспорт. Этот урок про то, что строится поверх этого транспорта, чтобы получать аналитику в реальном времени: агрегации, джойны, materialized views с миллисекундной задержкой. Это вотчина Flink, Spark Structured Streaming и ksqlDB.

Аудитория: прошёл урок 06 (Kafka, partitions, exactly-once), знаком с базовой Kafka-клиентской работой. Урок не учит Flink или Spark с нуля; он раскладывает по полкам общие концепции stream-processing (event time, watermarks, windowing, state, checkpoints), которые звучат одинаково в любом движке, и сравнивает три самых популярных подхода.

Главный практический результат: выбрать правильный движок под задачу и не пораниться о watermarks при первом столкновении с late events.

0. Что за зверь stream processing

Batch (Spark batch, ClickHouse SELECT): обрабатываем фиксированный dataset. Завершено, когда обработали всё.

Stream (Flink, Spark Streaming, ksqlDB): обрабатываем поток без конца. “Завершить” нельзя; ответ должен подаваться непрерывно, обновляться по мере прихода новых событий.

Это не просто “batch на маленьких порциях”. Stream-processing вводит две новые проблемы, которых в batch не существует:

1. Время не то, что кажется. У события есть event_time (когда оно произошло) и processing_time (когда мы его получили). Между ними может быть лаг минуты, часы или больше. Поэтому “минутное окно за 14:05”, это какое: события с timestamp 14:04-14:05 (event time) или события, которые прибыли в систему между 14:04-14:05 (processing time)?
2. Состояние нужно держать вечно. Batch-job стартует с пустым state, заканчивает, забывает. Stream-job может работать годами; state накапливается; падение процесса не должно потерять накопленный state.

Дальше разбираем эти две проблемы, потом смотрим, как их решают конкретные движки.

1. Event time vs processing time vs ingestion time

Три модели времени:

• Event time (событийное), момент, когда событие реально случилось. Хранится в payload ({ts: "2026-05-07T10:00:00Z", ...}). Источник истины для бизнес-логики.
• Processing time (процессорное), момент, когда событие попало в обработчик. Это локальное время процесса, “сейчас”.
• Ingestion time (вход в Kafka), момент записи в брокер. Промежуточный вариант: точнее processing-time, но не идеально точный.

Простой пример: пользователь сделал клик в офлайн-приложении в 10:00, но интернет появился только в 10:30. Событие пришло в 10:30, event_time = 10:00, processing_time = 10:30. Какой timestamp использовать?

• Бизнес-вопрос “сколько кликов было в 10:00 ± 30 секунд”, надо использовать event_time.
• Технический вопрос “сколько событий получили за последнюю минуту”, processing_time.

В data engineering event time, дефолт, потому что это единственный способ дать корректный отчёт по бизнесу. Processing time использовалось в ранних streaming-системах (Storm, ранний Spark Streaming), потому что было проще; современные системы (Flink, Spark Structured Streaming) фокусируются на event time.

1.1 Цена event time

Если все события приходят строго по порядку event_time, проблем нет. На практике события приходят out of order (вне порядка), потому что:

• Сетевая задержка переменная.
• Источники (мобильные клиенты, IoT) пишут с задержкой.
• Партицирование Kafka не гарантирует ordering между партициями.
• Replays и retry смешивают порядок.

Это значит: на момент времени T мы можем получить событие с event_time = T-5min. Если мы уже “закрыли” окно [T-6min, T-5min) и эмитили его агрегат, что делать с поздним событием?

Решает watermark.

2. Watermarks: promise of completeness

Watermark, это маркер времени, обещание: “все события с event_time меньше этого watermark уже прибыли (или не прибудут)”. Когда watermark проходит конец окна, окно можно “закрыть” и эмитить финальный результат.

Простейшая стратегия: watermark(t) = max(event_time among arrived) - maxOutOfOrderness. Параметр maxOutOfOrderness (5 секунд, 5 минут, 1 час) задаёт, насколько сильно мы верим в “опоздания”: события, отставшие больше maxOutOfOrderness, не учитываются.

2.1 Trade-off

• Низкий maxOutOfOrderness (5 секунд): быстрая эмиссия результатов, но больше late events дропается. Подходит для realtime dashboard, где скорость важнее полноты.
• Высокий maxOutOfOrderness (1 час): медленная эмиссия, но почти все события учтены. Подходит для billing, где полнота важнее скорости.
• Allowed lateness (специфичный параметр Flink/Spark): окно остаётся “soft-closed” дополнительный промежуток, и принимает дозапоздавшие события (агрегат обновляется). За пределами allowed lateness, событие падает в side output.

2.2 Демо: watermark в действии

Виджет проигрывает поток событий с заданным maxOutOfOrderness, windowSize и allowedLateness. Двигай слайдеры и нажимай play, смотри:

• Watermark (вертикальная цветная линия) ползёт за курсором “now”, отставая на maxOutOfOrderness.
• Окна закрашиваются, когда watermark проходит их конец + allowed lateness.
• Событие, прибывшее в уже окончательно закрытое окно (помечено красным), летит в side output.

Двигай windowSize вверх, увидишь, что окна закрываются позже, событий в каждом больше. Увеличивай maxOutOfOrderness, late events почти исчезают, но эмиссия результатов замедляется.

2.3 Watermark generation strategies

Apache Flink, например, даёт несколько стратегий:

• WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)), простая фиксированная out-of-orderness.
• WatermarkStrategy.forMonotonousTimestamps(), для отсортированных потоков (доверяем источнику).
• Custom strategy: смотреть процентиль или адаптивно подгонять.

В Spark Structured Streaming watermark задаётся одной строкой: .withWatermark("event_time", "10 minutes").

В ksqlDB watermark выводится из timestamp-поля сообщения автоматически (TIMESTAMP в DDL топика).

3. Windowing: tumbling, sliding, session

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

3.1 Tumbling window

Tumbling, фиксированный непересекающийся интервал. Окна [0, 5), [5, 10), [10, 15), …

-- Flink SQL
SELECT TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS minute,
       count(*) AS events
FROM clicks
GROUP BY TUMBLE(event_time, INTERVAL '1' MINUTE);

Используй для регулярной аналитики: counts per minute, average per hour.

3.2 Sliding window (hopping)

Sliding, окна с перекрытием. Размер 5 минут, шаг 1 минута, окна [0, 5), [1, 6), [2, 7), …

SELECT HOP_START(event_time, INTERVAL '1' MINUTE, INTERVAL '5' MINUTES) AS window_start,
       avg(latency_ms) AS avg_latency
FROM requests
GROUP BY HOP(event_time, INTERVAL '1' MINUTE, INTERVAL '5' MINUTES);

Каждое событие попадает в windowSize / slide окон одновременно. Полезно для скользящих метрик: средняя latency за последние 5 минут, обновляемая каждую минуту.

3.3 Session window

Session window, динамическое окно, закрывающееся по gap-таймауту. Если между событиями одного key больше N минут, новая сессия.

SELECT user_id,
       SESSION_START(event_time, INTERVAL '30' MINUTES) AS session_start,
       SESSION_END(event_time, INTERVAL '30' MINUTES) AS session_end,
       count(*) AS clicks
FROM clicks
GROUP BY user_id, SESSION(event_time, INTERVAL '30' MINUTES);

Используется для аналитики поведения: длительность сессии пользователя, события в одной сессии.

3.4 Global window

Global + custom trigger, низкоуровневый кирпичик. Все события в одном окне; trigger явно говорит, когда эмитить (например, по count, не по времени).

В реальной работе используются tumbling и session, sliding в специфичных метриках. Global, на крайний случай.

4. Stateful operators

Большинство интересных stream-операций требуют состояния: количество событий по ключу, последний sketch, сумма за окно, lookup table.

4.1 Keyed state

Состояние, привязанное к ключу (обычно partition key). Каждый ключ имеет свой “слот”:

// Flink, ValueState<Long> для count по user_id
public class UserCountFn extends KeyedProcessFunction<String, Event, Tuple2<String, Long>> {
    private transient ValueState<Long> count;

    @Override
    public void open(Configuration cfg) {
        count = getRuntimeContext().getState(
            new ValueStateDescriptor<>("count", Long.class)
        );
    }

    @Override
    public void processElement(Event e, Context ctx, Collector<Tuple2<String, Long>> out) {
        Long current = count.value();
        if (current == null) current = 0L;
        count.update(current + 1);
        out.collect(Tuple2.of(e.userId, current + 1));
    }
}

Виды keyed state в Flink:

• ValueState<T>, одно значение.
• ListState<T>, список значений.
• MapState<K, V>, словарь.
• ReducingState / AggregatingState, инкрементальная агрегация.

В Spark Structured Streaming эквивалент, mapGroupsWithState / flatMapGroupsWithState. В ksqlDB state скрыт за SQL и держится в RocksDB на каждом инстансе.

4.2 Broadcast state

Состояние, реплицированное на все параллельные операторы. Используется для редко меняющихся справочников (например, список заблокированных пользователей, который обновляется раз в час и нужен каждому event-обработчику).

Реализуется как два потока: data stream (события) и broadcast stream (обновления справочника). Broadcast state синхронизирован между tasks.

5. State backend: RocksDB, in-memory

Где физически живёт state.

5.1 In-memory state backend

State в Java heap. Быстрый доступ (наносекунды), но размер ограничен heap-ом (несколько GB). Не подходит для больших state-ов (миллионы ключей с сложными values).

В Flink, дефолт до 1.13: MemoryStateBackend (state в heap, snapshots в JobManager) или FsStateBackend (state в heap, snapshots на distributed FS). Сейчас оба deprecated в пользу унифицированного HashMapStateBackend.

5.2 RocksDB state backend

State в RocksDB на локальном диске. Объём ограничен диском (терабайты), доступ медленнее (микросекунды против наносекунд in-memory).

Это дефолт для production в Flink. Без RocksDB большие jobs не помещаются в память; с RocksDB можно держать миллионы keys и десятки GB state-а.

В ksqlDB и Kafka Streams RocksDB всегда дефолт; альтернативы нет.

5.3 Snapshot и recovery

Локальный state на узле, не отказоустойчивый сам по себе: если узел упадёт, state потеряется. Решает checkpointing: периодически state снапшотится в distributed storage (S3, HDFS).