Движок распределённых вычислений: драйвер строит план из операций над DataFrame, разбивает его на стадии и задачи, менеджер кластера (YARN, Kubernetes, standalone) выделяет исполнители, каждый исполнитель обрабатывает свои партиции данных в памяти. Ленивые трансформации, выполнение по действию (write, count, collect), стадии разделяются shuffle. Catalyst оптимизирует план, Tungsten отвечает за память и кодогенерацию.
MapReduce записывает результат каждого шага на диск и строит задачу из жёсткой пары map/reduce, цепочка шагов это несколько проходов по диску. Spark держит промежуточные данные в памяти, строит DAG из многих операций и оптимизирует его целиком, поддерживает SQL, стриминг и ML в одном API. На практике Spark быстрее в разы и проще в разработке; MapReduce остался в легаси.
RDD это низкоуровневая распределённая коллекция объектов с функциональным API (map, reduceByKey), без схемы и без оптимизатора: Spark не знает, что внутри. DataFrame это таблица со схемой, операции выражаются декларативно (SQL или API), Catalyst оптимизирует план, Tungsten хранит данные компактно вне кучи JVM. В PySpark DataFrame ещё и быстрее RDD, потому что обработка идёт в JVM без сериализации в Python. RDD нужен только для нетипичной логики.
local[n] всё в одном процессе на машине разработчика. Standalone это собственный простой менеджер кластера Spark (master и workers) без YARN и Kubernetes. Режим деплоя client держит драйвер там, где запущен spark-submit (ноутбук, Airflow-воркер), cluster запускает драйвер внутри кластера; для продовых задач нужен cluster, чтобы падение клиентской машины не убивало job, а для интерактивной работы client.
row_number() без партиции собирает все данные в одну партицию на одном исполнителе: миллиард строк упадёт или будет считаться часами. Правильно: monotonically_increasing_id() даёт уникальные, но не сплошные номера (номер партиции в старших битах). Для сплошной нумерации: посчитать число строк в каждой партиции, вычислить смещение каждой партиции накопительной суммой, и к номеру внутри партиции (row_number по партиции или zipWithIndex) прибавить смещение. Это один проход плюс маленькая таблица смещений.
SparkSubmitOperator (или BashOperator с spark-submit) в режиме cluster, параметры конфигурации и дата периода передаются аргументами, зависимости по данным через сенсоры. Задача Airflow только запускает и ждёт завершения, код job живёт в репозитории и собирается в артефакт (jar, wheel, образ). Логи и статус в Spark UI/History Server, в Airflow ретраи и алерты. В Kubernetes вместо spark-submit часто SparkKubernetesOperator.
Узкие (map, filter, union, withColumn): каждая партиция результата зависит от одной партиции родителя, данные между узлами не ходят, операции конвейеризуются в один этап. Широкие (groupBy, join, distinct, repartition, оконные функции): партиция результата собирается из многих партиций родителя, поэтому нужен shuffle, запись промежуточных файлов и передача по сети. Граница между этапами (stage) в плане проходит ровно по широким трансформациям, и именно их число и объём определяют время джобы.
Запрос DataFrame или SQL проходит четыре фазы. Анализ: разрешение имён колонок и таблиц через каталог, проверка типов, получается логический план. Логическая оптимизация по правилам: predicate pushdown, column pruning, свёртка констант, упрощение выражений. Физическое планирование: для одного логического плана строится несколько физических (разные стратегии соединения) и выбирается лучший по стоимости. Кодогенерация: Tungsten генерирует байткод JVM для целого этапа (whole-stage codegen), чтобы не интерпретировать дерево выражений для каждой строки. Увидеть это можно через explain(True).
RDD хранит обычные объекты JVM: Spark не знает их структуру, не может оптимизировать операции, тратит память на заголовки объектов и нагружает сборщик мусора, а при shuffle сериализует через Java-сериализацию или Kryo. DataFrame имеет схему, поэтому Tungsten хранит строки в компактном бинарном формате в off-heap памяти, работает с ними без десериализации, использует кеш процессора эффективнее и генерирует специализированный код. Плюс Catalyst оптимизирует план целиком. Поэтому RDD оставляют для нетабличных данных и кастомной логики, а всё табличное пишут на DataFrame.
AQE перестраивает план во время выполнения по статистике, собранной после завершения shuffle-этапов. Три главные оптимизации: объединение мелких shuffle-партиций в более крупные, чтобы не плодить тысячи мелких задач; переключение sort-merge join на broadcast join, если одна сторона после фильтрации оказалась маленькой; разбиение перекошенных партиций при соединении на части. Включается spark.sql.adaptive.enabled=true (по умолчанию с 3.2) и снимает значительную часть ручной настройки spark.sql.shuffle.partitions.
При spark.speculation=true Spark замечает задачи, которые выполняются заметно дольше остальных в этапе, и запускает их копии на других узлах; побеждает та, что закончится первой, вторая убивается. Помогает против медленных узлов: больной диск, шумный сосед, проблемная сеть, особенно в облаке. Не помогает, если задача медленная из-за перекоса данных: копия будет обрабатывать ту же большую партицию. И может навредить, если задача делает побочные эффекты, которые нельзя повторять.
Сериализация переводит объекты в байты для shuffle, кэширования с сериализацией и передачи broadcast-переменных. Java-сериализация по умолчанию медленная и многословная, Kryo в разы быстрее и компактнее, включается spark.serializer=org.apache.spark.serializer.KryoSerializer, классы можно зарегистрировать заранее для экономии места. Для DataFrame без UDF это почти не важно: Tungsten хранит данные в своём бинарном формате. Kryo имеет значение для RDD, Dataset с объектами и UDF, возвращающих свои классы.
Через watermark: порог, насколько старые события по event time система ещё принимает в агрегации. Spark следит за максимальным временем события в потоке и вычитает заданную задержку (withWatermark("ts", "10 minutes")); окна старше порога закрываются и их состояние удаляется из state store, а события старше порога отбрасываются. Без watermark потоковая агрегация по окнам держала бы состояние бесконечно и упала бы по памяти. Выбор задержки это компромисс между полнотой данных и объёмом состояния.
Shuffle это пересылка строк между исполнителями так, чтобы строки с одним ключом оказались вместе: при groupBy, join, distinct, repartition, оконных функциях с partitionBy, сортировке. Это самая дорогая операция: запись на диск, сеть, граница стадии. Избежать: broadcast маленькой таблицы в join, предагрегация, партиционирование данных по ключу соединения при записи (bucketing), фильтры до соединения, меньше distinct. repartition(n) всегда делает shuffle и выравнивает партиции; coalesce(n) только склеивает партиции на том же исполнителе без shuffle, поэтому может дать перекос; shuffle при coalesce возникает, если число партиций увеличивают (coalesce с shuffle=true или repartition).
Broadcast Hash Join: маленькая сторона рассылается на все исполнители, без shuffle большой; Shuffle Hash Join: обе стороны перераспределяются по ключу, на каждой партиции строится хеш по меньшей; Sort Merge Join: обе стороны перераспределяются и сортируются, по умолчанию для больших таблиц; Broadcast Nested Loop для неравенств. Оптимизатор выбирает по оценке размера: если сторона меньше spark.sql.autoBroadcastJoinThreshold (10 МБ), broadcast. SortMerge вместо broadcast выбирается, когда оценка размера больше порога, статистики нет или таблица после фильтра кажется большой; лечится подсказкой broadcast(df) или повышением порога, если память исполнителей позволяет. Бродкастить ту сторону, которая меньше после фильтров и помещается в память каждого исполнителя.
Узкая операция обрабатывает партицию независимо (filter, map), широкая требует данных из многих партиций и порождает shuffle (join, groupBy). При соединении двух больших таблиц Spark считает хеш ключа, перераспределяет обе таблицы так, чтобы одинаковые ключи попали в одну партицию, сортирует партиции по ключу и сливает их (Sort Merge Join). Отсюда стоимость: две записи на диск и пересылка по сети, и отсюда же уязвимость к перекосу ключа.
Несколько ключей встречаются в разы чаще остальных, после shuffle их партиции огромны, и одна задача работает, пока остальные ждут; в UI видно по разбросу времени и объёма задач в стадии и по «хвостовой» задаче. Лечение: AQE (spark.sql.adaptive.skewJoin.enabled) сам режет перекошенные партиции; «соль» к ключу на большой стороне и размножение маленькой на число солей; broadcast маленькой стороны; вынос и отдельная обработка NULL-ключей и топ-ключей; предагрегация.
Открыть Spark UI: найти стадию, посмотреть распределение задач (перекос), объём shuffle read/write, spill, GC. Посмотреть физический план (explain): какой join выбран, есть ли лишний Exchange, не перепутан ли broadcast большой таблицы. Типовые причины: перекос ключа, SortMergeJoin вместо broadcast из-за отсутствия статистики, слишком мало партиций после shuffle (spark.sql.shuffle.partitions), взрыв строк из-за неуникального ключа, маленькие файлы на входе.
repartition(n) или repartition(col) делает полный shuffle и раскладывает данные равномерно, им можно увеличить число партиций или разложить данные по ключу перед записью. coalesce(n) только уменьшает число партиций, склеивая соседние на тех же узлах без shuffle, поэтому дешевле, но даёт неравномерные партиции и, если применить слишком рано, сокращает параллелизм всего предыдущего этапа. Типичный приём: перед записью coalesce до разумного числа файлов, а при перекосе или увеличении параллелизма repartition.
Маленькая таблица собирается на драйвере и рассылается в память каждого исполнителя, после чего соединение идёт локально без shuffle. Это ломается, если «маленькая» таблица превышает spark.sql.autoBroadcastJoinThreshold (10 МБ по умолчанию) и оптимизатор выбирает sort-merge, или если её принудительно broadcast-ят хинтом, а она не помещается в память драйвера или исполнителей: получается OOM на драйвере при сборе или долгие сборки мусора на исполнителях. Размер оценивается по статистике и после фильтра может быть неизвестен, поэтому помогает ANALYZE TABLE, AQE, который переключает стратегию по фактическому размеру, и ручной хинт только для таблиц, чей размер известен.
Партиционирование при записи раскладывает данные по каталогам по значению колонки (year=2024/month=05): подходит для колонок с малым числом значений, читатель пропускает ненужные каталоги (partition pruning). Бакетирование делит данные внутри партиции на фиксированное число файлов по хешу колонки с высокой кардинальностью (user_id): две таблицы, бакетированные по одному ключу и числу бакетов, соединяются без shuffle, агрегация по ключу тоже обходится без него. Бакетирование работает только для таблиц в каталоге (Hive, Spark) и требует, чтобы писатель и читатель были Spark; партиции понимают все движки.
Оптимизация Spark 3 для соединения большой партиционированной таблицы фактов с небольшим измерением, на которое наложен фильтр. Без неё Spark читал бы все партиции фактов, потому что условие стоит на измерении. С DPP Spark сначала вычисляет отфильтрованные ключи измерения (обычно через broadcast), а затем подставляет их как фильтр партиций при чтении фактов, читая только нужные каталоги. Работает, когда ключ соединения совпадает с колонкой партиционирования фактов.
Меньше данных в shuffle: фильтровать и отбирать колонки до join и groupBy, предпочитать агрегации с локальной предредукцией (reduceByKey вместо groupByKey для RDD, обычные агрегаты DataFrame уже так работают). Правильное число партиций: spark.sql.shuffle.partitions по умолчанию 200, его подбирают под объём, целясь в партиции по 100–200 МБ, или отдают AQE. Избегать shuffle целиком: broadcast join для маленькой стороны, бакетирование для повторяющихся соединений по одному ключу, repartition по ключу один раз перед несколькими операциями. И следить за перекосом, потому что один тяжёлый ключ сводит на нет остальное.
Память исполнителя (spark.executor.memory) делится на резерв (300 МБ), пользовательскую часть под объекты кода и унифицированную область (spark.memory.fraction, 60%), внутри которой execution (shuffle, join, сортировка) и storage (кэш) берут память друг у друга; плюс память вне кучи (memoryOverhead, важна для PySpark). Расчёт: оценить объём данных после фильтров и число партиций, взять 2–5 ядер на исполнитель, память так, чтобы партиция помещалась с запасом, число исполнителей по параллелизму и квоте очереди; включить динамическую аллокацию, чтобы не держать ресурсы во время простоя. Проверять в UI spill и GC и корректировать.
Без кэша каждая ветка пересчитает DataFrame с нуля. cache() / persist(level) сохраняет результат после первого действия. Смотреть: помещается ли он в память (MEMORY_AND_DISK безопаснее MEMORY_ONLY, иначе часть партиций вытеснится и пересчитается), не держит ли кэш память, нужную для shuffle, вызван ли unpersist после использования. Кэшировать после фильтров, а не до; для больших промежуточных результатов надёжнее записать в Parquet и читать заново.
Фильтры и выбор колонок до соединений (pushdown в Parquet), broadcast маленьких таблиц, борьба с перекосом (AQE, соль), правильное число партиций (shuffle.partitions, repartition перед записью, чтобы не плодить маленькие файлы), кэш переиспользуемых DataFrame, избегание UDF на Python в пользу встроенных функций, партиционирование данных по дате при записи, динамическая аллокация. В UI: длительность стадий, разброс задач, shuffle read/write, spill, GC time, план с Exchange.
Режим, в котором Spark запрашивает исполнители у менеджера кластера по мере появления задач в очереди и освобождает простаивающие после таймаута. Нужен внешний shuffle service или его аналог, чтобы данные shuffle пережили удаление исполнителя. Плюс: не держать ресурсы в простое, лучше делить кластер; минус: задержка на запуск исполнителей и непредсказуемое время при перегруженном кластере.
Одно Spark-приложение работает в одном кластере; «мультикластер» на практике означает либо запуск джобов на нескольких кластерах через оркестратор (Airflow с разными подключениями), либо чтение данных из разных источников (HDFS одного кластера, S3 другого) одним приложением, либо федерацию через Kubernetes. Честный ответ: уточнить, что имеется в виду, и описать маршрутизацию задач по кластерам оркестратором с общим хранилищем данных в S3.
collect() или toPandas() на большом DataFrame тянут все данные в память драйвера; broadcast слишком большой таблицы собирает её на драйвере перед рассылкой; очень большой план или метаданные (десятки тысяч партиций, тысячи колонок, глубокая цепочка трансформаций в цикле) раздувают память драйвера. Избегать: take, limit, запись результата в хранилище вместо сбора, контроль размера broadcast, checkpoint для обрезания длинного плана, увеличение spark.driver.memory только когда причина понятна.
Перекос данных: одна партиция после shuffle не помещается в память, особенно при оконных функциях и collect_list по тяжёлому ключу. Слишком много ядер на исполнителе: каждая задача получает долю памяти, при 10 ядрах на 20 ГБ на задачу остаётся мало. Большие объекты в UDF, сложные вложенные структуры, кэширование десериализованных RDD. Лечат увеличением числа партиций, чтобы каждая стала меньше, борьбой с перекосом, разумным числом ядер (обычно до пяти), spark.executor.memoryOverhead для off-heap и Python-процессов при PySpark, и пересмотром кэша.
Когда execution-памяти исполнителя не хватает для сортировки, агрегации или хеш-таблицы соединения, Spark сбрасывает часть промежуточных данных на локальный диск и потом читает обратно. Это видно в Spark UI как Spill (Memory) и Spill (Disk) у этапа, и это один из главных признаков медленной джобы. Уменьшают его увеличением памяти на задачу (больше spark.executor.memory или меньше ядер), увеличением числа партиций, чтобы каждая обрабатывалась меньшей порцией, заменой sort-merge join на broadcast, устранением перекоса и удалением лишних колонок до тяжёлых операций.
cache() и persist(level) сохраняют данные в памяти или на локальном диске исполнителей, но цепочка трансформаций (lineage) остаётся: при потере исполнителя партиция пересчитывается заново. checkpoint() записывает данные в надёжное хранилище (HDFS, S3) и обрезает lineage: дальнейший план начинается с этих данных как с нового источника. Нужен для итеративных алгоритмов и длинных циклов, где план разрастается и драйвер тратит память и время на его анализ, а также для Structured Streaming, где checkpoint хранит смещения и состояние для восстановления после сбоя.
Parquet колоночный бинарный формат со схемой и сжатием: файл делится на row groups, внутри каждой колонка хранится отдельным chunk со статистикой (min/max, число null), словарным кодированием и сжатием. Движок читает только нужные колонки и пропускает row groups по статистике. CSV это текст: нет типов, нет сжатия по колонкам, читать нужно целиком и парсить каждое значение; он нужен только для обмена с людьми и системами, которые ничего другого не понимают. Parquet обычно в 5–10 раз меньше CSV и в разы быстрее для аналитики.
Spark режет вход на сплиты размером spark.sql.files.maxPartitionBytes (128 МБ по умолчанию) с учётом границ row group и файлов: примерно сумма размеров файлов / 128 МБ, маленькие файлы объединяются до openCostInBytes. Число row group влияет на границы: row group не делится между задачами. Поэтому много мелких файлов и мелкие row groups дают много задач с накладными расходами, а один огромный row group не параллелится.
Каждый файл это как минимум одна задача и обращение к метаданным: тысячи файлов по мегабайту дают тысячи коротких задач, планирование и открытие файлов занимает больше времени, чем чтение. На HDFS мелкие файлы ещё и съедают память NameNode. Лечение: при записи repartition/coalesce до разумного числа файлов (100–500 МБ каждый), компакция существующих данных отдельной задачей, в Iceberg/Delta встроенная компакция. В S3 проблема мягче (нет NameNode), но накладные расходы на запросы к объектам остаются.
Оба колоночные со сжатием и статистикой. ORC родом из Hive: stripe вместо row group, встроенные индексы по строкам и bloom-фильтры, лучше интеграция с Hive и ACID-транзакциями Hive. Parquet родом из экосистемы Hadoop/Twitter, стал стандартом де-факто для Spark, Arrow, Iceberg, DuckDB и облачных сервисов. На практике выбирают Parquet, кроме случаев, когда хранилище живёт в Hive.
Текстовые: CSV, JSON, JSON Lines (просто, без схемы, большие). Строковые бинарные: Avro (схема, эволюция схемы, хорош для Kafka и построчной записи). Колоночные: Parquet, ORC (аналитика, сжатие, чтение нужных колонок). Табличные форматы поверх файлов: Iceberg, Delta, Hudi (транзакции, версии, схема). Колоночные выигрывают на аналитике за счёт чтения части колонок, сжатия однотипных данных и пропуска блоков по статистике, проигрывают на построчных записях и обновлениях.
Оптимизатор спускает условия фильтра к источнику данных, чтобы они применились при чтении, а не после загрузки в память. Для баз данных через JDBC фильтр уходит в SQL-запрос. Для Parquet и ORC работает статистика: каждая группа строк хранит min и max по колонкам, и если условие не попадает в диапазон, группа пропускается без чтения с диска; вместе с отбором только нужных колонок это и даёт колоночным форматам преимущество. Для CSV и JSON pushdown не работает: файл всё равно читается и разбирается целиком. Pushdown не срабатывает, если колонка обёрнута в функцию или UDF, поэтому фильтры стоит писать по голым колонкам.