В производных данных мы договорились: писать надо в источник правды, а кеши, индексы и витрины — деривировать из него единым потоком изменений. Эта статья — про то, как этот поток устроен на практике, и про две вещи, на которых спотыкаются чаще всего: как именно изменения из базы попадают в производные системы (CDC) и почему аналитика по потоку врёт, если перепутать время события и время обработки. Потоковая обработка — это батч, который никогда не заканчивается: те же преобразования, но над бесконечным потоком, с низкой задержкой.
CDC: база как источник потока изменений
Как заставить поисковый индекс всегда соответствовать базе, не скатываясь в двойную запись? Ответ — перехват изменений данных (change data capture, CDC): назначить базу ведущим узлом, читать её журнал изменений и применять эти изменения к производным системам в том же порядке, в каком они легли в базу.
Механика: у базы уже есть журнал упреждающей записи (WAL) — по нему работает репликация. CDC-инструмент подключается к нему как ещё одна «реплика» и декодирует поток изменений: Debezium читает logical decoding PostgreSQL, бинарный лог MySQL, oplog MongoDB — и публикует изменения в лог-брокер (Kafka). Дальше индекс, кеш, аналитическое хранилище — просто потребители этого потока, и раз он сохраняет порядок, они сходятся к состоянию базы.
Две детали, без которых CDC не полон:
- Начальный снимок. Журнал хранит не всю историю (старые сегменты удаляются), а новому индексу нужны и давно не менявшиеся записи. Поэтому старт — это снимок всей базы на известной позиции журнала, а затем поток изменений с этой позиции.
- Уплотнение журнала (log compaction). Чтобы не хранить журнал вечно, брокер выбрасывает старые записи с уже перезаписанным ключом, оставляя по ключу только последнее значение. Тогда журнал = полная копия текущего состояния, и производную систему можно пересобрать с нуля из него одного.
CDC асинхронен: производные системы отстают на задержку репликации — как обычные реплики. Плюс — источник правды почти не чувствует медленных потребителей.
Время события против времени обработки
Вторая ловушка тоньше и коварнее. Потоковый процессор часто считает что-то «за последние 5 минут» — и здесь два разных времени:
- время события (event-time) — когда действие реально произошло;
- время обработки (processing-time) — когда процессор до события добрался.
Между ними разрыв: события задерживаются в очередях, при сетевых сбоях, при перезапуске потребителя (который потом залпом доедает накопленное). И тогда группировка по времени обработки даёт фантом: процессор на минуту завис, после перезапуска разом обработал накопленные события — на графике «частота запросов» рисуется всплеск, хотя реальная частота не менялась. Считать надо по времени события.
Аналогия Клеппмана: «Звёздные войны» вышли не в порядке эпизодов (IV, V, VI, потом I, II, III). Если смотреть в порядке выхода, порядок повествования нарушен. Дата события — это год по сюжету, дата обработки — когда посмотрел. В потоке порядок событий нарушается так же, и алгоритм обязан это учитывать.
Отдельная боль — отставшие события (stragglers): окно 37-й минуты вы вроде закрыли и посчитали, а тут приходит событие с меткой 37:59, застрявшее в сети. Вариантов два: игнорировать опоздавших (и мониторить долю отброшенных) или публиковать поправку к уже выданному окну. «Правильного» ответа нет — это осознанный выбор.
Типы окон
Раз считаем по времени события, надо решить, как нарезать время на окна:
- Падающее (tumbling) — фиксированная длина, каждое событие ровно в одно окно: минутные окна 10:03:00–10:03:59, 10:04:00–10:04:59. Простейший вариант.
- Прыгающее (hopping) — фиксированная длина, но окна перекрываются для сглаживания: 5-минутное окно с шагом 1 минута.
- Скользящее (sliding) — все события, попавшие в интервал заданной ширины друг от друга; без фиксированных границ.
- Сессионное (session) — без фиксированной длины: группирует события одного пользователя, идущие близко во времени, и закрывается, когда пользователь замолкает (например, 30 минут тишины). Классика веб-аналитики.
Где это применяется
Как только производную систему (индекс, кеш, витрину) надо держать свежей почти в реальном времени или считать метрики по потоку событий — вы в потоковой обработке. Практическая рамка: синхронизируйте производные системы через CDC/лог-брокер, а не двойной записью; любую аналитику по времени считайте по времени события, а не обработки; для окон честно решите, что делаете с опоздавшими.
Где спотыкаются начинающие:
- Держат индекс/кеш в синхроне двойной записью — расходятся молча. CDC делает базу ведущим, производные — потребителями журнала.
- Считают частоту/среднее по времени обработки — при отставании процессора получают фантомные всплески, которых в реальности не было.
- Считают окно закрытым по стенным часам — и теряют отставшие события. Нужна политика: дропать (и мониторить долю) или публиковать поправку.
- Путают event-time и processing-time при перезапуске — переигровка накопленного потока рисует аномалию на ровном месте.
Что почитать дальше: производные данные — зачем вообще нужен единый поток изменений; AMQP vs Kafka — лог-брокер, на котором держится CDC и replay; event sourcing — поток событий как источник правды; двойная запись — антипаттерн, который CDC заменяет. Первоисточник — Мартин Клеппман, «Высоконагруженные приложения», глава 11.