Архитектура потоковой обработки: Structured Streaming и режимы обработки

Потоковая обработка в Apache Spark реализуется через Structured Streaming, который предоставляет единый API на базе DataFrame/Dataset и интегрированную модель выполнения. Архитектура этой подсистемы строится вокруг понятия длительного запроса (streaming query), который видит поток как непрерывно поступающие данные, разделённые на микро-батчи или, в экспериментальной непрерывной обработке, - на более мелкие шаги обработки. Эта глава посвящена архитектурным принципам Structured Streaming, режимам обработки, особенностям управления временем и состоянием, а также паттернам интеграции в современные ETL- и аналитические конвейеры.

Structured Streaming подойдёт для сценариев от конвейеров ETL до реального времени аналитики, где критически важны гарантии корректности и наблюдаемость процессов. В рамках курса мы рассмотрим, как принципы архитектуры влияют на проектирование пайплайнов, какие trade-off лежат в основе выбора режима обработки и какие практики применяются на практике для достижения требуемой задержки, устойчивости и масштабируемости.

Краткое содержание главы
Архитектура Structured Streaming: компоненты, планирование и выполнение.
Режимы обработки: микробатчинг и непрерывная обработка** - преимущества и ограничения.
Управление временем, состоянием и водяными метками: паттерны окон, задержки и устойчивость к задержкам.
Интеграция источников/приёмников и паттерны проектирования ETL-конвейеров с аппаратами мониторинга и обеспечения качества данных.

Концепции и архитектура Structured Streaming

Structured Streaming представляет поток как непрерывный набор таблиц, которые прогоняются через единый движок выполнения DataFrame/Dataset API. В отличие от традиционных пакетных подходов, здесь каждая микропартия данных обрабатывается как часть непрерывной задачи, что позволяет согласованно применять трансформации, агрегации и соединения в рамках единой сборки плана выполнения. Внутренне запрос превращается в StreamingQuery, который управляет чтением источников, обработкой и записью результатов через конвейеры исполнения.

Ключевые компоненты архитектуры:

Источник данных (Source): Kafka, файлы в HDFS/S3, сокеты, Kinesis и др. Источник обеспечивает прокси-слой для данных и метаданных (например, смещение в Kafka).
Преобразование (Processing): стандартный набор трансформаций DataFrame/Dataset - map, filter, groupBy, window, объединения и сложные операции состояния. Встроенный оптимизатор выполняет планирование как для статических так и для потоковых данных.
Источник и приемник состояния (Stateful operators): операции, которые требуют сохранения состояния между микро-батчами (например, агрегации по окнам, session-based паттерны, mapGroupsWithState и аналогичные).
Хранилище времени и состояния (Time and State): водяные метки (watermarks) и окна управляют тем, как долго можно держать данные в памяти для последующих вычислений и когда можно очищать состояние.
Конвейер записи (Sink): вывод конечного результата в файлы (Parquet, Delta Lake), базы данных, консолидированные логи вывода или визуализации. Поддерживаются режимы вывода Append, Update и Complete, в зависимости от типа запроса.
Менеджмент устойчивости и восстановления (Fault Tolerance): checkpointLocation и иные механизмы журналирования сохраняют прогресс и обработанные оффсеты. Это позволяет повторно запустить обработку с того места, где выполнение было остановлено.

На уровне исполнения Spark структурированный поток реализуется как серия микробатчей. По умолчанию структурированный поток собирает данные за фиксированные интервалы времени и обрабатывает их как пакет, используя одно и то же планирование. Это даёт гибкость и детерминированность, необходимые для предсказуемой производительности и масштабирования в рамках существующей архитектуры Spark. Важной частью является единая модель обработки, которая упрощает тестирование, мониторинг и поддержку - поведение одинаково применимо к пакетной и к потоковой обработке, что упрощает переход между режимами.

Почему именно такая архитектура имеет смысл для больших конвейеров:

Унифицированный API упрощает сопровождение и развитие пайплайнов: одна логика для обработки как потоковых, так и пакетных входов.
Гарантия точного выполнения (exactly-once) достигается через согласованный механизм управления оффсетами, журналированием и повторной обработкой, особенно при работе с внешними Sink-ами, поддерживающими атомарность транзакций.
Эффективное управление состоянием и водяными метками позволяет лимитировать затраты на память и дисковое пространство, особенно при работе с большими окнами или частыми задержками данных.

Понимание архитектуры имеет важное значение для проектирования пайплайнов, поскольку выбор источника, типа окон и политики вывода напрямую влияет на требования к памяти, задержкам и устойчивости к задержкам. Например, при работе с Kafka как источником, Offset Management и семантика обработки требуют корректной интеграции с выбором режима вывода и стратегии обработки задержанных данных.

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

val spark = SparkSession.builder()
  .appName("StructuredStreamingArchitectureExample")
  .getOrCreate()

// Пример источника: Kafka
val kafkaDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
  .option("subscribe", "events")
  .option("startingOffsets", "latest")
  .load()

// Преобразование и парсинг значения
val valueSchema = new StructType()
  .add("user_id", StringType)
  .add("action", StringType)
  .add("event_time", TimestampType)

val parsed = kafkaDF
  .selectExpr("CAST(value AS STRING) as json", "timestamp")
  .select(from_json(col("json"), valueSchema).as("data"), col("timestamp"))
  .select("data.*", "timestamp")

// Водяная метка и окно
val withWatermark = parsed
  .withWatermark("event_time", "10 minutes")

val windowed = withWatermark
  .groupBy(window(col("event_time"), "15 minutes"), col("action"))
  .count()

// Вывод: пример приема в Delta Lake
val query = windowed.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/checkpoints/streaming-architecture")
  .outputMode("append")
  .start("/path/delta/events_summary")

query.awaitTermination()

Режимы обработки: микробатчинг и непрерывная обработка

Structured Streaming поддерживает два основных режима обработки, и их выбор зависит от требований к задержке, точности и совместимости с источниками/приёмниками данных.

Микробатчинг (micro-batching)
- Это базовый и наиболее широко применяемый режим. Выполнение делится на последовательность микро-батчей фиксированной длительности (например, 5-30 секунд). Каждый батч обрабатывается независимо, результаты накапливаются в состоянии и публикуются в приемник.
- Преимущества: простота реализации, широкая совместимость со всеми источниками и Sink-ами, наличие детальных гарантий точного выполнения и прозрачная интеграция с существующими инструментами мониторинга.
- Ограничения: задержка пропуска данных обратно пропорциональна длительности микро-батча; в требованиях к задержке ниже нескольких секунд может потребоваться другая архитектура, а также более сложные паттерны для удержания и очистки состояния.
- Практические выводы: для ETL-конвейеров, где важна предсказуемость и детерминированность, микробатчинг обеспечивает устойчивую производительность и совместимость, а также упрощает интеграцию с внешними источниками и Sinks.
Непрерывная обработка (continuous processing, экспериментальная)
- Этот режим направлен на минимизацию задержки и обработку отдельных записей почти в реальном времени. Часто достигается за счёт пессимистичных или догоняющих стратегий коммита и упрощённых траекторий обработки.
- Преимущества: значительно меньшая задержка по сравнению с микробатчингом, потенциально выше пропускная способность для простых трансформаций.
- Ограничения: ограниченная совместимость источников и приемников, ограниченный набор поддерживаемых операций (часто без сложной агрегации по окнам, сложных соединений и больших состояний), меньшая зрелость по устойчивости к ошибкам и координации, более сложные требования к транзакционности и откатам. В продакшн-проектах этот режим применяется редко и обычно на стадиях пилотов под конкретные сценарии низкой задержки.
- Практические выводы: непрерывная обработка подходит для узких кейсов низкой задержки и простых трансформаций, но для большинства реальных пайплайнов предпочтение остаётся за микробатчингом с должной настройкой параметров задержки и памяти.
Взаимодействие режимов и выбор стратегии
- В большинстве проектов целесообразно начинать с микробатчинга, чтобы обеспечить надёжность, мониторинг и простоту поддержки. По мере роста требований к задержке и контролю над латентностью можно рассмотреть переход к более низкоуровневым стратегиям, включая тестирование возможностей непрерывной обработки, ограничивая применение к тем задачам, где требования к точности и порядок операций совпадают с поддерживаемыми ограничениями.
- Мониторинг и тестирование критичны: показатели задержки, размер состояния, частота массирования и устойчивость к задержкам - всё это влияет на выбор режима. Разделение конвейеров по требованиям к задержке и точности - эффективная практика архитектуры больших данных.

Модель состояния, окна и водяные метки

Управление временем и состоянием - один из ключевых аспектов потоковой архитектуры в Spark. Водяные метки (watermarks) устанавливают допуски по задержке данных и позволяют ограничить накопление состояния для оконных агрегатов, что критично для устойчивости к задержкам и ресурсному управлению.

Водяная метка
- Вводит концепцию допустимой задержки: late data может быть принято в рамках установленного окна, после чего данные позднее не учитываются. Это обеспечивает баланс между латентностью и корректностью агрегатов.
Окна и паттерны
- Тьюблинговые окна (tumbling), скользящие окна (sliding) и сессийные окна (session windows) позволяют агрегировать данные по времени событий. В Spark они реализуются через window(col("event_time"), "durations") и сопутствующие функции, причем согласование с watermark задаёт момент очистки состояния.
Состояние и операторы
- Stateful operators сохраняют данные между батчами: агрегаты по ключам, соединения, пользовательские функции состояния (MapGroupsWithState, FlatMapGroupsWithState и т.д.). Эффективность этих операторов во многом определяется размером состояния и периодами очистки, которые обеспечиваются водяными метками.
Семантика времени
- Event-time зависимая обработка позволяет обрабатывать данные в порядке времени события, а не по порядку поступления в систему.Processing-time - более простая семантика, привязанная к моменту обработки в системе. В real-time сценариях event-time обеспечивает корректность с учётом задержек и задержанных данных.

Пример кода: водяная метка, окно и агрегация

import org.apache.spark.sql.functions._
val df = spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092")
  .option("subscribe", "events")
  .load()

val events = df.selectExpr("CAST(value AS STRING) as json", "timestamp")
  .select(from_json(col("json"), schema).as("data"), col("timestamp"))
  .select("data.*", "timestamp")

val withWatermark = events.withWatermark("event_time", "10 minutes")

val windowed = withWatermark
  .groupBy(window(col("event_time"), "15 minutes"), col("event_type"))
  .count()

val query = windowed.writeStream
  .format("parquet")
  .option("checkpointLocation", "/path/checkpoints/streaming-wm")
  .outputMode("append")
  .start("/path/parquet/events_summary")

Эти конструкции позволяют реализовать ETL-паттерны с обработкой по окнам, сохраняя разумный баланс между задержкой и точностью. В реальном проекте важна согласованность окон, водяных меток и конфигураций вывода: неоднозначности в этих конфигурациях приводят к пропуску данных или дублированию результатов.

Интеграции, схемы обработки и паттерны ETL

Архитектура структуры потоковой обработки Spark предполагает тесную интеграцию с внешними системами, а также грамотное управление схемами и изменяемостью данных. В типичных практиках стоит рассматривать следующие аспекты:

Источники и приемники
- Kafka остаётся основным источником для потоковой обработки из-за своей надёжности, масштабируемости и Ordering. В качестве приёмников чаще применяют Delta Lake, Parquet в распределённом хранилище, а также базы данных через JDBC. Delta Lake особенно полезен в сценариях, требующих ACID-конечности и upsert-операций на больших объёмах данных.
Управление схемой
- Системы потоковой обработки часто сталкиваются с изменяемостью схем: новые поля, изменение типов. Spark поддерживает ограниченную схему-эволюцию на уровне чтения и записи, а Delta Lake прямо предоставляет схемы и схему-эволюцию в контексте делтовской таблицы.
Паттерны ETL
- CDC (Change Data Capture) может быть реализован через потоковую обработку изменений, сопоставление ключей и временных меток. Объединение потоков с историческими данными - частый сценарий: например, соединение потоковых данных с статическими таблицами через JOIN, ограниченный по времени; для полноценной поддержки JOIN между потоком и статикой требуются окна и watermark.
- Upserts и удаление записей
  - Delta Lake обеспечивает ACID upserts через MERGE INTO. В Spark это реализуется через операции над Delta Lake: в streaming-процессах MERGE может быть применён к обновлению существующих записей в целевой Delta-таблице.
Контроль качества и мониторинг
- Важна интеграция со Spark UI, средствами мониторинга и трассировкой. Логи Offsets и прогресса выполнения в checkpoint позволяют быстро определить этапы конвейера и узкие места.

Open-source примеры, которые реально усиливают смысл:

Apache Kafka в качестве источника стабильно используется во многих проектах, обеспечивая надёжную доставку и порядок событий.
Delta Lake как дополнение к Spark для обеспечения ACID-операций и эффективного управления схемой и обновлениями. Эти решения получают широкое распространение в промышленной среде и хорошо документированы.

Практическая реализация и дизайн конвейера: пример и best practices

Типовой потоковый ETL-конвейер строится на следующих принципах:

Определение источника с учётом требований к задержке, порядка событий и потребления offsets.
Применение парсинга и валидации данных на стадии преобразований.
Построение окон и водяных меток для агрегаций и сложных трансформаций.
Вывод в устойчивые хранилища с поддержкой версий и транзакций (Delta Lake) с журналированием прогресса через checkpoint.
Мониторинг, алертинг и тестирование на основе эмуляции задержек в тестовой среде.

Ниже приведён демонстративный пример кода, который иллюстрирует создание простого streaming конвейера: чтение из Kafka, парсинг JSON-сообщений, агрегирование по 15-минутным окнам и запись в Delta Lake с checkpoint. Пример рассчитан на иллюстрацию архитектуры и не внедряется как демонстрационный код ради демонстрации.

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

val spark = SparkSession.builder()
  .appName("StreamingETLExample")
  .getOrCreate()

val sparkSchema = new StructType()
  .add("user_id", StringType)
  .add("action", StringType)
  .add("event_time", TimestampType)

val raw = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092")
  .option("subscribe", "user_events")
  .load()

val parsed = raw.selectExpr("CAST(value AS STRING) as json")
  .select(from_json(col("json"), sparkSchema).as("data"), col("timestamp"))
  .select("data.*", "timestamp")

val withWatermark = parsed
  .withWatermark("event_time", "10 minutes")

val windowed = withWatermark
  .groupBy(window(col("event_time"), "15 minutes"), $"action")
  .count()

val query = windowed.writeStream
  .format("delta")
  .option("checkpointLocation", "/path/checkpoints/etl-demo")
  .outputMode("append")
  .start("/path/delta/etl_events_summary")

query.awaitTermination()

Лучшие практики проектирования потоковых конвейеров:

Чётко определяйте требования к задержке и через какие каналы данные должны проходить. Это определяет выбор режима обработки, окон и watermark.
Выбирайте источники с учётом поддержки транзакций и порядка событий. Kafka остаётся надёжной основой во многих сценариях.
Применяйте Delta Lake для устойчивого хранилища и поддержки upsert-операций, особенно когда требования к консистентности данных высоки.
Устанавливайте checkpoint-локейшены и тестируйте обработку с задержками, чтобы обеспечить повторную обработку без потерь данных при сбоях.
Сопровождайте пайплайны мониторингом и тестированием на реальных сценариях роста нагрузки - это позволяет своевременно поднимать лимиты ресурсов и предотвращать деградацию производительности.

Key takeaways

Structured Streaming обеспечивает единый подход к обработке как потоковых, так и пакетных данных через DataFrame/Dataset API и обеспечивает консистентность выполнения через checkpoint и точку входа offsets.
Микробатчинг остаётся базовым и надёжным режимом, позволяющим достигать BALANCED latency и throughput; непрерывная обработка предоставляет низкую задержку, но имеет ограниченную совместимость и устойчивость к сложным операциям.
Водяные метки и окна - критические инструменты управления временем и состоянием, позволяющие ограничить размер сохранённого состояния и корректно обрабатывать задержанные данные.
Интеграции с Kafka и Delta Lake помогают строить устойчивые ETL-конвейеры с поддержкой версии данных, транзакций и схемы эволюции.
При проектировании пайплайна следует уделять внимание выбору режима обработки, устойчивости к задержкам и стратегиям обработки ошибок, а также мониторингу и тестированию конвейера.
Применение паттернов управления временем, состоянием и окон в сочетании с надёжной инфраструктурой обеспечивает масштабируемость и надёжность больших потоковых систем.

FAQ

Что такое Structured Streaming и чем он отличается от обычного "потока"?

Structured Streaming - это слой поверх API DataFrame/Dataset, который позволяет описывать потоковую обработку как выражение на таблицах. В отличие от классических потоков, у него единая модель выполнения и возможность использования всего набора трансформаций Spark, включая агрегации, окна иjoins, с поддержкой чекпоинтов и exactly-once semantics. Это упрощает миграцию между пакетной и потоковой обработкой и облегчает мониторинг конвейера.

Какие режимы обработки поддерживаются в Spark Structured Streaming?

Основной режим - микробатчинг, который обеспечивает надёжность, совместимость и хорошую предсказуемость. Непрерывная обработка - экспериментальная технология с низкой задержкой, но ограничениями по совместимости источников/приёмников и поддерживаемыми операциями. Выбор между ними зависит от требований к задержке и сложности трансформаций.

Что такое watermark и зачем он нужен?

Watermark - это метка времени, которая задаёт допустимую задержку для данных, приходящих позднее. Она позволяет ограничить долгоживущее состояние и управлять очисткой окон и состояния, чтобы не держать данные вечно. Правильная настройка watermark критична для баланса между задержкой и точностью.

Какие паттерны эпизодов состояния применяются в Structured Streaming?

Ключевые паттерны: оконная агрегация (tumbling, sliding, session windows), stateful агрегирования (mapGroupsWithState, flatMapGroupsWithState), join-паттерны между потоками и статическими данными. Эффективность определяется размером состояния и частотой его обновления.

Какие источники и приёмники чаще всего используют в промпроизводстве?

Источники: Kafka остаётся ведущим выбором благодаря порядку сообщений и масштабируемости. Приёмники: Delta Lake для ACID-операций и консистентных апдейтов, Parquet в HDFS/S3 для долговременного хранения. Сложности с совместимостью между источниками и sink понимаются через ограничения режимов выполнения.

Как обеспечить надёжность и повторяемость обработки?

Включение checkpoint-логирования, выбор устойчивых Sink-ов (например Delta Lake), правильная политика обработки задержанных данных через watermark и окна. Тестирование конвейера с эмуляцией задержек и сбоев - лучшее средство для повышения надёжности.

Какой паттерн выбрать для ETL-пайплайна в реальном бизнес-случае?

Начать с микробатчинга, чтобы обеспечить надёжность и простоту сопровождения. По мере необходимости задержки и latency можно рассмотреть упрощённые сценарии непрерывной обработки, но с учётом ограничений по поддержке и транзакционности. В большинстве реальных проектов Delta Lake и Kafka вместе обеспечивают надёжную, масштабируемую архитектуру.

Какие инструменты мониторинга нужны в Streaming-пайплайне?

Spark UI для настройки планов выполнения и мониторинга прогресса. Метрики JVM, мониторинг задержек и пропускной способности источников, лаги в потоковой обработке, размер состояния и частота очищения состояния. Интеграция с системами алертинга (Prometheus/Grafana) обычно рекомендуется.

Какие сложности возникают при эволюции схемы данных в потоках?

Добавление новых полей, изменение типов и удаление полей требуют аккуратной миграции схемы и обеспечения обратной совместимости. Delta Lake упрощает часть этой задачи за счёт поддержки схемной эволюции и MERGE-операций; Spark адаптирует чтение и запись под новую схему, но в потоковой обработке следует планировать миграцию с минимальными прерываниями.

Какие ограничения следует помнить при выборе непрерывной обработки?

Непрерывная обработка требует ограниченной сложности трансформаций, меньшей поддержки сложных окон и соединений, меньшей совместимости с некоторыми источниками и приемниками, а также меньшей зрелости по сравнению с микробатчингом. Она подходит в узких сценариях низкой задержки и в случаях, когда можно пожертвовать рядом возможностей, не критичных для бизнес-логики.

← Предыдущая статья

Риски и типовые ошибки в Spark-проектах

Следующая статья →

Data lakehouse и современные архитектурные паттерны