Transformations vs Actions: анатомия ленивых вычислений

В уроке 3 мы познакомились с Lazy Evaluation концептуально. Теперь разберём механику вглубь: как именно Spark строит DAG, что происходит при нескольких Actions, и какие ловушки расставляет «ленивое» выполнение для невнимательного разработчика. Этот урок - фундамент, без которого невозможно осознанно профилировать и оптимизировать Spark-пайплайны.

Философия: почему Spark «ленивый»?¶

Два мира: императивный и декларативный¶

Прежде чем разбирать механику, важно понять, откуда вообще взялась эта архитектура. Существуют два принципиально разных подхода к вычислениям.

Императивный подход - это то, как работает классический Python или Pandas. Вы пишете инструкции, и каждая из них выполняется немедленно, строчка за строчкой. Интерпретатор не знает, что будет написано в следующей строке, поэтому не может «заглянуть вперёд» и оптимизировать.

# Pandas: императивный, eager execution
df = pd.read_csv("events.csv")       # читает ВСЕ колонки сразу
df = df[df["status"] == "active"]    # создаёт копию в RAM немедленно
df = df[["user_id", "revenue"]]      # ещё одна копия
result = df.groupby("user_id").sum() # ещё одна копия

Каждая строка немедленно создаёт новый объект в памяти. На ноутбуке с файлом в 100 МБ это работает прекрасно. Но когда данные - 5 ТБ на кластере из 50 машин, такой подход ломается: нет центрального места, которое «держит» всю таблицу, а выполнение каждого шага немедленно требует синхронизации между узлами.

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

# Spark: декларативный, lazy evaluation
df = spark.read.parquet("s3a://data/events/")  # только открывает метаданные
df = df.filter(col("status") == "active")      # добавляет узел в граф
df = df.select("user_id", "revenue")           # добавляет узел в граф
df = df.groupBy("user_id").sum("revenue")      # добавляет узел в граф
# Пока ничего не вычислено. Spark держит в памяти лишь "рецепт"

Ключевое слово - рецепт. Spark не готовит блюдо при добавлении каждого ингредиента. Он записывает рецепт целиком, а потом готовит всё сразу - наиболее эффективным способом.

Что даёт лень оптимизатору¶

Пока план не выполнен, оптимизатор Catalyst видит всю цепочку операций целиком. Это позволяет ему делать то, что невозможно при немедленном выполнении:

Перестановка операций. Если вы написали join() перед filter(), но Catalyst видит, что фильтр можно применить раньше - он переставит их. Фильтрация до join уменьшает объём обрабатываемых данных на порядки. При eager execution вы уже выполнили join - делать фильтрацию «до» него невозможно.

Устранение ненужных колонок. Если в конце пайплайна нужны только 3 из 40 колонок, Catalyst может сказать Parquet reader: «читай только эти три». Это называется column pruning - и оно работает только потому, что Spark знает всю цепочку заранее.

Объединение операций в один проход. Вместо того чтобы делать filter, потом withColumn, потом ещё один filter тремя отдельными проходами по данным - Spark «сплавляет» их в единый цикл. Каждая строка проходит через все три операции за один проход по памяти.

Выбор стратегии join. Если в конце пайплайна выяснилось, что одна из таблиц после фильтрации стала маленькой - Catalyst может заменить дорогой Shuffle Hash Join на дешёвый Broadcast Join. Но это возможно только если план строится целиком, а не построчно.

Левый путь - то, что вы написали в коде. Правый - то, что Spark реально выполняет после оптимизации. Разница в скорости может быть 10–100× на реальных данных.

Жизненный цикл Spark-запроса¶

Прежде чем говорить о трансформациях и actions, важно понять полный цикл от вашего кода до физического выполнения. Это знание позволит осмысленно читать explain() и Spark UI.

Разберём каждый шаг подробно.

Unresolved Logical Plan. Каждый вызов трансформации добавляет узел в дерево. На этом этапе Spark не знает, существует ли колонка user_id в источнике - он лишь записывает намерение. Попытка обратиться к несуществующей колонке пока не является ошибкой.

Analyzed Logical Plan. Spark обращается к Catalog (метаданные источников данных), разрешает имена колонок, проверяет типы данных. Если колонка user_id не существует - ошибка возникает именно здесь, не при вызове трансформации.

Optimized Logical Plan. Catalyst применяет десятки правил оптимизации к логическому плану. Правила делятся на две категории: rule-based (применяются всегда: predicate pushdown, column pruning, constant folding) и cost-based (требуют статистик: join reordering, broadcast threshold). Результат - более эффективный логический план с теми же семантикой и результатом.

Physical Plans. Catalyst генерирует несколько вариантов физического плана. Для каждого join, например, рассматриваются: SortMergeJoin (универсальный, требует сортировки), BroadcastHashJoin (быстрый, работает только если одна таблица помещается в память), ShuffleHashJoin (хэш-join без сортировки). Выбор зависит от размеров данных и настроек.

WholeStageCodegen. Выбранный физический план компилируется в «сырой» Java-байткод. Вместо цепочки виртуальных вызовов методов Spark генерирует один единственный цикл, который делает все операции над каждой строкой. Это убирает накладные расходы на вызовы методов и делает узкие трансформации почти бесплатными.

Execution. Только на этом этапе данные реально читаются из источника. Executors получают Tasks - каждый Task обрабатывает одну партицию данных, применяя весь скомпилированный план к каждой строке.

Ключевой момент: шаги 1–6 происходят на Driver. Executors задействуются только на шаге 7. Именно поэтому explain() выполняется мгновенно - он строит план, но не читает данных.

Полный каталог трансформаций¶

Трансформация - это операция, которая принимает один или несколько DataFrame/RDD и возвращает новый DataFrame/RDD. При вызове трансформации ничего не вычисляется - добавляется узел в логический граф (DAG lineage).

Трансформации делятся на два принципиально разных класса по типу зависимости между входными и выходными партициями.

Narrow трансформации - без shuffle¶

Узкая зависимость означает: каждая выходная партиция зависит ровно от одной входной партиции. Данные не перемещаются между узлами кластера. Все операции над одной партицией выполняются одним Executor без координации с остальными.

Это делает narrow трансформации фундаментально дешёвыми: нет сетевого обмена, нет записи на диск, нет ожидания других Executors.

Каждый Executor обрабатывает свою партицию независимо. Ни байта данных не пересекает сетевую границу.

Важное замечание про coalesce(n). Эта функция - пограничный случай. Когда вы уменьшаете число партиций (например, с 200 до 10), Spark просто объединяет несколько партиций на одном Executor без перемещения данных по сети. Но если попытаться увеличить число партиций через coalesce - это невозможно без shuffle, поэтому для увеличения нужен repartition.

Важное замечание про union(). В отличие от SQL UNION, union() в Spark работает по позиции колонок, а не по именам. Если схемы двух DataFrame совпадают по структуре, но колонки переставлены - union молча перепутает данные. Используйте unionByName() если схемы могут отличаться.

Wide трансформации - через shuffle¶

Широкая зависимость означает: выходная партиция зависит от нескольких входных партиций, возможно из разных узлов кластера. Данные физически перемещаются по сети - это называется shuffle.

Одинаковые ключи user_id могут быть в разных партициях на разных узлах. Чтобы агрегировать их вместе, Spark должен собрать все записи с одинаковым ключом в одном месте. Это требует сетевого обмена - shuffle.

Shuffle - самая дорогостоящая операция в Spark по нескольким причинам:

• Запись на диск: перед shuffle Executor записывает свои данные на локальный диск (shuffle write). Это защита от потери данных при сбое.
• Чтение с диска: после shuffle Executor читает данные, предназначенные для него, с дисков других Executors (shuffle read).
• Сетевой обмен: данные передаются по сети между узлами кластера.
• Сортировка: большинство shuffle-алгоритмов требуют сортировки данных по ключу.
• Барьер синхронизации: второй Stage не может начаться, пока первый полностью не завершился и не записал shuffle files.

Особенность orderBy(). Глобальная сортировка требует, чтобы все данные были собраны в правильном порядке. Это означает shuffle со специальным алгоритмом: Spark сначала семплирует данные для определения границ диапазонов, а затем делает range partitioned shuffle. Это особенно дорого на больших данных. Если глобальная сортировка не критична - предпочитайте sortWithinPartitions(), которая сортирует только внутри каждой партиции без shuffle.

Особые трансформации¶

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

cache() и persist(). Технически это трансформации - они не запускают вычисления немедленно. Они лишь помечают DataFrame для кэширования при первом Action. Реальное кэширование происходит при первом выполнении. Это контринтуитивно: вызов .cache() не означает «сейчас закэшировать» - он означает «закэшировать при следующем Action».

explode(col). Узкая трансформация, которая «взрывает» массив: каждый элемент массива становится отдельной строкой. Если строка содержит массив из 5 элементов - explode превращает её в 5 строк. Это может существенно увеличить размер DataFrame.

from pyspark.sql.functions import explode, col

# До explode: 3 строки
# | user_id | tags             |
# | 1       | [spark, python]  |
# | 2       | [spark, scala]   |
# | 3       | [hadoop]         |

df.select("user_id", explode("tags").alias("tag"))

# После explode: 5 строк
# | user_id | tag    |
# | 1       | spark  |
# | 1       | python |
# | 2       | spark  |
# | 2       | scala  |
# | 3       | hadoop |

from_json() и to_json(). Трансформации для работы с JSON-строками. from_json(col, schema) парсит JSON-строку в StructType. to_json(col) сериализует StructType обратно в JSON-строку. Парсинг - дорогая операция, поэтому по возможности храните данные в нативных форматах (Parquet, ORC) вместо JSON-строк.

Higher-order functions (transform, filter, aggregate из pyspark.sql.functions). Работают с элементами массивов и словарей без explode. Выполняются на JVM, поэтому существенно быстрее Python UDF для аналогичных задач.

from pyspark.sql.functions import transform, filter as spark_filter

# transform: применить функцию к каждому элементу массива (не explode)
df.select(transform("scores", lambda x: x * 2).alias("doubled_scores"))

# filter: оставить только элементы массива, удовлетворяющие условию
df.select(spark_filter("tags", lambda x: x != "deprecated").alias("active_tags"))

Полный каталог Actions¶

Action - это операция, которая запускает выполнение всего накопленного DAG. Action либо возвращает данные в Driver (Python-процесс на вашей машине), либо записывает данные во внешнее хранилище. После каждого Action DAG выполняется полностью - с нуля, если результат не был закэширован.

Понимание Actions критично для оптимизации: каждый Action - потенциально дорогая операция.

Возвращают данные в Driver¶

О collect() и toPandas(). Оба метода переносят весь DataFrame на одну машину - Driver. Если в DataFrame 100 млн строк по 1 КБ каждая, это 100 ГБ на Driver. Driver типично имеет 4–16 ГБ RAM. Результат - OutOfMemoryError. Используйте эти методы только для небольших агрегированных результатов (тысячи строк, не миллионы).

Возвращают агрегат, не данные¶

О count(). Этот Action выглядит безобидно, но он сканирует все партиции данных. Если ваш DataFrame читается из S3 и содержит 10 ТБ - count() прочитает все 10 ТБ. Для отладки используйте df.limit(1000).count() вместо df.count(). В продакшн-коде count() оправдан только если число строк действительно нужно.

О show(). Многие разработчики думают, что show(10) - это «просто просмотреть несколько строк, это быстро». На самом деле show(n) запускает полноценный Spark Job. Если ваш DataFrame строится через сложный join и groupBy - show(10) выполнит весь этот join и groupBy, а потом покажет 10 строк результата. Разница с take(10) только в том, что show красиво форматирует вывод и не возвращает данные в Python.

Об isEmpty(). Это оптимизированный Action: Spark останавливает выполнение после нахождения первой строки. Он не сканирует весь DataFrame. Используйте его вместо count() == 0 для проверки на пустоту.

Записывают данные во внешнее хранилище¶

Важно о записи. Методы write.*() - это Actions, которые запускают выполнение DAG. До вызова write.parquet() ничего не записывается. Это означает: если ваш пайплайн завершится ошибкой до write - на диске не будет никаких частичных данных (если используется режим записи overwrite или атомарный формат вроде Delta Lake).

Выполняют действие на каждой строке (на Executors)¶

Использование foreachPartition. Предпочтительнее foreach, когда нужно делать что-то с каждой строкой и это дорого инициализировать (например, открывать соединение с базой данных). С foreachPartition соединение открывается один раз на партицию, а не на каждую строку.

def write_to_postgres(rows):
    # Соединение открывается один раз на партицию
    conn = psycopg2.connect("host=db user=spark...")
    cursor = conn.cursor()
    for row in rows:
        cursor.execute("INSERT INTO events VALUES (%s, %s)", (row["id"], row["value"]))
    conn.commit()
    conn.close()

df.foreachPartition(write_to_postgres)  # эффективно
# vs
df.foreach(lambda row: write_row_to_postgres(row))  # N соединений, неэффективно

Специальный случай: explain()¶

explain() - это не совсем Action в традиционном смысле. Он не запускает выполнение DAG и не читает данные. Но он выводит результат немедленно, поэтому визуально похож на Action. На самом деле он запускает только первые 6 шагов жизненного цикла (до физического плана) и печатает результат.

df.explain()                    # краткий физический план
df.explain(mode="formatted")    # читаемый форматированный план
df.explain(mode="extended")     # полный план: logical + physical
df.explain(mode="codegen")      # сгенерированный Java код
df.explain(mode="cost")         # с оценками стоимости (если есть статистики)

DAG под капотом: пайплайнинг и Operator Fusion¶

Что такое DAG и зачем он нужен¶

DAG (Directed Acyclic Graph, направленный ациклический граф) - это формальное представление вашего пайплайна в виде графа, где узлы - это операции (трансформации), а рёбра - зависимости между ними.

Слово «ациклический» важно: граф не может иметь циклов. Это фундаментальное свойство, которое позволяет:

• Восстанавливать данные без репликации: если Executor упал и потерял свою партицию - Spark просто повторно выполняет часть DAG для этой партиции. Не нужно хранить резервную копию данных.
• Оптимизировать глобально: без циклов можно применять топологическую сортировку и глобальные правила оптимизации.
• Строить Stage-ы: широкие трансформации (shuffle) разделяют DAG на Stage-ы, которые выполняются последовательно.

Как Spark разбивает DAG на Stage-ы. DAG Scheduler смотрит на граф и находит границы shuffle (Exchange operators). Каждый Exchange создаёт новый Stage. Stage-ы выполняются последовательно: Stage 2 не может начаться, пока Stage 1 не записал все свои shuffle files. Внутри Stage все операции над одной партицией выполняются в один проход - это и есть пайплайнинг.

Пайплайнинг и Operator Fusion¶

Внутри одного Stage Spark объединяет несколько операций в один физический цикл выполнения. Это называется Operator Fusion или пайплайнинг. Вместо того чтобы:

1. Пройти по всем строкам партиции, применить filter(), сохранить результат в промежуточный буфер
2. Пройти по промежуточному буферу, применить withColumn(), сохранить
3. Пройти по результату, применить второй filter(), сохранить
4. Пройти ещё раз, применить select(), сохранить

Spark делает всё в один проход:

• Читает строку из партиции
• Применяет filter() - если строка не прошла, берём следующую
• Применяет withColumn() - вычисляем новое значение
• Применяет второй filter() - проверяем условие
• Применяет select() - оставляем нужные колонки
• Передаём результирующую строку на shuffle write

Это даёт огромную выгоду по памяти: промежуточные буферы не создаются вообще. Строки текут через пайплайн по одной.

Whole-Stage Codegen: *(N) в explain()¶

Пайплайнинг идёт ещё дальше. Вместо цепочки вызовов виртуальных методов (которые в JVM требуют dynamic dispatch) Spark генерирует Java-байткод, объединяющий все операции в одну скомпилированную функцию.

В выводе explain() это обозначается звёздочкой: *(1) Project, *(1) Filter. Все операции с одинаковым числом в скобках компилируются в один сгенерированный класс. Exchange (shuffle) разрывает кодоген-цикл.

result.explain(mode="formatted")

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[user_id#10L], functions=[sum(amount_usd#11)])
   +- Exchange hashpartitioning(user_id#10L, 200)          ← граница кодогена
      +- HashAggregate(keys=[user_id#10L], functions=[partial_sum(amount_usd)])
         +- Project [user_id#10L, (amount#12 / rate#13) AS amount_usd#11]  ← *(1)
            +- Filter (event_type = purchase AND (amount / rate) > 0.0)     ← *(1)
               +- FileScan parquet [...]                                     ← *(1)
                    PushedFilters: [IsNotNull(event_type), EqualTo(event_type,purchase)]
                    ReadSchema: struct<user_id:bigint,amount:double,rate:double>

Читается снизу вверх:

1. FileScan parquet - исток данных. PushedFilters означает, что часть фильтров передана прямо в Parquet reader: строки не читаются в JVM, если уже понятно, что они не пройдут фильтр. ReadSchema показывает, что читаются только 3 из 40 колонок (column pruning).

2. Filter - часть фильтров, которую нельзя передать в Parquet reader (например, вычисляемые условия). Выполняется в JVM.

3. Project - withColumn() и select(). Вычисляет новую колонку amount_usd и оставляет только нужные колонки.

4. Exchange - граница Stage-ов. Здесь данные сортируются по user_id и записываются на диск для shuffle.

5. HashAggregate (partial) - частичная агрегация до shuffle. Если в одной партиции есть записи с одинаковым user_id, их суммы считаются ещё до отправки по сети. Это называется Map-side combiner - значительно уменьшает объём данных в shuffle.

6. HashAggregate (final) - финальная агрегация после shuffle, когда все записи с одним user_id уже собраны вместе.

Два HashAggregate - это не ошибка и не неэффективность. Это намеренная оптимизация: частичная агрегация сокращает объём shuffle write в разы.

Catalyst и Lazy Evaluation: синергия оптимизаций¶

Рассмотрим три ключевых оптимизации Catalyst, которые возможны только благодаря Lazy Evaluation.

Predicate Pushdown: фильтруем до чтения¶

Predicate Pushdown - перенос условий фильтрации (filter(), where()) как можно «ближе» к источнику данных, вплоть до передачи их в формат хранения.

# Что вы написали:
df = spark.read.parquet("s3a://data/events/")
df = df.filter(col("country") == "RU")
df = df.filter(col("event_date") >= "2024-01-01")
df.write.parquet("s3a://result/")

Без оптимизации Spark прочитал бы все данные из Parquet, передал в JVM, и только там применил фильтры. С Predicate Pushdown:

• Фильтр по event_date - если данные партиционированы по дате, Spark не читает партиции за прошлые периоды вообще (Partition Pruning).
• Фильтр по country - Parquet хранит статистики (min/max) для каждого row group. Если в row group нет записей country = 'RU' (min > 'RU' или max < 'RU') - этот блок пропускается без чтения (Statistics-based row group skipping).
• Оставшиеся фильтры передаются в виде PushedFilters в Parquet reader - строки отбрасываются ещё до передачи в JVM.

Практический эффект: чтение 10% данных вместо 100% при правильной партиционированности и хорошей статистике.

Column Pruning: читаем только нужные колонки¶

Если в конце пайплайна нужны только 4 колонки из 80 - Catalyst передаёт Parquet reader информацию: «читай только эти 4 колонки». Parquet - колоночный формат, поэтому остальные 76 колонок даже не читаются с диска.

# 80 колонок в таблице, но нужны только 4
result = df.select("user_id", "revenue", "country", "event_date")
           .groupBy("country")
           .agg(sum("revenue"))

Catalyst видит весь план: в финальном результате нужны только country и revenue. Колонка user_id нужна только в промежуточных операциях. В ReadSchema вы увидите только 4 колонки, не 80.

Практический эффект: Parquet-файл с 80 колонками, но если запрос использует только 4 - чтение в 10–20 раз быстрее.

Join Reordering: порядок join имеет значение¶

Если вы джоините три таблицы: большую (10 ТБ), среднюю (100 ГБ) и маленькую (10 МБ) - порядок join определяет сложность вычислений.

# Написали так - большая × средняя, потом результат × маленькая
result = large_df.join(medium_df, "key").join(small_df, "id")

# Catalyst может переставить так - маленькая переводится в broadcast,
# large × broadcast_small, потом × medium - меньше shuffle

Catalyst использует статистики о размерах таблиц (если они доступны через ANALYZE TABLE или собираются автоматически в Delta Lake/Iceberg) для выбора оптимального порядка join. Без Lazy Evaluation это было бы невозможно: первый join уже выполнен к тому моменту, когда мы узнаём о втором.

Ловушка №1: Двойная работа при нескольких Actions¶

Почему Spark не помнит результаты¶

По умолчанию Spark - stateless движок. Когда Action завершился, вычисленные данные не сохраняются автоматически. Они живут в памяти Executors только в процессе выполнения. После завершения Action - память освобождается.

Это означает: следующий Action на том же DataFrame перечитает данные из источника и выполнит все трансформации заново.

# Строим тяжёлый граф: join + groupBy + несколько фильтров
heavy_df = (
    raw_events
    .filter(col("status") == "active")
    .join(enrichment_df, "user_id")          # дорогой join: shuffle
    .join(geo_df, "ip_address")              # ещё один join: ещё shuffle
    .groupBy("region", "category")
    .agg(
        count("*").alias("cnt"),
        sum("revenue").alias("total_revenue"),
        avg("session_duration").alias("avg_session")
    )
)

# ❌ S3 читается дважды, оба join выполняются дважды, groupBy - дважды
print(f"Строк в отчёте: {heavy_df.count()}")          # Action 1: полный DAG
heavy_df.write.parquet("s3a://bucket/daily_report/")  # Action 2: полный DAG заново

В Spark UI вы увидите два отдельных Job с идентичными Stage-ами. Вся работа сделана дважды.

Цена ошибки на реальном кластере. Предположим, кластер стоит $10/час, пайплайн занимает 20 минут ($3.33). Два Action без кэширования = два пайплайна = $6.66. С кэшированием - $3.33 + небольшой overhead на materialization. На 100 запусков в месяц разница = $333. На больших данных (час вместо 20 минут) - $1000+.

Решение: persist() обрывает Lineage¶

# ✅ Вычисляем один раз, используем дважды
heavy_df = (
    raw_events
    .filter(col("status") == "active")
    .join(enrichment_df, "user_id")
    .join(geo_df, "ip_address")
    .groupBy("region", "category")
    .agg(
        count("*").alias("cnt"),
        sum("revenue").alias("total_revenue"),
        avg("session_duration").alias("avg_session")
    )
    .persist()  # помечаем для кэширования
)

# Action 1: выполняет DAG полностью, сохраняет результат в память Executors
print(f"Строк в отчёте: {heavy_df.count()}")

# Action 2: читает из памяти Executors, не идёт к источнику
heavy_df.write.parquet("s3a://bucket/daily_report/")

# Освобождаем память явно - не ждём сборщика мусора
heavy_df.unpersist()

Важные нюансы persist():

• persist() сам по себе - lazy трансформация. Реальное кэширование происходит при первом Action.
• По умолчанию persist() без аргументов использует уровень MEMORY_AND_DISK: сначала в RAM, при нехватке - на диск Executor.
• cache() - синоним persist() без аргументов.
• unpersist() нужно вызывать явно. Spark самостоятельно инвалидирует кэш при нехватке памяти (LRU eviction), но явный unpersist() освобождает память немедленно.

Уровни хранения¶

Когда кэшировать: дерево решений¶

DataFrame используется в 2+ Actions?
├── НЕТ → не кэшировать (overhead на сериализацию не оправдан)
└── ДА:
    └── Вычисление дорогое (join, groupBy, dropDuplicates)?
        ├── НЕТ → скорее всего не стоит (пересчитать может быть быстрее)
        └── ДА:
            └── Данные помещаются в RAM?
                ├── ДА → persist(MEMORY_ONLY) или cache()
                └── НЕТ → persist(MEMORY_AND_DISK) или пересчитывать

Ловушка №2: Отложенные ошибки и «логические бомбы»¶

Ошибки в трансформациях не возникают сразу¶

Поскольку трансформации не выполняются при вызове, ошибки в данных (деление на ноль, несовместимые типы, нарушение схемы) не возникают при написании трансформации. Они «прячутся» до первого Action.

# Строка 15: трансформация с потенциальной ошибкой - никакой ошибки нет
df_with_ratio = df.withColumn("conversion_rate", col("purchases") / col("visits"))
# Если в данных есть строки где visits=0 - здесь ничего не произойдёт

# Строка 16: ещё одна трансформация - ещё ничего
df_enriched = df_with_ratio.filter(col("conversion_rate") > 0.01)

# ...50 строк кода позже...

# Строка 100: первый Action - ВОТ ЗДЕСЬ взрыв произойдёт,
# потому что Spark попытается вычислить 0/0 = NaN или выбросит исключение
df_enriched.write.parquet("s3a://result/")
# SparkException: Division by zero at line 15 in your_script.py

Это делает отладку Spark-кода сложнее, чем отладку обычного Python. Стектрейс укажет на строку с Action, но ошибка концептуально находится в трансформации намного раньше.

Как защититься:

# ✅ Защита от деления на ноль
from pyspark.sql.functions import when, col

df_with_ratio = df.withColumn(
    "conversion_rate",
    when(col("visits") > 0, col("purchases") / col("visits")).otherwise(0.0)
)

# ✅ Проверяйте данные на выборке перед полным запуском
sample = df.limit(10000)
sample_ratio = sample.withColumn("conversion_rate", col("purchases") / col("visits"))
sample_ratio.show(5)  # быстрая проверка на маленьких данных
# Если здесь проблема - увидим быстро, не через 40 минут full job

Схемы: ошибки до Action, не во время¶

В отличие от ошибок в данных, ошибки в схеме (обращение к несуществующей колонке) обнаруживаются раньше - на этапе Analyzed Logical Plan, который строится при вызове трансформации. Это происходит потому что Catalyst проверяет схему немедленно.

df = spark.read.parquet("s3a://data/events/")
# df имеет колонки: user_id, event_type, amount

# ❌ Ошибка будет здесь, при вызове трансформации - не при Action
df_bad = df.filter(col("typo_column") > 0)
# AnalysisException: Column 'typo_column' not found in [user_id, event_type, amount]

# ❌ Ошибка в данных - только при Action
df_ok = df.withColumn("ratio", col("amount") / col("user_id"))  # нет ошибки синтаксиса
df_ok.count()  # здесь может быть ошибка если user_id = 0

Ловушка №3: Side effects в трансформациях¶

Почему print() внутри map() не работает как ожидается¶

Код внутри map(), filter(), withColumn() (через UDF), foreach() выполняется на Executors - процессах на удалённых машинах. Driver (ваша Python-сессия) не получает их stdout напрямую.

В локальном режиме (local[*]) вы увидите вывод, потому что Driver и Executors - один процесс. Это маскирует проблему. В production-кластере вывод уходит в логи Executor на удалённых серверах.

Non-idempotent операции в трансформациях¶

Spark повторяет Tasks при сбое. Если Task упал и был перезапущен - функция внутри трансформации вызовется повторно для тех же данных. Это называется retry. Spark гарантирует exactly-once семантику только для записи (с некоторыми оговорками), но не для кода внутри трансформаций.

Что это означает практически:

# ❌ HTTP-запрос внутри трансформации - может вызваться 2, 3, N раз
def enrich_from_api(row):
    response = requests.post("https://api.example.com/enrich", json={"id": row["id"]})
    return Row(**row.asDict(), **response.json())

enriched = df.rdd.map(enrich_from_api).toDF()
enriched.write.parquet("result/")
# Если Task упадёт на 90% - повторный запрос к API, деньги списываются дважды?

# ❌ Вставка в базу данных внутри трансформации - дубликаты
def insert_to_db(row):
    db.execute("INSERT INTO events VALUES (...)")

df.foreach(insert_to_db)  # может вставить дважды при retry

# ✅ Правильно: используйте idempotent операции
# - Upsert (INSERT OR REPLACE, MERGE INTO) вместо INSERT
# - Idempotency keys на стороне API
# - foreachPartition с явным управлением транзакцией

Мутация глобальных переменных¶

# ❌ Не работает: bad_count останется 0 на Driver
bad_count = 0

def validate_row(row):
    global bad_count
    if row["amount"] < 0:
        bad_count += 1  # обновляется КОПИЯ на Executor, не оригинал на Driver
    return row

df.rdd.foreach(validate_row)
print(bad_count)  # всегда 0

# ✅ Правильно: Accumulator для распределённых счётчиков
bad_count = spark.sparkContext.accumulator(0)

def validate_row(row):
    if row["amount"] < 0:
        bad_count.add(1)  # thread-safe, агрегируется на Driver

df.foreach(validate_row)
print(f"Некорректных строк: {bad_count.value}")  # читается на Driver

Ловушка №4: Action в цикле¶

Анатомия катастрофы¶

Запуск Action внутри Python-цикла - один из самых частых антипаттернов. Каждая итерация цикла создаёт отдельный Spark Job. Каждый Job включает: DAG-планирование, выделение ресурсов, отправку Tasks на Executors, сбор результатов, освобождение ресурсов. Overhead одного Job - сотни миллисекунд даже для тривиальных вычислений.

# ❌ Антипаттерн: 30 Spark Jobs в Spark UI
dates = pd.date_range("2024-01-01", "2024-01-30").strftime("%Y-%m-%d").tolist()

for date in dates:
    # Каждая итерация: отдельный Job, отдельный S3 read, отдельный filter
    daily_count = df.filter(col("event_date") == date).count()
    print(f"{date}: {daily_count} событий")
# Общее время: 30 × (time_per_job + overhead) ≈ 30 × 5 сек = 150 сек

# ✅ Правильно: один Job, одна агрегация
from pyspark.sql.functions import count

result = (
    df.filter(col("event_date").between("2024-01-01", "2024-01-30"))
    .groupBy("event_date")
    .agg(count("*").alias("daily_count"))
    .orderBy("event_date")
    .collect()
)

for row in result:
    print(f"{row['event_date']}: {row['daily_count']} событий")
# Общее время: 1 Job ≈ 8 сек (группировка сложнее, но overhead один)

В Spark UI разница будет наглядной: 30 отдельных Job против 1. При этом один Job с groupBy будет быстрее 30 отдельных filter+count даже если данных немного - просто за счёт устранения overhead.

Когда цикл с Actions всё-таки допустим¶

Не любой цикл - антипаттерн. Цикл оправдан, когда:

• Разные итерации обрабатывают принципиально разные источники (разные файлы, разные таблицы)
• Логика итераций нельзя выразить как агрегацию (например, итеративный алгоритм ML)
• Количество итераций мало (2–5) и каждая итерация неизбежна

# Допустимо: обрабатываем разные источники, нет общей агрегации
for table_name in ["users", "products", "orders"]:
    df = spark.read.parquet(f"s3a://lake/{table_name}/")
    cleaned = clean_table(df)
    cleaned.write.mode("overwrite").parquet(f"s3a://cleaned/{table_name}/")

# Недопустимо: однотипная обработка одного источника
for category in categories:  # ❌ если categories > 5
    df.filter(col("category") == category).count()

Ловушка №5: collect() и OOM на Driver¶

Почему collect() - самая опасная операция¶

collect() (и toPandas()) транспортируют все данные из распределённого хранилища (Executors, расположенные на десятках машин) в одну машину - Driver. Driver - это обычно ваш ноутбук или head-node кластера с 8–32 ГБ RAM.

Правило безопасного использования collect():

Используйте collect() только когда вы точно знаете, что результат мал. «Мал» - это субъективно, но практическое правило: если результат содержит меньше миллиона строк и каждая строка < 1 КБ (< 1 ГБ суммарно), collect() обычно безопасен.

# ❌ Опасно: collect() на большом DataFrame
all_events = df.collect()  # если df = 10 ТБ → OOM

# ✅ Безопасно: collect() только на агрегированных результатах
summary = df.groupBy("region").agg(sum("revenue")).collect()
# Если регионов 20 → 20 строк → безопасно

# ✅ Безопасно: toPandas() на небольшой выборке
sample = df.limit(10000).toPandas()  # явно ограничиваем

toPandas() - отдельная история¶

toPandas() делает collect() внутри, плюс конвертирует данные в Pandas DataFrame. Кроме риска OOM есть ещё один нюанс: при конвертации некоторые типы данных теряют точность (например, Spark DecimalType → Python float). При работе с финансовыми данными это критично.

# ❌ Плохо: конвертируем весь DataFrame
df.toPandas()

# ✅ Хорошо: сначала агрегируем, потом конвертируем
df.groupBy("product").agg(sum("revenue").alias("total")).toPandas()

# ✅ Хорошо: Arrow для быстрой конвертации (если нужно всё)
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
sampled_df = df.sample(fraction=0.01).toPandas()  # выборка

Ловушка №6: Data Skew невидим до Action¶

Lazy Evaluation скрывает дисбаланс данных до момента Shuffle. Вы не узнаете о data skew из трансформаций - Spark не читает данные и не знает распределения ключей. Проблема проявляется только при Action, когда один Executor обрабатывает 80% данных, а остальные простаивают.

# Эта трансформация не покажет skew - данные не читаются
df_grouped = df.groupBy("city").agg(sum("revenue").alias("total"))

# Skew обнаруживается в Spark UI → Stage → Tasks:
# Task для "Москва": 15 минут   (70% данных на одном Executor)
# Task для "Казань": 45 секунд
# Task для "Новгород": 30 секунд
# ...

# ✅ Диагностируйте распределение ДО запуска дорогой агрегации
key_distribution = df.groupBy("city").count().orderBy(col("count").desc())
key_distribution.show(20)

# Если топ-1 ключ >> остальных - нужен salting или AQE skew handling:
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")

Подробно о борьбе с data skew - в модуле оптимизации. Здесь важно запомнить: проверяйте распределение ключей до запуска дорогих агрегаций.

Checkpoint vs Cache: когда обрывать Lineage¶

Обе техники материализуют промежуточные результаты, но делают это по-разному и решают разные проблемы.

Проблема длинного Lineage¶

Если вы выполняете много итераций (например, в ML-алгоритме) и каждая итерация добавляет трансформации в граф, Lineage может вырасти до тысяч узлов. Это создаёт два проблемы:

1. Планирование медленное: Catalyst должен обработать весь граф при каждом Action.
2. Восстановление катастрофично: если Executor упадёт, Spark попытается восстановить данные, повторно выполнив весь Lineage с самого начала.

# Итеративный алгоритм: каждая итерация удваивает глубину DAG
rdd = sc.parallelize(range(1000))
for i in range(100):  # 100 итераций
    rdd = rdd.map(lambda x: x + 1)  # DAG растёт на 1 узел каждую итерацию

rdd.count()  # планирование 100-узлового DAG - медленно

cache() vs checkpoint()¶

# checkpoint() требует настройки директории
spark.sparkContext.setCheckpointDir("hdfs:///spark-checkpoints/")

# В итеративном алгоритме: checkpoint каждые N итераций
rdd = sc.parallelize(range(1000))
for i in range(100):
    rdd = rdd.map(lambda x: x + 1)
    if i % 10 == 0:  # каждые 10 итераций
        rdd = rdd.checkpoint()  # обрываем Lineage
        rdd.count()  # материализуем (checkpoint тоже lazy!)

Практика: профилирование и рефакторинг¶

Кейс: пайплайн, который работает 40 минут¶

Рассмотрим реалистичный «плохой» пайплайн с несколькими типичными ошибками:

# ❌ Плохой пайплайн - 40 минут, 15 Jobs в Spark UI

from pyspark.sql.functions import col, sum, count, avg

# 1. Читаем данные
events = spark.read.parquet("s3a://data-lake/events/year=2024/")
users = spark.read.parquet("s3a://data-lake/users/")
products = spark.read.parquet("s3a://data-lake/products/")

# 2. Проверяем что данные есть - ОШИБКА: ненужный Action
print(f"Events: {events.count()}")   # Action #1: сканирует весь S3
print(f"Users: {users.count()}")     # Action #2: сканирует S3
print(f"Products: {products.count()}")  # Action #3: сканирует S3

# 3. Основной пайплайн - тяжёлые join-ы
enriched = events.join(users, "user_id").join(products, "product_id")

# 4. Ещё одна проверка - ОШИБКА: join выполнится первый раз
print(f"После join: {enriched.count()}")   # Action #4: full join + scan

# 5. Фильтрация и агрегация
filtered = enriched.filter(col("status") == "active")
revenue_by_region = filtered.groupBy("region").agg(
    sum("revenue").alias("total"),
    count("*").alias("cnt"),
    avg("session_duration").alias("avg_session")
)

# 6. Проверяем результат - ОШИБКА: join выполнится второй раз
revenue_by_region.show()  # Action #5: full DAG заново

# 7. Записываем результат - ОШИБКА: join выполнится третий раз
revenue_by_region.write.mode("overwrite").parquet("s3a://result/revenue/")  # Action #6

# 8. Отдельные агрегации по категориям - ОШИБКА: цикл с Actions
for category in ["electronics", "clothing", "food", "sports"]:
    cat_stats = filtered.filter(col("category") == category).agg(
        sum("revenue"),
        count("*")
    )
    cat_stats.write.mode("overwrite").parquet(f"s3a://result/{category}/")
    # Action #7, #8, #9, #10: join выполняется 4 раза

В Spark UI: 10 отдельных Jobs, полный join events×users×products выполняется 5+ раз.

Рефакторинг: 8 минут, 2 Jobs¶

# ✅ Оптимизированный пайплайн

from pyspark.sql.functions import col, sum, count, avg

# 1. Читаем данные
events = spark.read.parquet("s3a://data-lake/events/year=2024/")
users = spark.read.parquet("s3a://data-lake/users/")
products = spark.read.parquet("s3a://data-lake/products/")

# 2. Основной пайплайн с кэшированием - join будет выполнен ОДИН раз
enriched = (
    events
    .filter(col("status") == "active")       # фильтруем ДО join - меньше данных
    .join(users, "user_id")
    .join(products, "product_id")
    .persist()  # помечаем для кэширования
)

# 3. Все агрегации в рамках одного DAG
revenue_by_region = (
    enriched
    .groupBy("region")
    .agg(
        sum("revenue").alias("total"),
        count("*").alias("cnt"),
        avg("session_duration").alias("avg_session")
    )
)

# Агрегации по категориям - через pivot, не цикл!
revenue_by_category = (
    enriched
    .filter(col("category").isin(["electronics", "clothing", "food", "sports"]))
    .groupBy("category")
    .agg(
        sum("revenue").alias("total"),
        count("*").alias("cnt")
    )
)

# 4. Action #1: материализует кэш + записывает регионы
# (первый Action на enriched - кэш заполняется здесь)
revenue_by_region.write.mode("overwrite").parquet("s3a://result/revenue/")

# 5. Action #2: категории из кэша - join НЕ повторяется
for row in revenue_by_category.collect():
    category = row["category"]
    # пишем на S3 от Driver - тут суммарно мало строк
    spark.createDataFrame([row]).write.mode("overwrite").parquet(f"s3a://result/{category}/")

# 6. Освобождаем кэш
enriched.unpersist()

# Диагностика: убеждаемся, что план разумен
print(f"Регионов: {revenue_by_region.count()}")  # маленький count - допустим

Что изменилось:

• filter("status" == "active") перенесён до join - уменьшает объём join в разы
• Убраны диагностические count() в начале - они не давали полезной информации
• persist() добавлен после join - дорогая операция выполняется один раз
• Цикл с Actions заменён на groupBy().agg() + collect() на маленьком результате
• unpersist() явно освобождает память

Чтение Spark UI: что искать¶

После запуска откройте http://localhost:4040 (или ссылку из spark.sparkContext.uiWebUrl):

print(spark.sparkContext.uiWebUrl)  # прямая ссылка на Spark UI

Итог: правила работы с ленивым Spark¶

Чек-лист инженера¶

Перед запуском любого пайплайна:

• [ ] explain(mode="formatted") - проверить физический план: Column Pruning работает? Фильтры pushed down? Join-стратегия разумна?
• [ ] Проверить количество Actions: больше одного на одном DataFrame без persist() - потенциальная двойная работа
• [ ] Проверить: нет ли Actions в Python-циклах? Заменить на groupBy() или filter().isin()
• [ ] Проверить: try/except оборачивает Actions, а не трансформации
• [ ] Проверить: нет ли collect() / toPandas() на потенциально больших DataFrame

Кэширование:

• [ ] persist() добавлен после дорогих операций (join, groupBy, dropDuplicates), если результат используется в 2+ Actions
• [ ] unpersist() вызывается явно после использования кэша
• [ ] Уровень хранения выбран осознанно: MEMORY_AND_DISK для надёжности, MEMORY_ONLY для скорости

Side effects:

• [ ] Нет print(), HTTP-запросов, DB insert внутри map(), filter(), UDF - только идемпотентные операции
• [ ] Для счётчиков используются Accumulator, а не глобальные переменные Python
• [ ] foreachPartition() вместо foreach() для операций с соединениями к внешним системам

После запуска:

• [ ] Spark UI открыт: Jobs → Stages → Tasks. Skew? Spill? GC overhead? Неожиданно большой Input?
• [ ] Количество Jobs в Spark UI соответствует ожидаемому количеству Actions в коде

В следующем уроке разберём Narrow vs Wide трансформации ещё подробнее: почему groupBy() несравнимо дороже filter(), как shuffle организован физически на уровне файловой системы и как читать Shuffle Read/Write в Spark UI.