Перейти к содержанию

Spark micro-batch, watermarking и checkpointing

Глубокие темы · Раздел03

Тип материала: архитектурный разбор. Схемы и примеры иллюстративные, если не указано иное.

Версии в источниках: Kafka/Connect/Streams 4.1, Debezium 3.3, Flink 1.20.3, Spark 4.0.1 и Airflow 3.1 либо явно названная система. Непроверенные вопросы и ограничения runtime — в журнале (архив рабочего проекта).

Механизм

Micro-batch — execution mode, а не гарантия bounded latency. Structured Streaming строит incremental logical plan над unbounded input; checkpoint lineage связывает source progress и state, sink contract завершает fault-tolerance boundary.

flowchart TB
  Range[Source range per trigger] --> Plan[Incremental plan]
  Plan --> State[State store update]
  State --> Sink[Batch sink effect]
  Sink --> Progress[Checkpoint progress]
  Progress --> Replay[Restart same lineage]
  Replay --> Plan

Пошаговое объяснение

Trigger выбирает доступный input range → tasks выполняют transforms/shuffle → stateful operators обновляют state store → sink callback/commit → checkpoint прогресс. AvailableNow завершает bounded доступный backlog, не превращает источник в бессрочный snapshot.

Failure scenarios и recovery

foreachBatch может исполниться снова; batch_id уникален лишь внутри lineage. Новый checkpoint namespace требует generation, иначе старый batch0 marker может скрыть новые данные. Внесённые в state schema/operator изменения не всегда совместимы с прежним checkpoint.

Trade-offs и эксплуатация

Append/update/complete определяют output interpretation. Watermark может ограничивать state и допускать финализацию, но не является универсальным filter для любого query. Complete output может требовать сохранения всего aggregate state. Slow sink растягивает batch независимо от configured trigger.

Практический пример

Executable Spark sample публикует full per-generation snapshot атомарной заменой local file, не демонстрирует production S3 transaction или финансовый ledger.

Типичные ошибки проектирования

Смешивать source/state/sink boundaries, скрывать failed records и неопределённый исход timeout; применять retry без identity/version и retention contract. Конкретные ограничения разобраны выше.

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

Можно удалить checkpoint, чтобы убрать ошибку? Только как новое вычисление с новым state/output protocol и reconciliation.

Почему batch=1s не даёт latency1s? Обработка/backlog/state/sink могут длиться дольше.

Практические упражнения

Повторите callback одного batch и restart с другой generation. Опишите отличия результата от additive append sink.

Дополнительные источники