Apache Spark / Structured Streaming¶

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

Область версий: Apache Spark / Structured Streaming 4.0.1, Java21, Python3.12; PySpark4.0.1. Проверка источников: 2026-11-02; это не рекомендация latest release.

Архитектура и роль¶

FACT. Driver строит logical/physical plans и stages, executors исполняют tasks над partitions. Spark SQL optimizer, shuffle и columnar execution обслуживают batch и Structured Streaming. Baseline4.0.1 использует Java17/21 и Scala2.13; его нельзя автоматически соединять с Delta3.3/Spark3.5 примерами раздела02.

Внутренние механизмы и движение данных¶

В micro-batch query источник фиксирует range offsets, plan обрабатывает порцию, state store обновляет window/join state, sink публикует batch и checkpoint хранит progress. Trigger определяет scheduling, не единый event-time boundary. AvailableNow обрабатывает доступный bounded backlog и завершает query; processingTime продолжает polling.

Shuffle перераспределяет rows по key, создавая network/disk IO; skew делает один task длинным. Watermark задаётся до stateful aggregation и связан с observed event times. Append output для windows требует условия finalization; update/complete output имеют другой contract. Streaming queries с несколькими stateful operators требуют проверки поддерживаемых combinations.

flowchart TB
  Files[Replayable source ranges] --> Driver[Driver incremental plan]
  Driver --> Tasks[Executor tasks and shuffle]
  Tasks --> State[Window state store]
  State --> Callback[foreachBatch or supported sink]
  Callback --> Commit[Sink commit contract]
  Driver --> Checkpoint[Offsets and checkpoint lineage]

Ordering и guarantees¶

Micro-batch fault tolerance опирается на replayable source и корректный sink/checkpoint. foreachBatch по умолчанию допускает повтор; batch_id может использоваться для dedup внутри одной checkpoint lineage, но новая checkpoint lineage может переиспользовать IDs. Нужны generation + batch identity либо business keys/versions. Continuous processing — отдельный ограниченный режим с at-least-once semantics, не автоматическое ускорение любого stateful query.

Документированный contract и версия.

Масштабирование и backpressure¶

Количество shuffle partitions, task size, executor memory, spill/state store и sink concurrency определяют capacity. Batch interval не доказывает latency: предыдущая порция может исполняться дольше интервала. AQE и batch optimizations не следует считать одинаково применимыми к каждому streaming plan.

Failure / recovery / replay¶

После restart читать ту же checkpoint location и совместимый query/state schema. Удаление checkpoint ради починки может повторить source history и сбросить state. Новая generation публикуется после reconciliation/barrier. Ошибка foreachBatch после external write до checkpoint снова вызовет batch; sink должен пережить этот повтор.

Документированный contract и версия.

Эксплуатация и мониторинг¶

StreamingQueryProgress: input/processed rows rate, batch duration, watermark, stateOperators rows/memory/dropped-by-watermark, source offsets и sink errors. Spark UI tasks/skew/shuffle/spill/GC. Хранить queryId/runId/checkpoint generation и input manifest; отсутствие exception не равно полнота финансового отчёта.

Ограничения и критерии выбора¶

Выбирать при Spark SQL/batch экосистеме, unified batch/reprocessing и подходящем micro-batch freshness. Не выбирать Airflow вместо streaming runtime или continuous mode без поддерживаемых operators. Arbitrary local Python side effects executors могут повторяться при task retry/speculation.

FinTech / iGaming и реальные источники¶

FACT, исторически: PayPal описывает Gimel как analytics platform; это не подтверждение Spark4.0.1 либо нашего sink protocol. F07. INFERENCE: reconciliation, reporting и event aggregates с явными corrections.

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

Structured Streaming sample выполняет availableNow JSON stream с window/state/checkpoint и generation-scoped snapshot sink. Локальный smoke и distributed recovery имеют разные статусы в отчёте.

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

batch_id уникален навсегда; удалить checkpoint без replay plan; dropDuplicates без horizon/бизнес-версии; считать update output самостоятельными additive facts.

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

foreachBatch автоматически EOS? Нет, callback может повториться.

Watermark гарантирует удаление всех более поздних events? Это policy/progress механизм; guarantees зависят от operators/режима, late data не имеет универсального исхода.

Почему больше executors не ускорили job? Возможны hot partition, driver, sink или shuffle bottleneck.

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

Используйте тот же input с duplicate ID, затем late correction. Сравните additive insert sink и versioned replacement; проверьте restart с прежним и новым checkpoint namespace.