Модуль 13 ⏱ ~39 мин чтения 📐 4 Mermaid диаграмм

9. Kafka Streams

Потоковая топология, дуализм потоков и таблиц и локальное состояние

«Программы, преобразующие поток входных данных в поток результатов, — это самые чистые и легко компонуемые абстракции в вычислительной технике».
— Брайан Керниган

Что такое Kafka Streams

Исторически потоковая обработка больших данных (Stream Processing) ассоциировалась с громоздкими распределенными кластерами: Apache Storm, Apache Spark Streaming, Apache Flink. Развертывание таких систем требовало выделенного кластера, отдельных узлов-мастеров (Master/JobManager), пулов воркеров (TaskManagers), интеграции с YARN или Kubernetes, а также отдельной команды инженеров по эксплуатации.

Kafka Streams предложила радикально иную, прорывную парадигму: потоковая обработка как обычный микросервис, а не тяжеловесный джоб в кластере.

Kafka Streams — это клиентская библиотека (первоначально реализованная для Java, но породившая аналоги во всех языках, включая Go), которая превращает ваше стандартное приложение в масштабируемый, отказоустойчивый потоковый процессор.

Ключевые свойства архитектуры:

Никаких дополнительных кластеров: Приложение запускается как обычный контейнер в Kubernetes или процесс на виртуальной машине.
Инфраструктура Kafka как координатор: Масштабирование, обнаружение сбоев, шардирование задач и эластичность строятся на базе стандартного протокола Consumer Group (4. Consumer groups).
Встроенное шардированное состояние (Embedded State Store): Приложение хранит промежуточные агрегаты не во внешней базе данных, а локально на NVMe-диске (по умолчанию в embedded движке RocksDB), реплицируя изменения в компактные changelog-топики Kafka.

С точки зрения Go-разработчика, понимание Kafka Streams необходимо не столько для написания кода на Java, сколько для усвоения фундаментальных паттернов потоковой обработки (Stream-Table Duality, Windowing, Local State Stores, Interactive Queries), которые ежедневно применяются при проектировании распределенных систем на Go.

Основные абстракции: граф процессоров

Kafka Streams моделирует вычисления в виде топологии процессоров (Processor Topology) — направленного ациклического графа (DAG), где:

Узлы (Nodes): Процессоры потока (Stream Processors) — чистые функции трансформации, фильтрации или агрегации.
Рёбра (Edges): Непрерывные потоки событий (Streams), связывающие процессоры между собой.

flowchart LR
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    IT["Входной топик<br>user-events"]:::entry --> Source["Source Processor<br>(Чтение из Kafka)"]
    Source --> Filter["Filter<br>(только активные сессии)"]
    Filter --> Map["Map / Enrich<br>(добавление гео-данных)"]
    Map --> Agg["Aggregate / Window<br>(подсчет кликов за 5 минут)"]
    Agg --> Sink["Sink Processor<br>(Запись в топик)"]
    Sink --> OT["Выходной топик<br>regional-stats"]

Дуализм потоков и таблиц (Stream-Table Duality)

Краеугольный камень потоковой обработки — это математический дуализм между потоком изменений и таблицей состояний.

Поток (Stream) — это история изменений. Таблица (Table) — это текущий снимок состояния.
Если воспроизвести весь поток изменений с начала — вы получите таблицу.
Если отслеживать все модификации строк таблицы — вы получите поток изменений (CDC).

flowchart TD
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    subgraph Duality["Дуализм Stream ↔ Table"]
        S["KStream: Поток событий<br>[k: Alice, +$10], [k: Bob, +$50], [k: Alice, -$5]"]
        T["KTable: Текущее состояние<br>Alice = $5, Bob = $50"]:::process
    end

    S -->|"Агрегация (Aggregate)"| T
    T -->|"Поток изменений (Changelog Stream)"| S

В Kafka Streams этот дуализм выражен тремя базовыми абстракциями:

KStream (Поток записей): Представляет собой непрерывную последовательность независимых фактов. Каждое сообщение — это новое событие в мире. Дубликаты ключей не заменяют друг друга, а дописываются в лог. (Аналог в SQL: таблица логов аудита INSERT INTO audit_log ...).
KTable (Материализованная таблица): Представляет собой поток обновлений состояния (Changelog). Новая запись с ключом K интерпретируется как UPSERT (обновление), перезаписывая предыдущее значение. Запись со значением null (Tombstone) интерпретируется как DELETE. Каждая партиция KTable представляет собой локальный срез таблицы, шардированный по ключу.
GlobalKTable (Глобальная таблица справочников): В отличие от KTable, которая шардируется между инстансами консьюмеров, GlobalKTable полностью реплицируется на каждый экземпляр приложения. Она вычитывает все партиции топика и идеально подходит для справочников (страны, валюты, тарифные сетки), позволяя выполнять быстрые локальные соединения (joins) без дорогостоящего перераспределения входного потока (re-keying / shuffling).

Stateless и stateful операции

Stateless (без сохранения состояния)

Операции, не зависящие от истории предшествующих сообщений. Каждое событие обрабатывается изолированно в оперативной памяти:

filter / filterNot: Пропуск или отсеивание сообщений по предикату;
map / mapValues: Преобразование ключа или полезной нагрузки;
flatMap / flatMapValues: Преобразование одного входящего сообщения в ноль, один или список исходящих сообщений;
branch: Разделение одного входящего потока на несколько независимых ветвей в зависимости от бизнес-условий.

Stateless-операции тривиальны в эксплуатации: они не требуют дисковой памяти, не создают changelog-топиков и мгновенно перебалансируются при перезапуске подов.

Stateful (с сохранением состояния)

Операции, результат которых зависит от накопленного исторического контекста:

Агрегация (aggregate, count, reduce): Накопление сумм, счетчиков, средних значений по каждому ключу.
Оконные вычисления (windowedBy): Группировка событий по временным интервалам (например, число запросов от пользователя за последние 60 секунд).
Потоковые соединения (join):
- KStream-KStream Join — соединение двух потоков событий в пределах скользящего временного окна (например, соединение клика по рекламе с последующей покупкой в течение 30 минут).
- KStream-KTable Join — обогащение входящего потока событий актуальным состоянием сущности из таблицы (например, обогащение заказа текущим адресом пользователя).
- KStream-GlobalKTable Join — обогащение потока глобальным справочником без ограничений по партиционированию.

Окна: как придать времени структуру

В бесконечном потоке данных невозможно вычислить агрегат «для всех записей», потому что поток никогда не завершится. Единственный способ агрегировать данные — нарезать непрерывное время на дискретные окна (Windows):

flowchart TD
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    subgraph Tumbling["Tumbling Window (Перекатывающееся, 5 мин)"]
        direction LR
        TW1["00:00 - 00:05"]:::process --> TW2["00:05 - 00:10"] --> TW3["00:10 - 00:15"]
    end

    subgraph Hopping["Hopping Window (Скользящее: размер 5 мин, шаг 1 мин)"]
        direction LR
        HW1["00:00 - 00:05"]:::process
        HW2["00:01 - 00:06"]:::process
        HW3["00:02 - 00:07"]:::process
    end

    subgraph Session["Session Window (Сессионное, gap = 15 мин)"]
        direction LR
        SW1["User Activity (12:00..12:10)"]
        SW2["Inactive > 15m"]:::process
        SW3["New Session (13:00..13:05)"]
    end

Tumbling window (перекатывающееся окно): Окна фиксированной длительности, следующие друг за другом встык без перекрытия ([00:00-00:05], [00:05-00:10]). Каждое событие попадает строго в одно окно. Идеально для периодических отчетов.
Hopping window (скользящее окно): Окна фиксированной длины, сдвигающиеся на шаг, меньший размера окна (например, окно 5 минут с шагом 1 минута). Окна перекрываются; одно событие попадает сразу в несколько окон. Применяется для скользящих средних и алертинга.
Session window (сессионное окно): Динамическое окно, границы которого определяются активностью пользователя. Окно открывается с первым событием и расширяется, пока интервал между событиями не превышает порог неактивности (inactivity.gap). Как только наступает пауза дольше порога — сессия запечатывается.
Sliding window (скользящий join): Окно, привязанное не к фиксированной временной сетке, а к временной метке конкретной записи. Используется в операциях соединения двух потоков.

Хранение состояния: RocksDB, changelog и восстановление

В чем состоит главная архитектурная слабость наивных микросервисов потоковой обработки?
Они пытаются хранить агрегаты во внешней базе данных: Redis, PostgreSQL, Cassandra. При потоке в 200 000 RPS каждый апдейт счетчика означает сетевой round-trip к Redis. Сеть перегружается, задержки взлетают до десятков миллисекунд, а внешняя база данных деградирует под нагрузкой.

Kafka Streams решает эту проблему через Embedded Local State Store:

flowchart TD
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    subgraph StreamApp["Инстанс потокового приложения"]
        P["Stream Processor (CPU)"]
        R[("Embedded RocksDB (NVMe SSD)<br>Задержка 5-15 мкс!")]
        P <-->|"Локальные Get/Put без сети"| R
    end

    subgraph KafkaBroker["Кластер Kafka"]
        CL[("Changelog Topic<br>(cleanup.policy=compact)")]
    end

    P -.->|"Асинхронная репликация батчами"| CL

Ключевые элементы подсистемы состояния:

Локальное хранилище (state store): По умолчанию каждый экземпляр запускает в своем процессе высокопроизводительный встраиваемый LSM-движок RocksDB (разработанный в Meta на базе LevelDB от Google). Данные хранятся в томах, привязанных к экземпляру (обычно /var/lib/kafka-streams). Запись в локальный RocksDB занимает микросекунды, так как не требует выхода в сеть. В Go-реализациях аналогами могут быть Badger, BoltDB, Pebble.
Changelog-топик: Чтобы локальное состояние не потерялось при гибели сервера, все изменения RocksDB непрерывно реплицируются в специальный внутренний топик Kafka — Changelog Topic с политикой уплотнения cleanup.policy=compact (8. Retention и compaction). Настройка параметров компактификации min.cleanable.dirty.ratio и delete.retention.ms критична для быстрого восстановления.
Восстановление после сбоев: Если инстанс падает, Kubernetes перезапускает его на другой ноде. Новый процесс консьюмера назначается на те же партиции и вычитывает компактный changelog-топик с диска брокера, восстанавливая локальную базу RocksDB до актуального состояния.
Standby Replicas (num.standby.replicas): Чтобы многогигабайтное состояние не восстанавливалось с нуля минутами при редеплое, Kafka Streams поддерживает горячие резервные копии. Standby-инстансы пассивно вычитывают changelog-топик в фоновом режиме. При отказе основного воркера Standby-узел мгновенно перехватывает обработку без задержки на холодный прогрев кэша.

Под капотом

RocksDB внутри использует MemTable (буфер записи в оперативной памяти) и фоновую компактификацию SST-файлов. При записи данные сначала попадают в MemTable и WAL на диске, затем асинхронно сливаются в отсортированные строковые таблицы (SST). Это идеально ложится на модель Kafka Streams: быстрая аккумуляция обновлений, периодическое слияние. Страничный кеш Linux активно используется при чтении SST-файлов, а Go-аналоги Badger/Pebble работают по тем же принципам.

Exactly-once семантика

Kafka Streams реализует полноценную семантику Exactly-Once Processing (EOS) внутри замкнутого цикла Kafka $\to$ Обработка $\to$ Kafka.

Это достигается благодаря двухфазному коммиту транзакций Kafka (6. Exactly once в Kafka):

Процессор начинает транзакцию.
Промежуточные результаты пишутся в выходной топик.
Изменения локального состояния отправляются в changelog-топик.
Смещения входящего топика фиксируются через вызов sendOffsetsToTransaction.
Координатор фиксирует транзакцию: данные на выходе, смещения на входе и локальное состояние фиксируются строго атомарно.

Если в процессе падает контейнер — незавершенная транзакция прерывается (aborted), а состояние RocksDB откатывается к точке последней зафиксированной контрольной точки (checkpoint).

Интерактивные запросы (Interactive Queries)

Обычно потоковые приложения воспринимают как «черный ящик»: на вход поступают события, на выходе генерируются другие события.

Однако Kafka Streams предлагает революционную возможность: Interactive Queries (интерактивные запросы). Поскольку каждый инстанс держит в RocksDB материализованное состояние назначенных ему партиций (например, актуальные балансы пользователей), приложение может поднять локальный HTTP/gRPC API и отвечать на внешние запросы на чтение прямо из RocksDB!

Если запрос от пользователя приходит на Инстанс 1, а данные ключа лежат в партиции Инстанса 2, библиотека предоставляет метаданные StreamsMetadata: Инстанс 1 узнает сетевой адрес коллеги (http://worker-2:8080) и прозрачно проксирует запрос к нему.

Таким образом, Kafka Streams превращается в полноценную распределенную NoSQL базу данных с шардированием по партициям, материализующую состояние из лога событий!

Mechanical Sympathy: почему локальное состояние эффективнее удалённого

Сравним физику выполнения операции:

Параметр	Обращение к удаленному Redis/PostgreSQL	Чтение из Local Embedded RocksDB/Badger
Физический путь	CPU $\to$ TCP Socket $\to$ Маршрутизатор $\to$ Сервер БД $\to$ Диск/RAM БД $\to$ Ответ TCP	CPU $\to$ L3 Cache / DRAM (MemTable) $\to$ PCIe NVMe (SST file)
Задержка (Latency)	1 000 – 5 000 микросекунд (1–5 мс)	5 – 20 микросекунд (0.005–0.020 мс)
Точка отказа	Падение кластера БД останавливает все сервисы	Изолировано: падение одной ноды задевает только её партиции
Сетевой трафик	$2 \times \text{RPS} \times \text{MsgSize}$ в локальной сети	0 сетевых пакетов на чтение!

Локальный SSD/NVMe критичен: RocksDB активно использует fsync для WAL и случайное чтение SST-файлов. На HDD throughput будет плачевным. Использование локальных NVMe-дисков (как в инстансах AWS i3) позволяет получить задержки менее миллисекунды.

Go и потоковая обработка: альтернативы Kafka Streams

В экосистеме Go нет официальной библиотеки от команды Kafka, но сообщество разработало мощные архитектурные решения:

1. Goka — библиотека, вдохновлённая Kafka Streams

Goka — зрелая библиотека от инженеров Lovoo, повторяющая философию Kafka Streams на чистом Go:

Абстракция Emitter, Processor, View;
Локальное шардированное состояние на движках Badger (LSM-хранилище на Go) или LevelDB;
Автоматическая репликация в changelog-топики и восстановление состояния при ребалансировках;
Потокобезопасные представления (Views) для интерактивных запросов.

2. Ручная реализация с franz-go

Можно построить потоковый процессор, используя kgo.Client, горутины и собственное локальное хранилище (например, Badger, Bolt или Pebble):

Горутины консьюмера вычитывают батчи через PollFetches.
Обработчик выполняет быстрые Get/Put в локальный Pebble Store.
Изменения состояния пакуются в батчи и отправляются в changelog-топик через транзакционный продюсер cl.BeginTransaction().
Ребалансировки отслеживаются через коллбеки OnPartitionsRevoked (сброс WAL Pebble) и OnPartitionsAssigned (прогрев из changelog).

3. Использование внешних фреймворков

Для особо масштабных потоковых топологий с экстремально сложными соединениями десятков потоков может быть оправдано использование Apache Flink или Spark Streaming, вызываемых из Go через API, но это уводит от простоты и легковесности микросервисной архитектуры.

Вопросы для собеседования

Собеседование

Вопрос: Как Kafka Streams обеспечивает восстановление состояния при перезапуске инстанса, и что такое standby replicas?
Ответ: Каждое локальное состояние (RocksDB) дублируется в служебный компактифицированный changelog-топик Kafka. При падении инстанса новый узел вычитывает changelog и восстанавливает базу на диске. Чтобы этот процесс не вызывал задержек при терабайтных базах, включают num.standby.replicas: теневые узлы непрерывно читают changelog в фоновом режиме, держа локальную базу теплой. При аварии standby мгновенно становится активным лидером без перечитывания лога.

Заключение и дальнейшие шаги

Kafka Streams — это кульминация проектирования потоковых систем на основе лога: она превращает партицированный, реплицированный журнал в платформу для stateful-обработки, сохраняя простоту и горизонтальную масштабируемость обычного консьюмера. Понимание её внутреннего устройства и компромиссов критически важно для системного архитектора, даже если реализация будет на Go.

Следующая статья в разделе Kafka — 10. Kafka Connect, которая расширяет экосистему за пределы кастомных приложений, предоставляя стандартизированные коннекторы для интеграции с внешними системами без написания кода.