Narrow vs Wide трансформации: shuffle изнутри

Partition dependencies, полный каталог narrow/wide трансформаций по документации Spark, механика shuffle в трёх фазах, reduceByKey vs groupByKey, Broadcast Join и coalesce vs repartition.

core internals optimization

В предыдущем уроке мы разобрали, как Lazy Evaluation и DAG позволяют Spark оптимизировать план до выполнения. Теперь погрузимся в самый важный архитектурный раскол: деление трансформаций на Narrow и Wide. Это деление определяет количество Stage в вычислительном графе, объём данных в сети и итоговую производительность пайплайна.

Урок состоит из семи частей: теория зависимостей партиций → полный каталог каждой трансформации с примерами → физика Shuffle изнутри → критический антипаттерн groupByKey → техники обхода Shuffle → лабораторная практика → домашнее задание.

1. Партиции - физическая основа распределённости¶

Что такое партиция физически¶

Когда Spark читает файл или создаёт DataFrame, данные делятся на партиции - независимые фрагменты, каждый из которых обрабатывается одним Task на одном Executor. Партиция - не абстракция, а реальный физический кусок данных: набор строк Parquet-файла, блок HDFS, часть результата JDBC-запроса.

Число партиций при чтении определяется следующими факторами:

для HDFS/S3: одна партиция ≈ один блок (по умолчанию 128 MB)
для JDBC: параметр numPartitions в настройках подключения
для sc.parallelize(): параметр numSlices
после shuffle: значение spark.sql.shuffle.partitions (по умолчанию 200)

df = spark.read.parquet("s3a://data/events/")
print(df.rdd.getNumPartitions())  # узнать число партиций

# Принудительно задать при чтении
df = spark.read.option("minPartitions", 400).csv("s3a://data/logs/")

# Посмотреть распределение данных по партициям
df.rdd.mapPartitions(lambda it: [sum(1 for _ in it)]).collect()

Вычислительная модель Spark строится вокруг партиций: каждый Task обрабатывает ровно одну. Больше партиций - выше параллелизм, но и выше overhead на запуск Tasks. Практическое правило: 128–256 MB данных на одну партицию.

Узкие зависимости (Narrow Dependencies)¶

Narrow dependency - зависимость, при которой каждая выходная партиция зависит от строго одной входной. Данные не покидают Executor, не передаются по сети, обработка полностью изолирована.

Каждая входная партиция порождает строго одну выходную. Task 1 работает только с Partition 1 и ничего не знает о Partition 2 и 3. Все Tasks запускаются параллельно и полностью независимо.

Несколько narrow-трансформаций подряд Spark объединяет в один Stage и выполняет за один проход по данным. filter → withColumn → select - это один Stage, один проход по каждой партиции, никаких промежуточных операций на диск. Это называется пайплайнинг (operator fusion).

Широкие зависимости (Wide Dependencies)¶

Wide dependency - зависимость, при которой одна выходная партиция зависит от множества входных. Чтобы посчитать сумму продаж по ключу user_id = "alice", нужно собрать все строки с alice из всех партиций - они физически разбросаны по разным Executor-ам.

Чтобы посчитать агрегат по ключу A, нужны строки с A из Partition 1 И из Partition 2. Физически они на разных машинах - Spark перемещает данные по сети (Shuffle). Это принципиальное следствие широкой зависимости.

Граница Stage: как Shuffle делит граф вычислений¶

Каждая wide-трансформация создаёт новый Stage в DAG. Stage - набор Tasks, которые можно выполнить без промежуточного обмена данными через сеть. Граница Stage - всегда точка Shuffle.

В этом примере три Stage и два Shuffle Exchange - два сетевых перемещения данных. Каждый Shuffle стоит дорого: запись на диск + сетевой трафик + чтение с диска на другом Executor. Понимание этой модели позволяет, глядя на код, предсказывать число Stage и находить лишние Shuffle до запуска.

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

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

RDD API: Narrow-трансформации¶

`map(f)` - поэлементное преобразование¶

Применяет функцию f к каждому элементу RDD и возвращает новый RDD того же числа элементов. Это фундаментальная операция - аналог Python map(), но выполняется параллельно на сотнях узлов кластера.

Почему Narrow: функция применяется к каждому элементу независимо. Элемент из партиции N остаётся в партиции N - данные с других партиций не нужны, никаких пересечений нет.

rdd = sc.parallelize(range(1, 6), numSlices=3)

# Базовый пример: удвоить каждый элемент
doubled = rdd.map(lambda x: x * 2)
# [2, 4, 6, 8, 10]

# Парсинг строк из текстового файла
raw_lines = sc.textFile("hdfs:///data/events/*.csv")
parsed = raw_lines \
    .map(lambda line: line.split(",")) \
    .map(lambda f: {
        "user_id": int(f[0]),
        "event": f[1].strip(),
        "ts": int(f[2]),
    })

# Обогащение данных через closure (константный словарь замкнут в функции)
category_map = {"001": "electronics", "002": "clothing", "003": "food"}
enriched = products_rdd.map(
    lambda p: (p["id"], category_map.get(p["cat_id"], "unknown"), p["price"])
)

Важно: map(f) всегда возвращает ровно столько же элементов, сколько было во входном RDD. Если нужно удалить элементы - filter(), если развернуть один в несколько - flatMap().

`flatMap(f)` - поэлементное преобразование с разворачиванием¶

Работает как map(f), но если функция возвращает список (или любой итерируемый объект), результат разворачивается в отдельные элементы. Количество выходных элементов может быть больше или меньше входных.

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

# Разбить каждую строку на слова - классика Word Count
lines = sc.textFile("hdfs:///data/text/")
words = lines.flatMap(lambda line: line.lower().split())
# "Hello World" → ["hello", "world"] - два элемента из одного

# Разворачивание вложенных структур
user_tags = sc.parallelize([
    {"user": "alice", "tags": ["spark", "python", "bigdata"]},
    {"user": "bob",   "tags": ["java", "kafka"]},
])
# Получить пары (user, tag) - каждый тег отдельной строкой
pairs = user_tags.flatMap(
    lambda u: [(u["user"], tag) for tag in u["tags"]]
)
# [("alice","spark"), ("alice","python"), ("alice","bigdata"),
#  ("bob","java"), ("bob","kafka")]

# Фильтрация через flatMap: вернуть [] для ненужных элементов
valid = events_rdd.flatMap(
    lambda e: [e] if e["status"] == "success" else []
)

`mapPartitions(f)` - преобразование всей партиции целиком¶

Вместо обработки каждого элемента отдельно, передаёт функции весь итератор партиции целиком. Функция получает Iterator[T] и возвращает Iterator[U]. Это самая важная оптимизация для операций с высоким overhead инициализации - подключения к БД, HTTP-клиенты, ML-модели.

Почему Narrow: каждая партиция обрабатывается независимо, без взаимодействия с другими.

import psycopg2
from typing import Iterator

def enrich_from_db(partition: Iterator[dict]) -> Iterator[dict]:
    """Обогатить записи из PostgreSQL - одно соединение на партицию."""
    conn = psycopg2.connect(host="pg-host", dbname="crm", user="spark")
    cur = conn.cursor()
    try:
        for row in partition:
            cur.execute(
                "SELECT segment FROM users WHERE id = %s", (row["user_id"],)
            )
            result = cur.fetchone()
            yield {**row, "segment": result[0] if result else "unknown"}
    finally:
        cur.close()
        conn.close()

enriched_rdd = events_rdd.mapPartitions(enrich_from_db)

# Загрузка тяжёлой ML-модели - один раз на партицию, не на каждую строку
def predict_batch(partition: Iterator[dict]) -> Iterator[dict]:
    import pickle
    # Модель загружается один раз при инициализации партиции
    model = pickle.load(open("/models/churn_model.pkl", "rb"))
    for row in partition:
        features = [row["recency"], row["frequency"], row["monetary"]]
        yield {**row, "churn_prob": float(model.predict_proba([features])[0][1])}

predictions = customers_rdd.mapPartitions(predict_batch)

Используйте mapPartitions() всякий раз, когда инициализация ресурса дорогостояща. Для простых CPU-трансформаций map() достаточно и читается проще.

`mapPartitionsWithIndex(f)` - партиция с её номером¶

Идентичен mapPartitions(f), но функция получает дополнительный первый аргумент - индекс партиции (0, 1, 2, …). Используется когда нужно знать, в какой партиции выполняется Task.

Почему Narrow: обработка каждой партиции строго изолирована.

def process_with_index(partition_idx: int, partition) -> Iterator:
    for row in partition:
        yield {**row, "partition_id": partition_idx}

result = data_rdd.mapPartitionsWithIndex(process_with_index)

# Классический трюк: пропустить заголовок CSV
# Заголовок всегда попадает в партицию 0 при чтении одного файла
def skip_header(idx, rows):
    if idx == 0:
        next(rows, None)  # пропустить первую строку
    return rows

data_rdd = raw_rdd.mapPartitionsWithIndex(skip_header)

# Диагностика: подсчитать строки в каждой партиции
sizes = (
    rdd.mapPartitionsWithIndex(
        lambda i, it: [(i, sum(1 for _ in it))]
    ).collect()
)
print(sizes)  # [(0, 1200), (1, 800), (2, 1500), ...] - виден skew!

`filter(f)` - фильтрация элементов по предикату¶

Применяет предикат f к каждому элементу и возвращает только те, где f вернула True. Один из самых часто используемых методов. Главное правило: ставьте filter() как можно раньше в пайплайне - до любых wide-трансформаций.

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

# Базовая фильтрация
active_users = users_rdd.filter(lambda u: u["status"] == "active")

# Комбинирование условий
valid_events = events_rdd.filter(
    lambda e: e["ts"] >= 1700000000
              and e["amount"] > 0
              and e["user_id"] is not None
)

# Эффективный паттерн: map + filter в одном проходе
result = raw_rdd \
    .map(lambda line: line.split(",")) \
    .filter(lambda f: len(f) == 5 and f[2] != "") \
    .map(lambda f: {"user_id": f[0], "amount": float(f[2])})

Золотое правило: filter раньше join. Уменьшить таблицу до join значит уменьшить объём Shuffle на несколько порядков.

`union(rdd)` - объединение двух RDD¶

Создаёт новый RDD, содержащий все элементы обоих RDD. Spark просто «дописывает» список партиций второго RDD к партициям первого - никакого перемещения данных по сети нет.

Почему Narrow: это операция на уровне метаданных - Driver добавляет указатели на партиции второго RDD. Данные физически не перемещаются.

# Объединение логов за два периода
logs_jan = sc.textFile("hdfs:///logs/2024-01/")
logs_feb = sc.textFile("hdfs:///logs/2024-02/")
all_logs = logs_jan.union(logs_feb)

# Объединение нескольких источников через reduce
from functools import reduce
rdd_list = [sc.textFile(f"hdfs:///logs/{m}/") for m in months]
combined = reduce(lambda a, b: a.union(b), rdd_list)

Важно: union() не удаляет дубликаты. Если нужна дедупликация - distinct() после, но это уже Wide-трансформация.

`sample(withReplacement, fraction, seed)` - случайная выборка¶

Возвращает случайный набор элементов с заданной долей. Работает через Bernoulli sampling (без замены) или Poisson sampling (с заменой) - решение для каждого элемента принимается локально.

Почему Narrow: каждый элемент решает сам, попадать ли в выборку, только на основе случайного числа и fraction. Никакой глобальной координации нет.

# 10% выборка без замены, воспроизводимо
sample = big_rdd.sample(withReplacement=False, fraction=0.1, seed=42)

# Стратифицированная выборка по ключу (тоже narrow!)
fractions = {"RU": 0.1, "US": 0.05, "DE": 0.2}
stratified = rdd.sampleByKey(withReplacement=False, fractions=fractions)

`glom()` - партиция в список¶

Преобразует каждую партицию в список Python. Каждый элемент результирующего RDD - это полный список всей партиции. Число партиций остаётся прежним.

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

rdd = sc.parallelize(range(10), 3)
glommed = rdd.glom().collect()
# [[0, 1, 2, 3], [4, 5, 6], [7, 8, 9]] - три списка

# Диагностика: размер каждой партиции
sizes = rdd.glom().map(len).collect()
# [4, 3, 3]

# Групповые операции без shuffle - только если данные уже нужно сгруппированы
stats = rdd.glom().map(lambda part: {
    "count": len(part),
    "sum":   sum(part),
    "avg":   sum(part) / len(part) if part else 0,
})

`mapValues(f)` и `flatMapValues(f)` - трансформация только значений¶

Для RDD пар (K, V) применяет функцию только к значению V, оставляя ключ K нетронутым. Это критически важное свойство: ключ не меняется → партиционирование сохраняется. Если RDD уже партиционирован по ключу (например, после join()), следующий join() по тому же ключу может обойтись без Shuffle.

Почему Narrow: ключ не изменяется, партиция определяется тем же ключом.

# mapValues: нормализовать суммы, сохранив ключи
sales_rdd = sc.parallelize([("alice", 100), ("bob", 200), ("alice", 50)])
normalized = sales_rdd.mapValues(lambda v: round(v / 1000, 3))
# [("alice", 0.1), ("bob", 0.2), ("alice", 0.05)]

# flatMapValues: развернуть список значений, сохранив ключ
tags_rdd = sc.parallelize([
    ("alice", ["spark", "python"]),
    ("bob",   ["java"]),
])
flat = tags_rdd.flatMapValues(lambda tags: tags)
# [("alice","spark"), ("alice","python"), ("bob","java")]

# Ключевое преимущество: если RDD уже партиционирован по ключу,
# mapValues сохраняет партиционирование - следующий join без shuffle!
partitioned = rdd.partitionBy(100)
result = partitioned.mapValues(expensive_transform)  # narrow, партиционирование сохранено
result.join(other.partitionBy(100))  # Spark видит совпадающее партиционирование → без shuffle

`keys()` / `values()` - извлечение ключей или значений¶

Для RDD пар (K, V) возвращают RDD только ключей или только значений. Это синтаксический сахар над map(lambda kv: kv[0]) и map(lambda kv: kv[1]).

Почему Narrow: трансформация каждой пары независима от других.

pairs = sc.parallelize([("a", 1), ("b", 2), ("c", 3)])
keys   = pairs.keys()    # RDD: ["a", "b", "c"]
values = pairs.values()  # RDD: [1, 2, 3]

# Практический пример: получить только ключи для проверки
word_index = words_rdd.zipWithIndex()
word_list  = word_index.keys().collect()

`zip(rdd)` - попарное объединение двух RDD¶

Объединяет элементы двух RDD в пары (elem1, elem2). Аналог Python zip(). Требует строго одинакового числа партиций и элементов в каждой партиции - иначе исключение.

Почему Narrow: пара формируется из элементов одной и той же партиции. Никакого межпартиционного взаимодействия нет.

features = sc.parallelize([1.0, 2.0, 3.0], numSlices=2)
labels   = sc.parallelize([0,   1,   0],   numSlices=2)

# Важно: одинаковое число партиций И одинаковое число элементов!
dataset = features.zip(labels)
# [(1.0, 0), (2.0, 1), (3.0, 0)]

# Частый паттерн: добавить порядковый индекс к элементам
data_with_index = data_rdd.zipWithIndex()
# [("row_a", 0), ("row_b", 1), ("row_c", 2)]

`coalesce(n, shuffle=False)` - уменьшить число партиций без shuffle¶

Уменьшает число партиций до n путём слияния соседних партиций. По умолчанию shuffle=False - никакого сетевого обмена. Если shuffle=True - эквивалент repartition().

Почему Narrow (при shuffle=False): Spark объединяет соседние партиции на уровне метаданных, данные физически не перемещаются между узлами.

# После фильтрации данных стало меньше → объединить пустые партиции
filtered = big_rdd.filter(lambda x: x["country"] == "RU")
compacted = filtered.coalesce(10)  # без shuffle

# Перед записью: уменьшить число выходных файлов
result_rdd.coalesce(5).saveAsTextFile("hdfs:///output/")

# ЛОВУШКА: coalesce не увеличивает число партиций!
rdd.coalesce(400)    # rdd имеет 200 партиций → ничего не изменится
rdd.repartition(400)  # правильно - это уже Wide

DataFrame API: Narrow-трансформации¶

`select()` / `selectExpr()` - выбор и вычисление колонок¶

select() выбирает колонки или вычисляет новые через Column-выражения. selectExpr() принимает SQL-строки - удобно для сложных вычислений. Обе трансформации Narrow: каждая строка обрабатывается независимо. Catalyst применяет Column Pruning - если downstream-операции используют только часть колонок, Spark автоматически добавит select над источником и не будет читать лишние колонки из Parquet.

from pyspark.sql.functions import col, upper, year, round as spark_round

# Выбор существующих колонок
df.select("user_id", "amount", "event_date")

# Вычисление новых колонок
df.select(
    col("user_id"),
    upper(col("country")).alias("country_upper"),
    year(col("event_date")).alias("event_year"),
    (col("amount") * 1.2).alias("amount_with_vat"),
)

# selectExpr: SQL-синтаксис - удобнее для сложных формул
df.selectExpr(
    "user_id",
    "UPPER(country) AS country_upper",
    "YEAR(event_date) AS event_year",
    "CASE WHEN amount > 1000 THEN 'high' ELSE 'low' END AS tier",
    "amount * 1.2 AS amount_with_vat",
)

# Выбрать все колонки кроме одной
df.select([c for c in df.columns if c != "internal_id"])

# Переименовать несколько колонок через select
df.select(
    col("usr_id").alias("user_id"),
    col("amt").alias("amount"),
    col("evt_dt").alias("event_date"),
)

`drop(*cols)` - удаление колонок¶

Удаляет одну или несколько колонок из DataFrame. Narrow: просто убирает колонки из схемы, никаких вычислений по строкам нет. Безопасен - если колонки нет, ошибки не будет.

# Удалить чувствительные данные перед записью
clean_df = df.drop("ssn", "password_hash", "internal_id")

# Альтернатива: select всех кроме ненужных (семантически то же самое)
clean_df = df.select([c for c in df.columns if c not in {"ssn", "password_hash"}])

# drop безопасен: несуществующие колонки игнорируются
df.drop("non_existent_column")  # не упадёт, вернёт df как есть

`filter()` / `where()` - фильтрация строк¶

Абсолютно идентичны: where() это синоним filter() для SQL-совместимости. Оставляют только строки, удовлетворяющие условию. Narrow: каждая строка проверяется независимо. Catalyst применяет Predicate Pushdown - условие проталкивается к источнику (Parquet, JDBC), и ненужные строки не читаются вообще.

from pyspark.sql.functions import col, year

# Column expression (рекомендуется: тип-безопасно, IDE-autocomplete)
df.filter(col("amount") > 1000)
df.filter(col("country") == "RU")

# SQL string (удобно для сложных условий)
df.filter("amount > 1000 AND country = 'RU'")

# where() - синоним filter()
df.where(col("status").isin("active", "trial"))

# Комбинирование через &, |, ~
df.filter(
    (col("amount") > 0) &
    (col("country").isin("RU", "BY", "KZ")) &
    col("deleted_at").isNull()
)

# Фильтрация по дате
df.filter(year(col("event_date")) == 2024)

# Regexp-фильтр
df.filter(col("email").rlike(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"))

`withColumn()` / `withColumnRenamed()` / `withColumns()` - работа с отдельными колонками¶

withColumn(name, expr) добавляет новую или заменяет существующую колонку результатом выражения. Вычисление построчное - Narrow. Важно: каждый вызов withColumn() добавляет один узел в логический план. При массовом добавлении (50+ колонок) это приводит к гигантскому плану. Используйте select() или withColumns() (Spark 3.3+) для батчевого добавления.

from pyspark.sql.functions import col, lit, current_timestamp, sha2, when

# Добавить новую колонку
df.withColumn("amount_usd", col("amount") / col("exchange_rate"))

# Изменить тип существующей колонки
df.withColumn("amount", col("amount").cast("double"))

# Добавить константу через lit()
df.withColumn("source", lit("kafka_topic_events"))
df.withColumn("processed_at", current_timestamp())

# Условная логика через when/otherwise
df.withColumn("tier",
    when(col("amount") > 10000, "platinum")
    .when(col("amount") > 1000,  "gold")
    .when(col("amount") > 100,   "silver")
    .otherwise("bronze")
)

# Хэширование для анонимизации PII
df.withColumn("user_id_hash", sha2(col("user_id"), 256))

# withColumnRenamed: переименовать без пересчёта
df.withColumnRenamed("usr_id", "user_id").withColumnRenamed("amt", "amount")

# withColumns (Spark 3.3+): несколько колонок в одном вызове - эффективнее!
df.withColumns({
    "amount_usd": col("amount") / col("exchange_rate"),
    "event_year": year(col("event_date")),
    "tier":       when(col("amount") > 1000, "high").otherwise("low"),
})

`limit(n)` - ограничение числа строк¶

Возвращает DataFrame из не более чем n строк. Технически Narrow, но требует частичной координации: каждая партиция возвращает до n строк, Driver выбирает первые n.

# Показать первые 10 строк для проверки схемы
df.limit(10).show()

# Создать маленькую тестовую выборку
sample_df = big_df.limit(1000)

# Антипаттерн: limit перед groupBy не ускоряет агрегацию!
# Из-за lazy evaluation, groupBy выполнится по всем данным
df.limit(1000).groupBy("country").count()  # limit применится ПОСЛЕ groupBy

# Правильно: filter уменьшает данные до агрегации
df.filter(col("year") == 2024).groupBy("country").count()

`union()` / `unionAll()` / `unionByName()` - объединение DataFrame¶

Объединяет строки двух DataFrame. union() и unionAll() идентичны в Spark (оба не дедуплицируют). unionByName() сопоставляет колонки по именам, а не по позиции. Narrow: партиции второго DataFrame просто добавляются к списку партиций первого.

# union: объединить по позиции колонок
df_2023 = spark.read.parquet("s3a://data/2023/")
df_2024 = spark.read.parquet("s3a://data/2024/")
all_data = df_2023.union(df_2024)

# unionByName: безопасно при разном порядке колонок
df_a = spark.createDataFrame([(1, "alice")], ["id", "name"])
df_b = spark.createDataFrame([("bob", 2)],  ["name", "id"])
combined = df_a.unionByName(df_b)  # колонки сопоставляются по имени

# allowMissingColumns=True (Spark 3.1+): колонки без пары заполняются null
df_new = df_b.withColumn("email", lit("bob@test.com"))
combined = df_a.unionByName(df_new, allowMissingColumns=True)

# Объединение нескольких источников
from functools import reduce
from pyspark.sql import DataFrame
dfs = [spark.read.parquet(f"s3a://data/{y}/") for y in range(2020, 2025)]
all_years = reduce(DataFrame.union, dfs)

`sample(fraction, seed)` - случайная выборка строк¶

Возвращает случайный набор строк. Реализован через Bernoulli sampling - каждая строка включается с вероятностью fraction. Narrow: решение принимается локально для каждой строки.

# 10% выборка, воспроизводимая
sample = df.sample(fraction=0.1, seed=42)

# Train/test split без Wide-трансформаций
from pyspark.sql.functions import rand
df_split = df.withColumn("rand", rand(seed=42))
train = df_split.filter(col("rand") < 0.8).drop("rand")
test  = df_split.filter(col("rand") >= 0.8).drop("rand")
# Лучше, чем df.subtract(train) - subtract это Wide!

`na.fill()` / `na.drop()` / `na.replace()` - работа с null и NaN¶

Методы объекта DataFrameNaFunctions (доступен как df.na). Все Narrow: каждая строка обрабатывается независимо.

# na.fill: заполнить null значениями по умолчанию
df.na.fill(0)                          # все числовые колонки
df.na.fill("unknown")                  # все строковые колонки
df.na.fill({"amount": 0.0, "country": "XX", "status": "unknown"})

# na.drop: удалить строки с null
df.na.drop()                              # любой null → удалить строку
df.na.drop(how="all")                     # все поля null → удалить
df.na.drop(thresh=3)                      # менее 3 непустых полей → удалить
df.na.drop(subset=["user_id", "amount"])  # null именно в этих колонках → удалить

# na.replace: заменить конкретные значения
df.na.replace({"N/A": None, "NULL": None, "": None})
df.na.replace({-1: None}, subset=["amount"])  # -1 означает "нет данных"

# Более гибкая альтернатива через withColumn
from pyspark.sql.functions import coalesce, lit
df.withColumn("amount", coalesce(col("amount"), lit(0.0)))

`coalesce(n)` - уменьшить партиции без shuffle¶

Аналог RDD-метода - объединяет существующие партиции без сетевого обмена. Идеален для уменьшения числа выходных файлов перед записью.

# Перед записью: объединить 200 партиций в 10 файлов (без shuffle)
df.filter(col("year") == 2024) \
  .coalesce(10) \
  .write.parquet("s3a://output/2024/")

# ЛОВУШКА: не увеличивает партиции
df.coalesce(1000)    # у df 200 партиций → ничего не изменится
df.repartition(1000)  # правильно - это Wide

`sortWithinPartitions(*cols)` - локальная сортировка внутри партиций¶

Сортирует данные внутри каждой партиции локально, без глобальной координации. Это Narrow, в отличие от orderBy() (Wide). Используется для оптимизации записи Parquet - отсортированные данные лучше сжимаются и быстрее читаются при точечных запросах.

# Отсортировать внутри партиций для лучшего сжатия Parquet
df.repartition(50, "user_id") \
  .sortWithinPartitions("user_id", "event_date") \
  .write.parquet("s3a://output/events/")

# Разница с orderBy:
df.sortWithinPartitions("date")  # Narrow: локально в каждой партиции
df.orderBy("date")               # Wide: глобальная сортировка = shuffle

`mapInPandas(f, schema)` / `mapInArrow(f, schema)` - Pandas и Arrow UDF¶

Позволяют применить произвольную Python/Pandas функцию к партиции как к pd.DataFrame. Каждая партиция Spark конвертируется в Pandas DataFrame, обрабатывается функцией, результат конвертируется обратно. Narrow: каждая партиция обрабатывается изолированно.

import pandas as pd
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

schema = StructType([
    StructField("user_id", StringType()),
    StructField("amount_normalized", DoubleType()),
])

def normalize_partition(pdf: pd.DataFrame) -> pd.DataFrame:
    """Min-max нормализация внутри каждой партиции."""
    pdf = pdf.copy()
    min_val, max_val = pdf["amount"].min(), pdf["amount"].max()
    pdf["amount_normalized"] = (pdf["amount"] - min_val) / (max_val - min_val + 1e-9)
    return pdf[["user_id", "amount_normalized"]]

result = df.mapInPandas(normalize_partition, schema=schema)

# mapInArrow: та же идея, но через Apache Arrow батчи (быстрее для числовых данных)

Предупреждение: конвертация Spark ↔ Pandas добавляет overhead. Для простых трансформаций используйте нативные Spark-функции. mapInPandas оправдан для сложных алгоритмов с numpy/scipy.

`explode()` / `posexplode()` / `explode_outer()` - разворачивание массивов¶

Разворачивает колонку с массивом или картой в отдельные строки - по одной строке на элемент. Narrow: каждая строка обрабатывается независимо, только увеличивается их количество в партиции.

from pyspark.sql.functions import explode, posexplode, explode_outer

df = spark.createDataFrame([
    (1, ["spark", "python", "bigdata"]),
    (2, ["java", "kafka"]),
    (3, []),
], ["user_id", "tags"])

# explode: один элемент = одна строка; пустые массивы ИСЧЕЗАЮТ
df.select("user_id", explode("tags").alias("tag"))
# user_id=3 (пустой массив) пропадёт из результата

# explode_outer: сохраняет строки с пустыми/null массивами
df.select("user_id", explode_outer("tags").alias("tag"))
# user_id=3 останется с tag=null

# posexplode: добавляет позицию элемента в массиве
df.select("user_id", posexplode("tags").alias("pos", "tag"))
# +-------+---+-------+
# |user_id|pos|tag    |
# |      1|  0|spark  |
# |      1|  1|python |

# Разворачивание MapType
df_map = spark.createDataFrame([(1, {"a": 1, "b": 2})], ["id", "data"])
df_map.select("id", explode("data").alias("key", "value"))

Встроенные col()-выражения без агрегации¶

Сотни встроенных функций pyspark.sql.functions вычисляют результат для каждой строки независимо - все они Narrow.

from pyspark.sql.functions import (
    # Числовые
    abs, round, ceil, floor, sqrt, pow, log, greatest, least,
    # Строковые
    upper, lower, trim, lpad, rpad, length, substring,
    concat, concat_ws, split, regexp_extract, regexp_replace,
    # Даты
    to_date, to_timestamp, date_add, date_diff,
    year, month, dayofmonth, hour, minute, second,
    date_format, from_unixtime, unix_timestamp,
    # Null-обработка
    coalesce, when, isnull, isnan,
    # Массивы и сложные типы
    array, array_contains, array_distinct, array_size,
    struct, map_keys, map_values, from_json, to_json,
    # Хэши и кодирование
    hash, md5, sha1, sha2, crc32, base64, unbase64,
)

# Практический пример: обогащение события за один проход
clean_events = raw_df.withColumns({
    "event_ts":    to_timestamp(col("ts_str"), "yyyy-MM-dd HH:mm:ss"),
    "event_date":  date_format(col("event_ts"), "yyyy-MM-dd"),
    "country_iso": upper(col("country")),
    "full_name":   concat_ws(" ", col("first_name"), col("last_name")),
    "amount":      col("raw_amount").cast("double"),
})

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

Wide-трансформации неизбежно создают Shuffle - физическое перемещение данных между Executor-ами. Каждая Wide-трансформация создаёт новый Stage. Знание каталога позволяет, глядя на код, предсказывать количество сетевых перемещений.

RDD API: Wide-трансформации¶

`groupByKey()` - группировка всех значений по ключу (антипаттерн)¶

Группирует значения с одинаковым ключом в ResultIterable. Критически важно: groupByKey() отправляет по сети все значения для каждого ключа без какой-либо предварительной агрегации. При миллионе записей с одним ключом - миллион значений летят по сети.

Почему Wide: для формирования группы ключа A нужны все строки с A из всех партиций.

# АНТИПАТТЕРН: groupByKey + агрегация после shuffle
result = rdd.groupByKey().mapValues(sum)
# Все значения ключа летят по сети без предварительного reduce!

# Правильно для числовых агрегатов - используйте reduceByKey:
result = rdd.reduceByKey(lambda a, b: a + b)

# groupByKey оправдан ТОЛЬКО когда нужен полный список значений:
user_products = transactions_rdd.groupByKey().mapValues(
    lambda prods: sorted(set(prods))
)
# Здесь нет альтернативы - нужен полный список для дедупликации

`reduceByKey(f)` - агрегация с partial reduce на Map-стороне¶

В отличие от groupByKey, reduceByKey выполняет частичную агрегацию внутри каждой партиции ДО отправки данных по сети. По сети идут только частичные результаты - на порядки меньше данных.

Почему Wide: данные всё равно перераспределяются по ключу, но объём в сети минимален.

# Подсчёт слов - классика
word_counts = (
    words_rdd
    .map(lambda w: (w, 1))
    .reduceByKey(lambda a, b: a + b)
    # Внутри каждой партиции: {hello: 3, world: 2, ...} - map-side combine
    # По сети: только эти частичные суммы
)

# Сумма продаж по пользователю
sales = (
    transactions_rdd
    .map(lambda t: (t["user_id"], t["amount"]))
    .reduceByKey(lambda a, b: a + b)
)

# Поиск максимума по ключу
max_by_key = data_rdd.reduceByKey(lambda a, b: max(a, b))

Функция f должна быть коммутативной и ассоциативной - это необходимо для корректной частичной агрегации.

`aggregateByKey(zero)(seqOp, combOp)` - гибкая агрегация с аккумулятором¶

Самая гибкая агрегация. Принимает начальное значение аккумулятора (zero), функцию обновления аккумулятора в рамках партиции (seqOp) и функцию объединения аккумуляторов из разных партиций (combOp). Позволяет вычислить то, что reduceByKey не может - например, среднее в один проход.

# Вычислить среднее значение по ключу в один проход
result = sales_rdd.aggregateByKey(
    (0.0, 0),                                                    # zero: (sum, count)
    lambda acc, val: (acc[0] + val, acc[1] + 1),                 # seqOp: внутри партиции
    lambda a, b: (a[0] + b[0], a[1] + b[1])                     # combOp: между партициями
)
averages = result.mapValues(lambda sc: sc[0] / sc[1] if sc[1] > 0 else 0.0)

# Вычислить (min, max, sum, count) за один проход
stats = data_rdd.aggregateByKey(
    (float("inf"), float("-inf"), 0.0, 0),
    lambda acc, v: (min(acc[0], v), max(acc[1], v), acc[2] + v, acc[3] + 1),
    lambda a, b: (min(a[0], b[0]), max(a[1], b[1]), a[2] + b[2], a[3] + b[3])
)

`combineByKey(createCombiner, mergeValue, mergeCombiners)` - низкоуровневый комбайнер¶

Самый низкоуровневый и гибкий RDD-комбайнер. Фундамент, на котором построены reduceByKey, aggregateByKey и foldByKey. Три функции:

createCombiner: создать начальный аккумулятор из первого значения
mergeValue: добавить следующее значение к аккумулятору (в рамках партиции)
mergeCombiners: объединить два аккумулятора из разных партиций

# Вычислить (среднее, уникальные значения) по ключу одновременно
def create(v):
    return {"sum": v, "count": 1, "unique": {v}}

def merge_val(acc, v):
    return {"sum": acc["sum"] + v, "count": acc["count"] + 1,
            "unique": acc["unique"] | {v}}

def merge_acc(a, b):
    return {"sum": a["sum"] + b["sum"], "count": a["count"] + b["count"],
            "unique": a["unique"] | b["unique"]}

result = data_rdd.combineByKey(create, merge_val, merge_acc)

`foldByKey(zero, f)` - fold по ключу с начальным значением¶

Аналог reduceByKey, но с явным начальным значением zero. Работает как aggregateByKey(zero)(f, f) - одна функция для обеих фаз.

# Сумма с явным нулём (то же что reduceByKey для +)
result = data_rdd.foldByKey(0, lambda a, b: a + b)

# Конкатенация строк с разделителем
result = text_rdd.foldByKey("", lambda acc, s: acc + "|" + s if acc else s)

`cogroup(rdd1, rdd2, ...)` / `groupWith(rdd)` - совместная группировка нескольких RDD¶

Объединяет несколько (K, V) RDD по ключу: для каждого ключа возвращает кортеж итераторов значений из каждого RDD. Wide: все RDD перераспределяются по ключу через shuffle. groupWith() - синоним cogroup().

users   = sc.parallelize([("alice", "admin"), ("bob", "user")])
orders  = sc.parallelize([("alice", 100), ("alice", 200), ("bob", 50)])
reviews = sc.parallelize([("alice", "good"), ("bob", "ok")])

cogrouped = users.cogroup(orders, reviews)
for key, (roles, amounts, revs) in cogrouped.collect():
    print(f"{key}: roles={list(roles)}, amounts={list(amounts)}, reviews={list(revs)}")
# alice: roles=['admin'], amounts=[100, 200], reviews=['good']
# bob: roles=['user'], amounts=[50], reviews=['ok']

# cogroup - фундамент реализации всех join в RDD API

`join()` / `leftOuterJoin()` / `rightOuterJoin()` / `fullOuterJoin()` - join двух RDD¶

Все четыре - Wide: оба RDD перераспределяются по ключу через shuffle. Семантика идентична SQL JOIN.

users  = sc.parallelize([("alice", "admin"), ("bob", "user"), ("carol", "dev")])
orders = sc.parallelize([("alice", 100), ("bob", 200), ("dave", 50)])

# inner join - только совпадающие ключи
inner = users.join(orders).collect()
# [("alice", ("admin", 100)), ("bob", ("user", 200))]

# left outer join - все ключи из левого, None если нет в правом
left = users.leftOuterJoin(orders).collect()
# [("alice", ("admin", 100)), ("bob", ("user", 200)), ("carol", ("dev", None))]

# right outer join - все ключи из правого
right = users.rightOuterJoin(orders).collect()
# [("alice", ("admin", 100)), ("bob", ("user", 200)), ("dave", (None, 50))]

# full outer join - все ключи из обоих с None где нет пары
full = users.fullOuterJoin(orders).collect()

`sortByKey()` / `sortBy(f)` - глобальная сортировка¶

Wide: глобальная сортировка требует Range Partitioner - Spark делает sampling для определения границ диапазонов, затем перераспределяет данные по диапазонам, затем сортирует каждую партицию локально.

# Сортировка по ключу по убыванию
sorted_rdd = rdd.sortByKey(ascending=False)

# Сортировка по произвольной функции
sorted_by_value = rdd.sortBy(lambda kv: kv[1], ascending=False)

# По нескольким полям с разным направлением
data_rdd.sortBy(lambda row: (row["country"], -row["amount"]))

`distinct()` - дедупликация элементов¶

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

unique_events = events_rdd.distinct()

# Указать число партиций результата
unique_events = events_rdd.distinct(numPartitions=100)

`intersection(rdd)` - пересечение и `subtract(rdd)` - разность¶

Оба Wide: требуют shuffle для сравнения элементов из разных источников.

rdd1 = sc.parallelize([1, 2, 3, 4, 5])
rdd2 = sc.parallelize([3, 4, 5, 6, 7])
common = rdd1.intersection(rdd2)  # [3, 4, 5]

# Пользователи без оплаты
all_users  = users_rdd.map(lambda u: u["user_id"])
paid_users = payments_rdd.map(lambda p: p["user_id"]).distinct()
unpaid = all_users.subtract(paid_users)

`repartition(n)` - полное перераспределение с shuffle¶

Принудительно перераспределяет данные по n партициям через Hash Shuffle. Все данные проходят через сеть. Используется для выравнивания нагрузки перед дорогими агрегациями.

# Выровнять данные перед тяжёлым groupBy
balanced = skewed_rdd.repartition(500)
result = balanced.groupByKey()

# Увеличить параллелизм (coalesce не поможет!)
small_rdd.repartition(200)

`cartesian(rdd)` - декартово произведение¶

Возвращает все возможные пары элементов из двух RDD. M × N пар - никогда не используйте на больших данных.

rdd1 = sc.parallelize(["A", "B", "C"])  # 3 элемента
rdd2 = sc.parallelize([1, 2, 3, 4])     # 4 элемента
product = rdd1.cartesian(rdd2)          # 12 пар

# КАТАСТРОФА: 1M × 1M = 10^12 строк → Out of Memory
# Используйте только если декартово произведение действительно нужно
# (например, перебор комбинаций гиперпараметров на маленьких данных)

DataFrame API: Wide-трансформации¶

`groupBy(cols).agg(exprs)` - агрегация по ключу¶

Основная агрегационная операция. Catalyst автоматически добавляет HashAggregate (partial) перед Shuffle Exchange - это автоматический Map-Side Combine. По сети идут уже частично агрегированные данные. Это ключевое преимущество DataFrame API перед RDD groupByKey().

from pyspark.sql.functions import (
    sum, avg, count, countDistinct, min, max,
    stddev, variance, first, last,
    collect_list, collect_set,
    percentile_approx, approx_count_distinct,
)

# Базовая агрегация по одному ключу
df.groupBy("country").agg(
    sum("amount").alias("total_amount"),
    avg("amount").alias("avg_amount"),
    count("*").alias("event_count"),
    countDistinct("user_id").alias("unique_users"),
    min("amount").alias("min_amount"),
    max("amount").alias("max_amount"),
)

# Агрегация по нескольким ключам
df.groupBy("country", "product_category", year("event_date").alias("year")).agg(
    sum("amount").alias("total"),
    approx_count_distinct("user_id", rsd=0.05).alias("approx_users"),
)

# Статистика: percentile, stddev
df.groupBy("user_id").agg(
    percentile_approx("amount", [0.5, 0.9, 0.99]).alias("percentiles"),
    stddev("amount").alias("amount_std"),
)

# collect_list / collect_set - собрать все значения в список/множество
df.groupBy("user_id").agg(
    collect_set("product_id").alias("purchased_products"),
    collect_list("event_type").alias("event_history"),
)
# ОСТОРОЖНО: если по одному ключу миллионы строк → OOM на Executor!
# collect_set/collect_list только для умеренных объёмов данных на ключ

`groupBy(*cols).pivot(col, values=None).agg(expr)` - сводная таблица (PIVOT)¶

Поворачивает значения одной колонки в отдельные колонки. Wide и дорогая: два прохода агрегации (один для поиска уникальных значений, второй для вычисления агрегатов). Всегда задавайте values явно - это исключает первый проход.

# Продажи по категориям и странам: страны становятся колонками
# Без values: Spark сначала сканирует весь датасет чтобы найти уникальные страны
df.groupBy("category").pivot("country").sum("amount")

# С values: только один проход - в ~2× быстрее!
df.groupBy("category") \
  .pivot("country", ["RU", "US", "DE", "FR"]) \
  .sum("amount")
# +------------+------+------+------+------+
# |category    |    RU|    US|    DE|    FR|
# +------------+------+------+------+------+
# |electronics |50000 |80000 |30000 |20000 |

`rollup(cols).agg(expr)` / `cube(cols).agg(expr)` - иерархическая агрегация¶

rollup вычисляет агрегаты для иерархических подмножеств (слева направо). cube вычисляет агрегаты для всех возможных комбинаций группировочных колонок. Используются для OLAP-отчётности.

from pyspark.sql.functions import sum, grouping_id

# rollup: year → year+country → year+country+city (иерархия)
df.rollup("year", "country", "city").agg(sum("amount").alias("total")) \
  .orderBy("year", "country", "city")
# Результат включает строки каждого уровня иерархии:
# (2024, null, null)       - итог за год
# (2024, "RU", null)       - итог за год и страну
# (2024, "RU", "Moscow")   - итог за год, страну и город
# (null, null, null)       - общий итог

# cube: все комбинации year × country
df.cube("year", "country").agg(sum("amount"))

# grouping_id() - определить, по каким колонкам агрегировано в данной строке
df.rollup("year", "country").agg(
    sum("amount").alias("total"),
    grouping_id("year", "country").alias("grouping_level")
)

`orderBy(cols)` / `sort(cols)` - глобальная сортировка¶

Сортирует весь DataFrame глобально. Wide: сначала данные перераспределяются по диапазонам через Range Partitioner (sampling → границы → shuffle), затем каждая партиция сортируется локально. sort() и orderBy() идентичны.

from pyspark.sql.functions import col, desc, asc

# Базовая сортировка
df.orderBy("amount")                           # по возрастанию
df.orderBy(col("amount").desc())               # по убыванию

# Многоколоночная сортировка
df.orderBy(col("country").asc(), col("amount").desc())

# Null-обработка при сортировке
df.orderBy(col("amount").asc_nulls_last())     # null в конце
df.orderBy(col("amount").desc_nulls_first())   # null в начале

# АНТИПАТТЕРН: orderBy перед записью без limit создаёт один файл
# (все данные стекаются в одну партицию для глобального порядка)
# Правильно: repartition + sortWithinPartitions
df.repartition(50, "country") \
  .sortWithinPartitions("amount") \
  .write.parquet("s3a://output/")

`repartition(n)` / `repartition(n, *cols)` - перераспределение с shuffle¶

Перераспределяет данные по n партициям. Без cols - равномерное распределение (Round Robin в Spark 3.0+). С cols - хэш по указанным колонкам: все строки с одинаковым значением cols попадут в одну партицию.

# Равномерное перераспределение
df.repartition(200)

# По колонке: все строки с одинаковым user_id в одной партиции
df.repartition(200, "user_id")

# По нескольким колонкам
df.repartition(500, "country", "product_category")

# Паттерн: repartition перед join по тому же ключу
events_by_user   = events.repartition(200, "user_id")
profiles_by_user = profiles.repartition(200, "user_id")
# Если оба партиционированы одинаково, Spark может пропустить shuffle!
result = events_by_user.join(profiles_by_user, "user_id")

`distinct()` / `dropDuplicates(*cols)` - дедупликация строк¶

distinct() удаляет строки-дубликаты по всем колонкам. dropDuplicates(subset) позволяет указать только те колонки, по которым определять дубликаты. Оба Wide.

# Удалить полные дубликаты
df.distinct()

# Дедупликация по подмножеству колонок (оставляет первую встретившуюся строку)
df.dropDuplicates(["user_id"])
df.dropDuplicates(["user_id", "event_date"])

# Детерминированная дедупликация: взять строку с максимальным timestamp
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

w = Window.partitionBy("user_id").orderBy(col("updated_at").desc())
deduped = df.withColumn("rn", row_number().over(w)) \
            .filter(col("rn") == 1) \
            .drop("rn")

Sort-Merge Join (SMJ) - стандартный join двух больших таблиц¶

Применяется по умолчанию когда обе стороны join слишком велики для broadcast. Spark выполняет два shuffle: обе таблицы перераспределяются по ключу, затем каждая пара партиций merge-сортируется. spark.sql.join.preferSortMergeJoin=true по умолчанию.

# Sort-Merge Join используется автоматически для больших таблиц
result = large_events.join(large_profiles, on="user_id", how="inner")

# Принудительно через хинт
result = large_events.hint("merge").join(large_profiles, on="user_id")

# explain() покажет SortMergeJoin в физическом плане:
result.explain(mode="simple")
# == Physical Plan ==
# SortMergeJoin [user_id], [user_id], Inner
# :- Sort [user_id ASC]
# :  +- Exchange hashpartitioning(user_id, 200)   ← Shuffle
# :     +- ... large_events
# +- Sort [user_id ASC]
#    +- Exchange hashpartitioning(user_id, 200)   ← Shuffle
#       +- ... large_profiles

Shuffle Hash Join (SHJ) - join через хэш-таблицу в памяти¶

Альтернатива SMJ: меньшая сторона join целиком загружается в хэш-таблицу в памяти Executor, большая сторона проходит потоком. Без сортировки - быстрее SMJ если меньшая сторона помещается в RAM.

# Принудительно через хинт
result = large_df.join(medium_df.hint("shuffle_hash"), "id")

# Когда SHJ предпочтительнее SMJ:
# - Меньшая сторона помещается в память Executor
# - Нет требования к сортированному выводу
# - Данные уже хорошо распределены (нет skew)

`crossJoin()` - декартово произведение¶

Возвращает все возможные пары строк. Результат M × N строк. Чрезвычайно дорого на больших данных.

# M × N строк!
result = df1.crossJoin(df2)

# Безопасное использование: один из DataFrame крошечный
months    = spark.createDataFrame([("2024-01",), ("2024-02",)], ["month"])
products  = spark.createDataFrame([("A",), ("B",)], ["product"])
calendar  = months.crossJoin(products)  # 2 × 2 = 4 строки - нормально

# 1M × 1M = 10^12 строк → Out of Memory
# Требует явного включения:
spark.conf.set("spark.sql.crossJoin.enabled", "true")

`intersect()` / `intersectAll()` и `except()` / `exceptAll()` - пересечение и разность¶

intersect() возвращает строки из обоих DataFrame с дедупликацией. intersectAll() сохраняет дубликаты. except() - строки из первого, которых нет во втором. Все Wide.

# Пользователи активные в обоих периодах
jan_users = jan_events.select("user_id").distinct()
feb_users = feb_events.select("user_id").distinct()
both = jan_users.intersect(feb_users)

# intersectAll сохраняет дубликаты
both_all = jan_users.intersectAll(feb_users)

# Пользователи только из января (ушедшие в феврале)
jan_only = jan_users.subtract(feb_users)  # subtract = except в RDD API

Оконные функции с `Window.partitionBy()` - вычисления по группам без свёртки¶

Оконные функции позволяют делать агрегации и ранжирования не сворачивая строки. partitionBy() в Window - это Wide: данные перераспределяются по ключу партиции через shuffle. Без partitionBy() - весь DataFrame летит в одну партицию (катастрофа).

from pyspark.sql.window import Window
from pyspark.sql.functions import (
    row_number, rank, dense_rank, ntile,
    lag, lead, sum, avg, min, max, count,
)

# Нарастающая сумма покупок по пользователю по времени
w_user = Window.partitionBy("user_id").orderBy("event_date")
df.withColumn("running_total", sum("amount").over(w_user))

# Номер покупки пользователя (1, 2, 3, ...)
df.withColumn("purchase_num", row_number().over(w_user))

# Предыдущее и следующее событие
df.withColumn("prev_amount", lag("amount", 1).over(w_user)) \
  .withColumn("next_amount", lead("amount", 1).over(w_user))

# Ранк внутри группы
w_country = Window.partitionBy("country").orderBy(col("amount").desc())
df.withColumn("rank_in_country", rank().over(w_country))
df.withColumn("dense_rank",      dense_rank().over(w_country))

# Скользящее окно: сумма за последние 3 события
w_rolling = (
    Window.partitionBy("user_id")
          .orderBy("event_date")
          .rowsBetween(-2, 0)
)
df.withColumn("rolling_3_sum", sum("amount").over(w_rolling))

# КРИТИЧНО: Window без partitionBy - весь DataFrame в одну партицию!
w_bad = Window.orderBy("amount")
df.withColumn("global_rank", rank().over(w_bad))
# Все данные летят на один Executor → OOM или многочасовое ожидание

4. Механика Shuffle изнутри: три фазы¶

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

Sort-Based Shuffle: архитектура по умолчанию в Spark 3.x¶

Spark использует Sort-Based Shuffle с версии 1.6. До этого существовал Hash Shuffle, у которого была критическая проблема: M mappers × N reducers файлов - миллионы мелких файлов при большом параллелизме. Sort-Based Shuffle создаёт ровно 2 файла на каждый Map Task: один файл данных (.data) и один индексный (.index) - независимо от числа reduce-партиций.

Фаза 1: Shuffle Write (сторона Map)¶

Каждый Map Task после завершения вычислений:

Вычисляет partition_id для каждой строки: partition_id = hash(key) % numReducers. Для DataFrame Spark использует MurmurHash3 колонок ключа.
Частичная агрегация (если есть): для groupBy().agg() Spark применяет HashAggregate (partial) - строки с одинаковым ключом внутри партиции предварительно агрегируются в памяти. Это Map-Side Combine: данные сжимаются ДО записи на диск.
Сортировка в RAM-буфере: строки сортируются по partition_id (а внутри - по ключу, если нужен сортированный вывод). Если число reducers меньше spark.shuffle.sort.bypassMergeThreshold (по умолчанию 200) - сортировка пропускается (bypass mode).
Запись на локальный диск Executor-а: один бинарный файл .data со всеми bucket-ами последовательно, и файл .index с байтовыми смещениями каждого bucket. Запись идёт на эфемерный диск узла - не на HDFS и не на S3.

spark.conf.set("spark.shuffle.file.buffer", "1m")           # буфер записи (по умолчанию 32k)
spark.conf.set("spark.shuffle.sort.bypassMergeThreshold", "200")  # отключить sort при малом N

Фаза 2: Сетевая передача (Shuffle Transfer)¶

После завершения всех Map Tasks (Shuffle Read не начинается раньше!) каждый Reduce Task:

Запрашивает у Driver адреса всех Map Task-ов, выполнивших shuffle write.
Отправляет HTTP-запросы к External Shuffle Service (если включён) или напрямую к Executor-ам, скачивая только свой block из каждого .data файла, используя смещения из .index. Данные передаются по TCP/HTTP - реальный сетевой трафик, пропорциональный объёму данных.

External Shuffle Service - отдельный демон на каждом узле (spark.shuffle.service.enabled = true). Он хранит shuffle-файлы и отдаёт их даже если Executor уже завершился. Критически важен для динамического масштабирования.

Фаза 3: Shuffle Read (сторона Reduce)¶

Каждый Reduce Task получает данные со всех Map Task-ов и:

Десериализует полученные блоки из бинарного формата в JVM-объекты.
Merge Sort: если операция требует сортированного вывода (Sort-Merge Join, orderBy), полученные отсортированные блоки сливаются через внешнюю сортировку.
Финальная агрегация: для groupBy().agg() - HashAggregate (final) применяет финальный агрегат к частичным результатам.

Spill: когда RAM не хватает¶

Если объём данных shuffle-буфера Reduce Task превышает доступную heap, Spark начинает Spill - сброс данных на диск.

Spill вдвое увеличивает disk I/O: сначала запись промежуточных данных, потом их чтение при merge. Метрики в Spark UI:

Spill (Memory): объём данных до сжатия, вытесненных из RAM
Spill (Disk): реальный объём записи на диск

Если видите Spill - увеличьте spark.executor.memory или уменьшите spark.sql.shuffle.partitions (меньше данных на партицию → меньше Spill).

Ключевые настройки Shuffle¶

# Число партиций после shuffle (по умолчанию 200)
# Правило: 128-256 MB данных на одну партицию
spark.conf.set("spark.sql.shuffle.partitions", "1000")

# AQE: объединяет мелкие партиции после shuffle автоматически
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")

# AQE: автоматически переключает SMJ → Broadcast Join
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "30m")

# AQE: обработка skew - автоматически делит большие partitioned блоки
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")

5. reduceByKey vs groupByKey: анатомия антипаттерна¶

Это один из самых важных вопросов производительности в Spark RDD API. В DataFrame API Catalyst решает это автоматически через HashAggregate (partial), но в RDD - только вы.

При 1 млрд строк, где каждый пользователь имеет 1000 транзакций:

groupByKey(): все 1000 значений на пользователя летят по сети → гигабайты трафика
reduceByKey(+): каждый Map Task шлёт одно число (локальную сумму) → килобайты

# АНТИПАТТЕРН: groupByKey + агрегация на Reduce-стороне
sales_by_user = (
    transactions_rdd
    .map(lambda t: (t["user_id"], t["amount"]))
    .groupByKey()              # все значения летят по сети без pre-reduce!
    .mapValues(sum)            # sum считается ПОСЛЕ shuffle
)

# ПРАВИЛЬНО: reduceByKey с map-side combine
sales_by_user = (
    transactions_rdd
    .map(lambda t: (t["user_id"], t["amount"]))
    .reduceByKey(lambda a, b: a + b)   # частичный reduce ДО shuffle
)

# ДЛЯ СЛОЖНЫХ АГРЕГАТОВ: aggregateByKey (среднее нельзя через reduceByKey)
avg_by_user = (
    transactions_rdd
    .map(lambda t: (t["user_id"], t["amount"]))
    .aggregateByKey(
        (0.0, 0),
        lambda acc, v: (acc[0] + v, acc[1] + 1),
        lambda a, b:  (a[0] + b[0], a[1] + b[1]),
    )
    .mapValues(lambda sc: sc[0] / sc[1])
)

# В DataFrame API: Catalyst делает это автоматически
df.groupBy("user_id").agg(sum("amount"))
# В плане видно: HashAggregate (partial) → Exchange → HashAggregate (final)

Используйте groupByKey() только когда реально нужны все значения - например, построить список продуктов. Для любой числовой агрегации - reduceByKey() или aggregateByKey().

6. Broadcast Join: ликвидация Shuffle через броадкаст¶

Стандартный Sort-Merge Join перераспределяет обе таблицы по сети. Если одна таблица маленькая (справочник, измерение), её можно разослать на каждый Executor и выполнить join локально - Shuffle для большой таблицы полностью исчезает.

from pyspark.sql.functions import broadcast

# Автоматически: Spark выбирает Broadcast если таблица < порога
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 50 * 1024 * 1024)  # 50 MB
result = large_transactions.join(small_countries, "country_id")

# Принудительно: игнорирует порог, всегда Broadcast
result = large_transactions.join(broadcast(small_countries), "country_id")

# Хинт (альтернативный синтаксис)
result = large_transactions.join(small_countries.hint("broadcast"), "country_id")

# Принудительно Sort-Merge Join (отключить broadcast):
result = large_df.join(medium_df.hint("merge"), "id")

# Проверить план:
result.explain(mode="simple")
# == Physical Plan ==
# BroadcastHashJoin [country_id], [country_id], Inner, BuildRight
# :- ... large_transactions   (не shuffled!)
# +- BroadcastExchange HashedRelationBroadcastMode
#    +- ... small_countries   (broadcast)

# AQE автоматически переключает SMJ → BroadcastHashJoin в runtime
# если одна из сторон оказалась меньше порога после shuffle
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "30m")

Применяйте Broadcast Join когда одна сторона join - справочная таблица: страны, категории, валюты, небольшой справочник пользователей. Ориентир: таблица меньше 100–500 MB (зависит от памяти Executor).

7. coalesce vs repartition: физическая битва стратегий¶

Обе функции изменяют число партиций, но механизм принципиально разный.

Параметр	`coalesce(n)`	`repartition(n)`
Механизм	Склейка соседних партиций	Полный Hash Shuffle
Сетевой трафик	Нет	Весь датасет через сеть
n < текущего	Работает	Работает (избыточно)
n > текущего	Не работает	Работает
Баланс данных	Может быть неравномерным	Всегда равномерный
Когда использовать	Перед записью файлов	Перед тяжёлым groupBy/join

# ПАТТЕРН 1: уменьшить число файлов перед записью - без shuffle
filtered_df = big_df.filter(col("year") == 2024)
filtered_df.coalesce(10).write.parquet("s3a://output/2024/")

# ПАТТЕРН 2: выровнять skewed данные перед агрегацией
skewed_df.repartition(500).groupBy("user_id").agg(sum("amount"))

# ПАТТЕРН 3: partition by key перед join - Spark может пропустить shuffle!
events.repartition(200, "user_id") \
      .join(profiles.repartition(200, "user_id"), "user_id")

# ЛОВУШКА: coalesce(1) снижает параллелизм до нуля
df.coalesce(1).write.parquet(...)  # один Task обрабатывает все данные!

# ЛОВУШКА: coalesce не увеличивает партиции
df_200.coalesce(500)    # ничего не изменится - всё равно 200 партиций
df_200.repartition(500)  # правильно

8. Лабораторная практика: оптимизация пайплайна транзакций¶

Бизнес-кейс: вычислить суммарный баланс пользователей за последние 90 дней и присоединить данные из справочника валют. Исходный код на 1 TB данных работает 40 минут.

Неоптимальный пайплайн¶

from pyspark.sql.functions import col, sum, datediff, current_date, year

# ОШИБКА 1: repartition ПЕРЕД filter - 1 TB данных уходит в сеть,
# а потом большинство из них отфильтровывается
transactions = spark.read.parquet("s3a://data/transactions/") \
    .repartition(500)          # дорого: shuffle 1 TB до фильтрации

# ОШИБКА 2: filter ПОСЛЕ repartition - должен быть ДО
recent = transactions.filter(
    datediff(current_date(), col("tx_date")) <= 90
)

# Загрузка справочника валют - 150 строк!
currencies = spark.read.parquet("s3a://data/currencies/")

# ОШИБКА 3: Sort-Merge Join без broadcast - обе таблицы shuffled по сети
with_currency = recent.join(currencies, "currency_code")

# ОШИБКА 4: groupBy без учёта распределения данных после join
result = with_currency.groupBy("user_id").agg(
    sum(col("amount_usd")).alias("total_usd")
)

result.write.parquet("s3a://output/user_balances/")

Этот код создаёт граф с тремя Shuffle:

В Spark UI виден Shuffle Write Size = 1 TB на первом Stage (repartition), Shuffle Read = 900 GB на Stage с join.

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

from pyspark.sql.functions import col, sum, datediff, current_date, broadcast

# ШАГ 1: filter СНАЧАЛА - уменьшаем данные до любых wide-операций
# Catalyst + Parquet Predicate Pushdown: часть фильтрации уйдёт прямо в чтение
transactions = spark.read.parquet("s3a://data/transactions/") \
    .filter(datediff(current_date(), col("tx_date")) <= 90)
# Данных стало ~10% от исходных = ~100 GB

# ШАГ 2: broadcast маленькой таблицы - Shuffle 2 ликвидирован полностью
currencies = spark.read.parquet("s3a://data/currencies/")  # 150 строк = ~10 KB
with_currency = transactions.join(broadcast(currencies), "currency_code")
# BroadcastHashJoin: нет shuffle для transactions!

# ШАГ 3: AQE сам выберет оптимальное число partition после groupBy
spark.conf.set("spark.sql.adaptive.enabled", "true")
result = with_currency.groupBy("user_id").agg(
    sum(col("amount_usd")).alias("total_usd")
)

# ШАГ 4: уменьшить число файлов перед записью без лишнего shuffle
result.coalesce(50).write.parquet("s3a://output/user_balances/")

Оптимизированный граф:

1 Shuffle вместо 3. Объём данных в shuffle упал с 1 TB до 100 GB. Результат: 40 минут → 6 минут.

Как читать Spark UI после оптимизации¶

В Spark UI → Stages после оптимизации:

Меньше Stage (2–3 вместо 4–5)
В Stage с broadcast join: нет узла Exchange на стороне большой таблицы в DAG Visualization
Shuffle Write Size пропорционален объёму после фильтрации, а не исходному
Spill (Disk) = 0 если данных стало меньше и они помещаются в RAM

Команда для проверки: result.explain("formatted") - смотрите на узлы Exchange (shuffle) и BroadcastExchange (broadcast). Чем меньше Exchange - тем лучше.

9. Best Practices и Домашнее задание¶

Золотые правила минимизации Shuffle¶

1. Filter раньше Shuffle. Уменьшить данные до wide-трансформации - самый дешёвый способ ускорить пайплайн. filter и select (Column Pruning) ставить сразу после источника.

2. Broadcast для справочников. Любая таблица до 100–500 MB в join - кандидат на broadcast(). Настройте порог: spark.sql.autoBroadcastJoinThreshold = 100m. AQE автоматически переключит SMJ → BHJ в runtime если после shuffle одна сторона оказалась мала.

3. coalesce для уменьшения файлов, repartition для балансировки. Перед записью - coalesce(N) без shuffle. Перед groupBy по skewed ключу - repartition(N) для равномерного распределения.

4. reduceByKey вместо groupByKey в RDD. Для числовых агрегатов всегда reduceByKey или aggregateByKey. groupByKey - только когда нужен полный список значений.

5. repartition по ключу join. При нескольких join по одному ключу - один раз repartition(N, "join_key") для обеих сторон. Spark может пропустить shuffle для последующих join.

6. AQE включить. По умолчанию в Spark 3.2+. AQE объединяет маленькие shuffle-партиции, переключает join-стратегии и делит skewed-партиции автоматически.

7. spark.sql.shuffle.partitions под объём данных. 1 TB данных → 4000–8000 партиций по 128–256 MB. Дефолт 200 подходит только для небольших датасетов.

8. Window без partitionBy - никогда. Всегда указывать Window.partitionBy(). Без него весь DataFrame стекается в одну партицию → OOM.

Домашнее задание¶

Дан следующий PySpark-код:

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

# Три источника данных
orders    = spark.read.parquet("s3a://data/orders/")     # 500 GB
customers = spark.read.parquet("s3a://data/customers/")  # 50 GB
products  = spark.read.parquet("s3a://data/products/")   # 5 MB (справочник!)

# Пайплайн
step1 = orders.repartition(1000)

step2 = step1.join(customers, "customer_id")

step3 = step2.join(products, "product_id")

step4 = step3.filter(year(col("order_date")) == 2024)

step5 = step4.groupBy("customer_id", "country").agg(
    sum("amount").alias("total"),
    count("*").alias("order_count"),
)

step6 = step5.orderBy(col("total").desc())

step6.write.parquet("s3a://output/summary/")

Задание 1 - Подсчитать Stage-ы. Сколько Stage создаст этот код? Перечислите, какая операция на каком шаге создаёт границу Stage (Shuffle Exchange).

Задание 2 - Найти лишний Shuffle. В коде есть минимум два Shuffle, которых можно избежать полностью. Укажите их и объясните, почему они лишние с учётом размеров таблиц и порядка операций.

Задание 3 - Переписать пайплайн. Оптимизировать код так, чтобы:

количество Shuffle уменьшилось минимум на два
данные фильтровались до первого Shuffle
products (5 MB) использовался через broadcast

Задание 4 - Доказать оптимизацию. Написать step6.explain(mode="formatted") для обоих вариантов и выделить разницу: какие узлы Exchange исчезли, какой BroadcastExchange появился.

Подсказка: обратите внимание на размеры таблиц, порядок filter относительно join и наличие repartition до filter.

В следующем уроке разберём DataFrame API подробно: типы данных, схемы, Column expressions и работу с вложенными структурами (struct, array, map).

1. Партиции - физическая основа распределённости¶

Что такое партиция физически¶

Узкие зависимости (Narrow Dependencies)¶

Широкие зависимости (Wide Dependencies)¶

Граница Stage: как Shuffle делит граф вычислений¶

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

RDD API: Narrow-трансформации¶

map(f) - поэлементное преобразование¶

flatMap(f) - поэлементное преобразование с разворачиванием¶

mapPartitions(f) - преобразование всей партиции целиком¶

mapPartitionsWithIndex(f) - партиция с её номером¶

filter(f) - фильтрация элементов по предикату¶

union(rdd) - объединение двух RDD¶

sample(withReplacement, fraction, seed) - случайная выборка¶

glom() - партиция в список¶

mapValues(f) и flatMapValues(f) - трансформация только значений¶

keys() / values() - извлечение ключей или значений¶

zip(rdd) - попарное объединение двух RDD¶

coalesce(n, shuffle=False) - уменьшить число партиций без shuffle¶

DataFrame API: Narrow-трансформации¶

select() / selectExpr() - выбор и вычисление колонок¶

drop(*cols) - удаление колонок¶

filter() / where() - фильтрация строк¶

withColumn() / withColumnRenamed() / withColumns() - работа с отдельными колонками¶

limit(n) - ограничение числа строк¶

union() / unionAll() / unionByName() - объединение DataFrame¶

sample(fraction, seed) - случайная выборка строк¶

na.fill() / na.drop() / na.replace() - работа с null и NaN¶

coalesce(n) - уменьшить партиции без shuffle¶

sortWithinPartitions(*cols) - локальная сортировка внутри партиций¶

mapInPandas(f, schema) / mapInArrow(f, schema) - Pandas и Arrow UDF¶

explode() / posexplode() / explode_outer() - разворачивание массивов¶