Apache Spark: как устроена распределенная обработка и когда она нужна

Прямой ответ

Apache Spark — движок распределённой обработки, который делит данные на партиции и выполняет операции на многих узлах параллельно. Главная абстракция — RDD (resilient distributed dataset), а для практической работы почти всегда используют DataFrame/Dataset поверх Catalyst-оптимизатора и Tungsten: они дают план выполнения, предикат-пушдаун и колоночное исполнение.

Ключевые принципы, без которых Spark выглядит «магией»:

1. Ленивость. Трансформации (filter, groupBy, join) не выполняются сразу — строится граф. Работа начинается на действии (count, write, collect).
2. Этапы (stages) и границы shuffle. Этап заканчивается там, где требуется перемешивание данных между узлами — например, при groupBy или join не по партиционному ключу. Shuffle — самая дорогая операция: данные пишутся на диск и пересылаются по сети.
3. Партиционирование определяет параллелизм. Число партиций задаёт, сколько задач выполнится одновременно. Слишком крупные партиции — нехватка памяти и долгие задачи, слишком мелкие — накладные расходы на планирование.
4. Узкие (narrow) и широкие (wide) зависимости. Узкие (map, filter) не требуют обмена между узлами и выполняются в одном этапе; широкие требуют shuffle.

Когда Spark оправдан: объёмы, не помещающиеся в память одной машины; повторяющиеся пакетные расчёты по большим наборам; необходимость читать данные из распределённого хранилища. Когда не оправдан: задачи, которые решаются на одной машине за секунды, — там DuckDB или pandas/polars окажутся быстрее и дешевле в эксплуатации.

Практика: типовые операции и настройки

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .appName("orders-daily")
         .config("spark.sql.shuffle.partitions", "400")
         .config("spark.sql.adaptive.enabled", "true")
         .getOrCreate())

orders = spark.read.parquet("s3a://warehouse/orders/")
daily = (orders
         .filter(F.col("created_at") >= "2026-10-01")
         .groupBy(F.col("created_at").cast("date").alias("d"), "region")
         .agg(F.sum("total").alias("revenue"), F.countDistinct("user_id").alias("users")))
daily.write.mode("overwrite").partitionBy("d").parquet("s3a://warehouse/agg_orders/")

Практические рекомендации:

1. Включайте AQE. spark.sql.adaptive.enabled=true позволяет переупаковывать мелкие партиции и менять стратегию join по факту, что часто убирает ручную подстройку shuffle.partitions.
2. Не вызывайте collect() на больших данных. Это пересылает весь результат в драйвер — типичная причина OOM. Для отладки используйте limit(20).toPandas().
3. Broadcast join для малых таблиц. Если одна сторона меньше spark.sql.autoBroadcastJoinThreshold, Spark разослает её всем узлам и избежит shuffle. Проверить стратегию можно через df.explain("formatted").
4. Формат и сжатие. Parquet с предикат-пушдауном и партиционированием по дате сокращает объём чтения в разы — выбор формата разобран в статье про CSV и Parquet.
5. Память executor. Разделение spark.executor.memory на execution/storage и overhead задаётся через spark.memory.fraction и spark.executor.memoryOverhead; при OOM чаще помогает увеличить overhead и число партиций, а не «добавить памяти».

Как читать Spark UI без магии. Три вкладки дают 80 % ответов. На вкладке SQL виден физический план с временем каждого оператора и числом строк на входе и выходе — именно там заметно, что фильтр применяется после соединения, а не до него. На вкладке Stages видно длительность задач внутри этапа: если медиана 5 секунд, а максимум 8 минут, это перекос. На вкладке Executors видны промахи по памяти и время в garbage collection: доля GC выше 10 % означает, что память executor настроена неверно либо партиции слишком крупные.

Практический сценарий: задача зависает на стадии shuffle

Контекст: ночной пайплайн агрегирует заказы за месяц; обычно 12 минут, сегодня висит час.

1. Диагностика. В Spark UI видно: один этап с 400 задачами, 398 завершились, две выполняются более 50 минут и обрабатывают объём в 20 раз больше остальных. Это перекос данных (skew): по популярному региону приходится непропорционально много строк.
2. Проверка плана. explain() показывает SortMergeJoin вместо ожидаемого broadcast: таблица-справочник подросла и превысила порог автотрансляции.
3. Исправления. (а) Повышаем порог трансляции либо явно указываем F.broadcast(dim); (б) включаем AQE с spark.sql.adaptive.skewJoin.enabled=true, чтобы перекошенные партиции дробились автоматически; (в) для «горячего» ключа применяем соление ключа — добавление случайного суффикса перед агрегацией с последующим досчётом.
4. Проверка. Перезапуск: длительность этапа выравнивается, общее время возвращается к 11 минутам. Фиксируем изменения в коде пайплайна и в dbt-моделях, если агрегация описана там.
5. Профилактика. Добавляем метрику длительности задачи-максимума по этапам и алерт на перекос; оркестрация и повторные запуски — в статье про Airflow.

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

FAQ

Чем DataFrame отличается от RDD? RDD — низкоуровневая абстракция без схемы; оптимизатор не может ничего предположить о данных. DataFrame несёт схему, поэтому Catalyst строит оптимизированный план, применяет предикат-пушдаун и колоночное исполнение. Для новых проектов RDD используют редко.

Как выбрать число партиций? Ориентир — размер партиции 64–256 МБ в распакованном виде и число задач, кратное числу ядер executor. Для shuffle число задаётся spark.sql.shuffle.partitions, и при включённом AQE мелкие партиции объединяются автоматически.

Можно ли использовать Spark для потоковой обработки? Да, Structured Streaming строится на том же движке и обрабатывает поток микропачками (или непрерывно в отдельных режимах). Задержки при этом выше, чем у специализированных потоковых движков, но код остаётся единым для пакетной и потоковой обработки.

Нужен ли Spark, если есть колоночная СУБД? Не всегда. Если данные уже лежат в колоночном хранилище и объёмы помещаются в кластер, агрегации выгоднее выполнять там: СУБД использует индексы и оптимизатор, а Spark придётся читать данные заново. Spark оправдан для сложных трансформаций и подготовки данных перед загрузкой.

Источники: Spark docs (RDD guide, SQL guide, tuning) — ссылки в шапке.