Structured Streaming¶

Structured Streaming — это масштабируемый и отказоустойчивый движок потоковой обработки данных, встроенный в Apache Spark (начиная с версии 2.0, выпущенной в 2016 году). Он позволяет обрабатывать потоки данных в реальном времени, используя тот же высокоуровневый API DataFrames и Dataset, что и для пакетных вычислений, реализуя парадигму «непрерывных приложений» (continuous applications). Ключевая особенность — представление непрерывного потока ввходящих данных как бесконечно растущей таблицы, к которой применяются реляционные операции (фильтрация, агрегация, соединения) по принципу микропакетной обработки (micro-batch processing) или, начиная с Spark 2.3, непрерывной обработки (continuous processing).

¶История и предпосылки

До появления Structured Streaming обработка потоков в Apache Spark была возможна через более низкоуровневый API — Spark Streaming (DStreams), основанный на дискретизированных потоках (Discretized Streams). Этот подход имел ряд ограничений: сложность в сочетании с пакетными операциями, проблемы с гарантиями «ровно один раз» (exactly-once semantics) и трудоёмкость поддержания состояния. Structured Streaming был разработан в 2015–2016 годах командой инженеров Databricks (в первую очередь Михаилом Армянским и Татхой Кристианом) как попытка унифицировать пакетную и потоковую обработку.

Концепция была впервые представлена широкой публике в 2016 году на конференции Spark Summit. К 2018 году, с выходом Apache Spark 2.3, была добавлена экспериментальная поддержка непрерывной обработки, обеспечивающая субмиллисекундные задержки. С 2019 года (Spark 2.4 и 3.0) Structured Streaming стал основным и рекомендуемым инструментом потоковой обработки в экосистеме Apache Spark.

¶Архитектура и принципы работы

¶Модель «поток как таблица»

В основе Structured Streaming лежит метафора «бесконечной таблицы». Каждый новый пакет данных (микропакет) или каждое новое событие (в режиме непрерывной обработки) интерпретируется как новая строка, добавляемая в эту таблицу. Пользователь задаёт запрос в терминах DataFrames или Dataset (например, spark.readStream().table("input").groupBy("key").count()), который выполняется инкрементально: результирующая таблица пересчитывается при добавлении новых данных.

Ключевое следствие такого подхода — логический план выполнения запроса (query plan) одинаков для потокового и пакетного сценариев. Это позволяет легко переключаться между ними, меняя только метод чтения и записи.

¶Режимы вывода (Output Modes)

Движок поддерживает три основных режима, определяющих, какие именно изменения в результирующей таблице будут записаны в выходной поток (sink):

• Append (добавление): Новые строки добавляются только в конец таблицы. Применим для запросов, не содержащих модификаций существующих данных (например, простая фильтрация). Для запросов с агрегацией требуется указать водяной знак (watermark) для определения, какие данные считаются «устаревшими».
• Complete (полный): Все содержимое результирующей таблицы выводится после каждой итерации. Используется преимущественно для глобальных агрегаций (например, общее количество событий), когда результат помещается в память или небольшое хранилище.
• Update (обновление): Выводятся только строки, которые были изменены или вставлены с последнего триггера. Этот режим наиболее эффективен для запросов с оконными агрегациями и группировками, когда ключи не удаляются.

¶Семантика обработки (Exactly-Once)

Одно из главных достижений Structured Streaming — гарантия «ровно один раз» (exactly-once) для всех поддерживаемых источников и приёмников, при условии их идемпотентности и транзакционности. Это достигается за счёт:

• Микропакетной обработки с фиксацией контрольных точек (checkpointing) в надёжном хранилище (например, Amazon S3, HDFS).
• Отслеживания офсетов (offset) в источниках (Kafka, Kinesis).
• Атомарной записи в приёмник.

¶Управление состоянием (State Management)

Для обработки событий, которые могут прийти с задержкой или в неправильном порядке, используется концепция водяного знака (watermark). Водяной знак — это временная граница, определяющая, до какого момента данные считаются «актуальными». Данные, поступившие позже водяного знака, отбрасываются или учитываются с пометкой опоздавших. Пользователь задаёт водяной знак в определении окна агрегации.

Движок поддерживает произвольные состояния (arbitrary stateful operations) через интерфейс mapGroupsWithState() и flatMapGroupsWithState(), позволяя реализовать сложную логику, например, обнаружение аномалий или поддержание сессий.

¶Применение

Structured Streaming применяется в задачах, требующих обработки непрерывных потоков данных:

• Мониторинг и аналитика в реальном времени: Агрегация метрик с серверов и датчиков (IoT), построение панелей управления (dashboards) с запаздыванием в секунды.
• Потоковая ETL (Extract-Transform-Load): Непрерывная загрузка данных из Apache Kafka, Amazon Kinesis или файловых систем с преобразованием (очистка, обогащение, схематизация) и записью в хранилища данных (например, Apache Hive, Delta Lake, Parquet/Avro).
• Машинное обучение на потоках: Вычисление признаков в реальном времени для моделей машинного обучения (например, обнаружение мошенничества, рекомендательные системы) с применением обученных моделей конвейера Spark MLlib.
• Обработка событий: Реализация простых событийно-ориентированных приложений, таких как уведомления, оповещения и маршрутизация сообщений.

¶Интеграция с источниками и приёмниками

Structured Streaming поддерживает множество источников (sources) и приёмников (sinks):

• Источники:
• kafka — чтение из Apache Kafka.
• kinesis — чтение из Amazon Kinesis Data Streams.
• rate — генератор псевдослучайных данных для тестирования и нагрузочного тестирования.
• socket — чтение из TCP-сокета (только для тестирования).
• file — чтение из файловой системы в форматах Parquet, JSON, CSV, ORC с отслеживанием появления новых файлов в указанном каталоге.
• Приёмники (Sinks):
• kafka — запись в Apache Kafka.
• console — вывод на консоль (для отладки).
• memory — хранение результата в оперативной памяти как внутренней таблицы.
• foreach / foreachBatch — вызов произвольной пользовательской логики для каждого пакета (например, запись в базу данных через JDBC).
• file — запись в файловую систему (Parquet, JSON, CSV, ORC).

¶Ограничения и критика

Несмотря на широкое распространение, Structured Streaming имеет ряд ограничений:

• Задержка: Микропакетная обработка (по умолчанию) вносит минимальную задержку в сотни миллисекунд (для небольших пакетов). Режим непрерывной обработки снижает её, но требует дополнительной настройки и поддерживает ограниченный набор операций (например, не поддерживает сложные агрегации с состоянием).
• Сложность управления состоянием: Для задач с большими и произвольными состояниями (например, скользящие окна произвольной длины) требуется эффективное управление памятью и сериализация.
• Обработка поздних событий: Механизм водяных знаков, хотя и решает проблему задержек, не гарантирует точную повторную обработку всех возможных опоздавших данных, особенно в условиях с нелинейными задержками.
• Производительность на высоких нагрузках: При очень высоких нагрузках (сотни тысяч событий в секунду) микропакетный подход может создавать накладные расходы на планирование и коммит контрольных точек. Альтернативы, такие как Apache Flink, могут демонстрировать более низкие накладные расходы на управление состоянием в таких сценариях.

¶Пример (без кода)

Типичный конвейер может выглядеть так: приложение получает поток событий от пользовательских кликов из Apache Kafka, фильтрует только события, содержащие покупки, группирует их по идентификатору товара в окне длительностью 5 минут и записывает результат в Amazon S3 в формате Parquet для последующего анализа в Amazon Athena или Databricks SQL. Все эти операции выполняются с гарантией «ровно один раз» и автоматической отказоустойчивостью при перезапуске из контрольной точки.

¶Источники

1. Документация Apache Spark: «Structured Streaming Programming Guide».
2. Armbrust, M., Das, T., Torres, J., et al. (2018). «Structured Streaming: A Declarative API for Real-Time Applications in Apache Spark». Proceedings of the 2018 International Conference on Management of Data (SIGMOD).
3. Apache Spark официальный репозиторий на GitHub — раздел «sql/core/src/main/scala/org/apache/spark/sql/execution/streaming».
4. Дастумы основных пакетов Databricks: «Best Practices for Streaming with Delta Lake».
5. Книга: Chambers, B., Zaharia, M. (2018). «Spark: The Definitive Guide». O'Reilly Media (Глава 13–15).