PIVOT: транспонирование строк в колонки и обратно

Превращаем длинные таблицы в широкие и обратно: pivot(), unpivot(), stack(), физический план, оптимизация по cardinality и паттерны для ML и BI.

core

Зачем вообще менять форму данных¶

В аналитике данные живут в двух принципиально разных форматах. Переход между ними - одна из самых частых операций при построении витрин, подготовке признаков для ML и передаче данных в BI-системы.

Длинный / узкий формат (Long Format)¶

Каждый факт - отдельная строка. Одна строка = одно измерение для одного объекта в один момент времени.

user_id | month | revenue
──────────────────────────
  101   |  Jan  |  1500
  101   |  Feb  |  2200
  101   |  Mar  |  1800
  102   |  Jan  |   900
  102   |  Feb  |  1200
  102   |  Mar  |  3100

Это транзакционный формат. Именно в нём данные живут в Bronze- и Silver-слоях Data Lake:

Легко добавлять новые периоды: просто append новых строк, не нужно менять схему
Эффективно хранится в Parquet: повторяющиеся user_id и month отлично сжимаются
Просто агрегировать: стандартный groupBy + sum без трансформаций схемы
Легко интегрировать с потоковыми системами: Kafka, Flink, Spark Streaming работают именно с событиями-строками

Широкий формат (Wide Format)¶

Каждый объект - одна строка, а измерения разложены по колонкам.

user_id | Jan_revenue | Feb_revenue | Mar_revenue
──────────────────────────────────────────────────
  101   |    1500     |    2200     |    1800
  102   |     900     |    1200     |    3100

Это аналитический / репортинговый формат. В нём данные живут в Gold-слое и BI-системах:

BI-дашборды: Excel, Tableau, Power BI ожидают одну строку на объект
ML-фичи: модель получает вектор признаков - один вектор на объект
OLAP-матрицы: кросс-таблицы "пользователь × месяц", "регион × категория"
Когортный анализ: каждая когорта - строка, а колонки - шаги воронки

Два направления: PIVOT и UNPIVOT¶

PIVOT - превращает уникальные значения одной колонки в отдельные колонки с агрегацией. В нашем примере значения month (Jan, Feb, Mar) становятся именами колонок, а значения revenue агрегируются (sum) для каждого user_id.

UNPIVOT - обратная операция: разворачивает колонки обратно в строки. Широкая таблица с Jan_revenue, Feb_revenue превращается в длинную с колонкой month и колонкой revenue.

PIVOT в PySpark DataFrame API¶

Синтаксическая цепочка¶

Pivot в PySpark строго последователен - три метода в определённом порядке:

df.groupBy("id").pivot("category").agg(...)

.groupBy("id") - определяет, что остаётся строками. Каждый уникальный id → одна строка в результате
.pivot("category") - определяет, что превращается в колонки. Каждое уникальное значение category → новая колонка
.agg(sum("amount")) - определяет значение ячейки. Агрегация по amount внутри пересечения id × category

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum as spark_sum, col

spark = SparkSession.builder.appName("pivot-demo").getOrCreate()

data = [
    (101, "Jan", 1500.0), (101, "Feb", 2200.0), (101, "Mar", 1800.0),
    (102, "Jan",  900.0), (102, "Feb", 1200.0), (102, "Mar", 3100.0),
    (103, "Jan", 2500.0), (103, "Mar", 4000.0),  # у 103 нет Feb
]

df = spark.createDataFrame(data, ["user_id", "month", "revenue"])

result = df.groupBy("user_id").pivot("month").sum("revenue")
result.show()

+-------+------+------+------+
|user_id|   Feb|   Jan|   Mar|
+-------+------+------+------+
|    101|2200.0|1500.0|1800.0|
|    102|1200.0| 900.0|3100.0|
|    103|  null|2500.0|4000.0|
+-------+------+------+------+

Обратите внимание на несколько важных деталей:

Порядок колонок алфавитный: Jan, Feb, Mar превратились в Feb, Jan, Mar. Spark сортирует значения pivot-колонки лексикографически
NULL для отсутствующих комбинаций: у пользователя 103 нет данных за Feb - ячейка остаётся null. Это нормальное поведение, таблица становится разреженной
Агрегация обязательна: нельзя сделать pivot без агрегации. Если в ячейке есть несколько значений (несколько транзакций за один месяц у одного пользователя), нужно решить как их объединить

Несколько функций агрегации¶

Можно агрегировать одновременно несколько метрик:

from pyspark.sql.functions import sum as spark_sum, count, avg

result_multi = df.groupBy("user_id").pivot("month", ["Jan", "Feb", "Mar"]).agg(
    spark_sum("revenue").alias("sum"),
    count("revenue").alias("cnt"),
)
result_multi.show()

+-------+-------+-------+-------+-------+-------+-------+
|user_id|Jan_sum|Jan_cnt|Feb_sum|Feb_cnt|Mar_sum|Mar_cnt|
+-------+-------+-------+-------+-------+-------+-------+
|    101| 1500.0|      1| 2200.0|      1| 1800.0|      1|
|    102|  900.0|      1| 1200.0|      1| 3100.0|      1|
|    103| 2500.0|      1|   null|      0| 4000.0|      1|
+-------+-------+-------+-------+-------+-------+-------+

При нескольких агрегациях имена колонок формируются как {pivot_value}_{alias}. Это важно учитывать: при 12 месяцах и 3 метриках получим 36 колонок - это уже граница разумного.

SUM, COUNT, MAX, MIN и другие паттерны¶

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

Функция	Типичное использование
`sum`	Суммарная выручка, объём продаж
`count`	Количество транзакций, событий
`avg`	Средний чек, средняя оценка
`max`	Пиковое значение за период
`min`	Минимальное значение, первое появление
`first` / `last`	Последнее состояние (нужна сортировка)

# Когортный анализ: сколько транзакций в каждом месяце
transactions_count = df.groupBy("user_id").pivot("month", ["Jan", "Feb", "Mar"]).count()

# Feature engineering: максимальная транзакция за месяц
max_transaction = df.groupBy("user_id").pivot("month", ["Jan", "Feb", "Mar"]).max("revenue")

# Активность: был ли пользователь активен (1/0)
from pyspark.sql.functions import lit
activity = df.groupBy("user_id").pivot("month", ["Jan", "Feb", "Mar"]).agg(
    (count("revenue") > 0).cast("integer").alias("active")
)

Критическая проблема: динамический поиск значений¶

Вот самая важная особенность pivot, о которой часто забывают в production.

Что происходит без явных значений¶

# ОПАСНЫЙ вариант - без явного списка значений
result_slow = df.groupBy("user_id").pivot("month").sum("revenue")

Spark не знает заранее, какие значения есть в колонке month. Перед тем как построить результат, ему нужно их собрать. Это приводит к двум проходам по данным:

Первый скрытый DISTINCT - это полноценный shuffle. Все данные перемещаются между узлами чтобы найти уникальные значения month. Затем второй shuffle при самой агрегации. Итого: два shuffle вместо одного.

Для таблицы с миллиардом строк разница между одним и двумя shuffle - это разница между 5 минутами и 10+ минутами выполнения.

Решение: явный список значений¶

# ПРАВИЛЬНЫЙ вариант - явный список значений
result_fast = df.groupBy("user_id").pivot("month", ["Jan", "Feb", "Mar"]).sum("revenue")

Второй аргумент pivot() - список ожидаемых значений. Spark больше не делает DISTINCT-проход. Только один shuffle на агрегацию.

Дополнительные преимущества:

Предсказуемая схема: колонки всегда в том порядке, в котором вы указали их в списке
Отсечение неизвестных значений: если в данных появится "Apr", он не попадёт в результат (это плюс для production - схема стабильна)
Возможность контроля порядка: ["Jan", "Feb", "Mar"] даст именно такой порядок колонок, а не алфавитный

Как получить список значений заранее¶

В реальных пайплайнах список значений может меняться. Правильный подход - вычислить его один раз и кэшировать:

# Получаем актуальный список значений
pivot_values = [
    row["month"]
    for row in df.select("month").distinct().orderBy("month").collect()
]
# pivot_values = ['Feb', 'Jan', 'Mar']

# Используем в pivot
result = df.groupBy("user_id").pivot("month", pivot_values).sum("revenue")

.collect() выполняет маленький job только на month-колонке (projecting down колонку Parquet). Это намного дешевле чем скрытый DISTINCT по всей таблице в процессе pivot.

В продакшене список значений часто берётся из конфигурации или отдельного справочника:

# Из конфигурации пайплайна
MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun",
          "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]

# Из справочной таблицы
categories = spark.read.table("gold.dim_categories") \
    .select("category_code") \
    .rdd.flatMap(lambda x: x) \
    .collect()

PIVOT в Spark SQL¶

Базовый синтаксис¶

В Spark SQL PIVOT - это постфиксная конструкция после FROM:

SELECT *
FROM (
    SELECT user_id, month, revenue
    FROM transactions
) AS src
PIVOT (
    SUM(revenue)
    FOR month IN ('Jan', 'Feb', 'Mar')
)

Структура конструкции:

PIVOT ( ... ) - блок пивота, всегда следует за источником данных
SUM(revenue) - агрегирующая функция. Любая стандартная: SUM, COUNT, AVG, MAX, MIN
FOR month - колонка, значения которой станут именами новых колонок
IN ('Jan', 'Feb', 'Mar') - явный список значений (обязательный в SQL-синтаксисе)

В SQL значения в IN берутся в одинарные кавычки и становятся именами колонок. Если значение содержит пробелы или дефисы, нужно использовать псевдонимы:

SELECT *
FROM (SELECT user_id, category, amount FROM orders) AS src
PIVOT (
    SUM(amount)
    FOR category IN (
        'Electronics'    AS electronics,
        'Home & Garden'  AS home_garden,
        'Sports & Toys'  AS sports_toys
    )
)

Несколько агрегаций в SQL¶

SELECT *
FROM (SELECT user_id, month, revenue FROM transactions) AS src
PIVOT (
    SUM(revenue) AS sum_rev,
    COUNT(revenue) AS cnt_rev,
    AVG(revenue) AS avg_rev
    FOR month IN ('Jan' AS jan, 'Feb' AS feb, 'Mar' AS mar)
)

Колонки будут названы jan_sum_rev, jan_cnt_rev, jan_avg_rev, feb_sum_rev и т.д.

Динамический PIVOT через генерацию SQL¶

Чистый SQL не поддерживает динамический список значений - конструкцию IN (SELECT DISTINCT month FROM ...) в блоке PIVOT использовать нельзя. Обход: генерация SQL-строки в Python.

# Шаг 1: собрать уникальные значения
months = spark.sql("SELECT DISTINCT month FROM transactions ORDER BY month") \
    .rdd.flatMap(lambda x: x).collect()

# Шаг 2: сгенерировать IN-список
in_clause = ", ".join(f"'{m}' AS {m.lower()}" for m in months)
# → "'Feb' AS feb, 'Jan' AS jan, 'Mar' AS mar"

# Шаг 3: выполнить сгенерированный запрос
query = f"""
    SELECT *
    FROM (SELECT user_id, month, revenue FROM transactions) AS src
    PIVOT (
        SUM(revenue)
        FOR month IN ({in_clause})
    )
"""

result = spark.sql(query)

Это стандартный паттерн для динамических пивотов в production-пайплайнах. Генерация строки запроса происходит на driver - нет дополнительного shuffle на Spark-стороне.

Под капотом: как Catalyst преобразует PIVOT¶

Logical → Physical plan¶

Посмотрим что происходит внутри Spark когда мы делаем pivot.

df.groupBy("user_id") \
  .pivot("month", ["Jan", "Feb", "Mar"]) \
  .sum("revenue") \
  .explain(mode="formatted")

Упрощённый вывод:

== Physical Plan ==
*(2) HashAggregate(
    keys=[user_id],
    functions=[
        sum(CASE WHEN (month = Jan) THEN revenue END),
        sum(CASE WHEN (month = Feb) THEN revenue END),
        sum(CASE WHEN (month = Mar) THEN revenue END)
    ]
)
+- Exchange hashpartitioning(user_id, 200)
   +- *(1) HashAggregate(
          keys=[user_id],
          functions=[
              partial_sum(CASE WHEN (month = Jan) THEN revenue END),
              partial_sum(CASE WHEN (month = Feb) THEN revenue END),
              partial_sum(CASE WHEN (month = Mar) THEN revenue END)
          ]
      )
      +- FileScan ...

Ключевое открытие: Spark превращает PIVOT в набор условных агрегаций. Каждый уникальный месяц превращается в отдельное выражение sum(CASE WHEN month = 'X' THEN revenue END). Никакого специального физического оператора "Pivot" не существует - это просто умная трансформация Catalyst.

Два уровня HashAggregate¶

Агрегация выполняется в два этапа - partial и final:

partial_sum: каждый executor считает частичные суммы в своих партициях локально, без shuffle. Это mapside aggregation - очень быстро
Exchange hashpartitioning(user_id, 200): один shuffle - перемешать данные по user_id чтобы все данные одного пользователя оказались на одном executor
sum (final): финальное суммирование частичных результатов

Вот почему pivot с явным списком значений делает только один shuffle: Spark знает список колонок заранее, может сгенерировать CASE WHEN для каждой из них, и один HashAggregate закрывает всю задачу.

Что происходит без явного списка¶

Без явных значений в физическом плане появляется дополнительный проход:

== Physical Plan ==
*(3) HashAggregate(...)  ← финальная агрегация
+- Exchange hashpartitioning(user_id, 200)
   +- *(2) HashAggregate(...)  ← partial агрегация
      +- FileScan ...
*(1) HashAggregate(...)  ← сбор DISTINCT значений
+- Exchange SinglePartition   ← ещё один shuffle, всё на один узел
   +- HashAggregate(...)  ← partial DISTINCT
      +- FileScan ...     ← ВТОРОЙ скан данных

Два скана данных, два shuffle. При чтении с S3 - два полных HTTP range-запроса по всем файлам.

Влияние cardinality на размер плана¶

Каждое уникальное значение pivot-колонки добавляет одно CASE WHEN выражение в агрегацию. При 3 месяцах - 3 выражения. При 365 днях - 365 выражений. Catalyst вынужден генерировать и оптимизировать гигантский план. Это влияет на:

Время планирования: Plan compilation может занять секунды при сотнях колонок
Размер сериализованного плана: план передаётся между driver и executors
Memory per task: каждый executor должен держать в памяти буферы для всех 365 агрегаций одновременно

Практическое правило: не более 50-100 уникальных значений в pivot-колонке. Если значений больше - пересмотрите архитектуру.

Опасность wide tables¶

Разреженные таблицы¶

Когда не все комбинации groupBy_key × pivot_value существуют в данных, результирующая таблица заполняется null. Это sparse table (разреженная таблица).

Пример: 10 000 пользователей × 52 недели × 20 категорий продуктов. Если каждый пользователь в среднем покупает в 3 категориях из 20 - 85% ячеек будут null.

В Parquet null хранится эффективно (через validity bitmask), но:

Читаемость: таблица с 1040 колонками тяжело воспринимается
Memory при обработке: Spark держит schema для всех колонок при чтении, даже если большинство - null
Join cost: join широкой таблицы с другими таблицами дорог по памяти

Предел кардинальности для production¶

Количество значений	Рекомендация
3–20	Нормально, используйте pivot
20–50	Допустимо, явно задавайте список
50–100	С осторожностью, мониторьте memory
100–365	Опасно, рассмотрите альтернативы
365+	Не делайте pivot, используйте GROUPING SETS

Pivot и Parquet: columnar эффективность¶

Wide tables плохо сочетаются с Parquet при чтении подмножества колонок. Допустим, аналитик хочет только Jan_revenue. Parquet сделает column pruning - прочитает только нужную колонку. Но если в таблице 200 колонок, Parquet-файл имеет 200 column chunks в каждом row group. Footer metadata (список колонок) становится большим, читается при каждом открытии файла.

Альтернатива для очень широких таблиц: хранить normalized (long format) и делать pivot в BI-слое или как view в Spark.

UNPIVOT: обратная операция¶

Зачем разворачивать обратно¶

Ситуации, когда нужно превратить широкую таблицу в длинную:

Получили wide-формат из внешней системы (Excel, CSV из BI), нужно загрузить в нормализованное хранилище
Построили pivot-таблицу для Gold-слоя, теперь нужно передать её в downstream-систему, которая ожидает long-формат
Нужно применить оконные функции по временному измерению, которое сейчас размазано по колонкам

unpivot() - нативный метод (Spark 3.4+)¶

Начиная со Spark 3.4 появился нативный метод DataFrame.unpivot():

wide_df = spark.createDataFrame([
    (101, 1500.0, 2200.0, 1800.0),
    (102,  900.0, 1200.0, 3100.0),
    (103, 2500.0,   None, 4000.0),
], ["user_id", "Jan", "Feb", "Mar"])

# unpivot: Jan/Feb/Mar → строки
long_df = wide_df.unpivot(
    ids=["user_id"],           # колонки, которые остаются строками
    values=["Jan", "Feb", "Mar"],  # колонки, которые разворачиваются
    variableColumnName="month",    # имя новой колонки с именами бывших колонок
    valueColumnName="revenue",     # имя новой колонки со значениями
)

long_df.orderBy("user_id", "month").show()

+-------+-----+-------+
|user_id|month|revenue|
+-------+-----+-------+
|    101|  Feb| 2200.0|
|    101|  Jan| 1500.0|
|    101|  Mar| 1800.0|
|    102|  Feb| 1200.0|
|    102|  Jan|  900.0|
|    102|  Mar| 3100.0|
|    103|  Jan| 2500.0|
|    103|  Mar| 4000.0|
+-------+-----+-------+

По умолчанию unpivot удаляет строки с null-значениями. Пользователь 103 без данных за Feb не появился. Это поведение можно изменить через алиас melt:

# melt - синоним unpivot, с параметром для сохранения null
long_with_nulls = wide_df.melt(
    ids=["user_id"],
    values=["Jan", "Feb", "Mar"],
    variableColumnName="month",
    valueColumnName="revenue",
)
# melt сохраняет null-строки по умолчанию
long_with_nulls.show()

+-------+-----+-------+
|user_id|month|revenue|
+-------+-----+-------+
|    101|  Jan| 1500.0|
|    101|  Feb| 2200.0|
|    101|  Mar| 1800.0|
|    102|  Jan|  900.0|
|    102|  Feb| 1200.0|
|    102|  Mar| 3100.0|
|    103|  Jan| 2500.0|
|    103|  Feb|   null|   ← сохранён
|    103|  Mar| 4000.0|
+-------+-----+-------+

UNPIVOT в Spark SQL (Spark 3.4+)¶

SELECT user_id, month, revenue
FROM wide_table
UNPIVOT (
    revenue           -- имя новой колонки значений
    FOR month IN (    -- имя новой колонки ключей
        Jan,
        Feb,
        Mar
    )
)

С псевдонимами для более читаемых значений:

SELECT user_id, month, revenue
FROM wide_table
UNPIVOT (
    revenue FOR month IN (
        Jan AS 'January',
        Feb AS 'February',
        Mar AS 'March'
    )
)

stack() - классический подход (Spark < 3.4)¶

До появления нативного unpivot, стандартным способом был SQL-оператор stack(). Он работает во всех версиях Spark и по сей день встречается в старых кодовых базах.

stack(n, expr1, val1, expr2, val2, ...) - генерирует n строк из переданных пар (название, значение):

from pyspark.sql.functions import expr

long_df = wide_df.select(
    "user_id",
    expr("""
        stack(3,
            'Jan', Jan,
            'Feb', Feb,
            'Mar', Mar
        ) AS (month, revenue)
    """)
)

long_df.orderBy("user_id", "month").show()

+-------+-----+-------+
|user_id|month|revenue|
+-------+-----+-------+
|    101|  Feb| 2200.0|
|    101|  Jan| 1500.0|
|    101|  Mar| 1800.0|
|    102|  Feb| 1200.0|
| ...   |  ...|   ...  |

Разбор синтаксиса stack():

Первый аргумент 3 - количество строк на выходе (для каждой входной строки)
Далее пары 'имя_колонки', значение_колонки - чередуются через запятую
AS (month, revenue) - имена двух выходных колонок

stack() всегда сохраняет null-строки. Если нужно их убрать:

long_df.filter(col("revenue").isNotNull())

Генерация stack() для большого числа колонок¶

При 20+ колонках писать stack() вручную утомительно. Используйте генерацию:

# Предположим, wide_df имеет колонки user_id + 12 месяцев
value_cols = [c for c in wide_df.columns if c != "user_id"]

stack_args = ", ".join(
    f"'{col_name}', `{col_name}`" for col_name in value_cols
)
stack_expr = f"stack({len(value_cols)}, {stack_args}) AS (month, revenue)"

long_df = wide_df.select("user_id", expr(stack_expr))

Практика: Feature Engineering для ритейла¶

Соберём реалистичный пайплайн: из сырых транзакций покупателей строим feature matrix для ML-модели.

Исходные данные (Bronze)¶

from pyspark.sql.types import *
from pyspark.sql.functions import (
    sum as spark_sum, count, avg, col, round as spark_round
)

# Транзакции покупателей супермаркета
transactions_data = [
    # (customer_id, category, amount, week)
    (1, "food",        450.0, "W01"), (1, "food",        380.0, "W02"),
    (1, "electronics", 2999.0,"W01"), (1, "clothing",    1200.0,"W02"),
    (2, "food",        620.0, "W01"), (2, "food",        590.0, "W02"),
    (2, "food",        550.0, "W03"), (2, "electronics", 450.0, "W03"),
    (3, "clothing",    800.0, "W01"), (3, "clothing",    950.0, "W02"),
    (3, "clothing",    720.0, "W03"), (3, "food",        300.0, "W01"),
    (4, "electronics", 5500.0,"W01"), (4, "electronics", 3200.0,"W02"),
    (4, "food",        280.0, "W03"),
]

transactions = spark.createDataFrame(
    transactions_data,
    ["customer_id", "category", "amount", "week"]
)

transactions.show(8)

+-----------+-----------+------+----+
|customer_id|   category|amount|week|
+-----------+-----------+------+----+
|          1|       food| 450.0| W01|
|          1|       food| 380.0| W02|
|          1|electronics|2999.0| W01|
|          1|   clothing|1200.0| W02|
|          2|       food| 620.0| W01|
|          2|       food| 590.0| W02|
|          2|       food| 550.0| W03|
|          2|electronics| 450.0| W03|
+-----------+-----------+------+----+

Шаг 1: Сборка feature matrix через оптимизированный pivot¶

# Явный список категорий - берём из справочника или конфига
CATEGORIES = ["food", "electronics", "clothing"]

# Feature matrix: сумма покупок по каждой категории
feature_matrix = transactions \
    .groupBy("customer_id") \
    .pivot("category", CATEGORIES) \
    .agg(
        spark_round(spark_sum("amount"), 2).alias("sum"),
        count("amount").alias("cnt"),
    ) \
    .fillna(0)  # null → 0 для ML-модели

feature_matrix.show()

+-----------+--------+--------+---------------+---------------+------------+------------+
|customer_id|food_sum|food_cnt|electronics_sum|electronics_cnt|clothing_sum|clothing_cnt|
+-----------+--------+--------+---------------+---------------+------------+------------+
|          1|   830.0|       2|         2999.0|              1|      1200.0|           1|
|          2|  1760.0|       3|          450.0|              1|         0.0|           0|
|          3|   300.0|       1|            0.0|              0|      2470.0|           3|
|          4|   280.0|       1|         8700.0|              2|         0.0|           0|
+-----------+-----------+--------+---------------+---------------+------------+------------+

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

Шаг 2: Обогащение признаками-отношениями¶

from pyspark.sql.functions import when

# Доля food в общих тратах - признак "тип покупателя"
enriched = feature_matrix.withColumn(
    "total_spend",
    col("food_sum") + col("electronics_sum") + col("clothing_sum")
).withColumn(
    "food_share",
    spark_round(col("food_sum") / col("total_spend"), 3)
).withColumn(
    "electronics_share",
    spark_round(col("electronics_sum") / col("total_spend"), 3)
).withColumn(
    "customer_segment",
    when(col("electronics_sum") > 5000, "tech_buyer")
    .when(col("food_sum") / col("total_spend") > 0.7, "grocery_buyer")
    .otherwise("mixed_buyer")
)

enriched.select(
    "customer_id", "total_spend", "food_share", "electronics_share", "customer_segment"
).show()

+-----------+-----------+----------+-----------------+----------------+
|customer_id|total_spend|food_share|electronics_share|customer_segment|
+-----------+-----------+----------+-----------------+----------------+
|          1|    5029.0 |     0.165|            0.596|     mixed_buyer|
|          2|    2210.0 |     0.796|            0.204|   grocery_buyer|
|          3|    2770.0 |     0.108|            0.000|     mixed_buyer|
|          4|    8980.0 |     0.031|            0.969|      tech_buyer|
+-----------+-----------+----------+-----------------+----------------+

Шаг 3: UNPIVOT обратно в длинный формат¶

Допустим, нам нужно передать feature_matrix в downstream-систему, которая ожидает long-формат для каждой (customer, feature, value) тройки:

# Берём только сумма-фичи для разворачивания
sum_cols = ["food_sum", "electronics_sum", "clothing_sum"]

# Spark 3.4+: нативный unpivot
features_long = feature_matrix.select("customer_id", *sum_cols).unpivot(
    ids=["customer_id"],
    values=sum_cols,
    variableColumnName="feature_name",
    valueColumnName="feature_value",
)

# Для Spark < 3.4 - через stack:
# stack_expr = "stack(3, 'food_sum', food_sum, 'electronics_sum', electronics_sum, 'clothing_sum', clothing_sum) AS (feature_name, feature_value)"
# features_long = feature_matrix.select("customer_id", expr(stack_expr))

features_long.orderBy("customer_id", "feature_name").show()

+-----------+---------------+-------------+
|customer_id|   feature_name|feature_value|
+-----------+---------------+-------------+
|          1|   clothing_sum|       1200.0|
|          1|electronics_sum|       2999.0|
|          1|       food_sum|        830.0|
|          2|   clothing_sum|          0.0|
|          2|electronics_sum|        450.0|
|          2|       food_sum|       1760.0|
...
+-----------+---------------+-------------+

Шаг 4: Сравнение оптимизированного и неоптимизированного pivot¶

import time

# Создадим датасет побольше
from pyspark.sql.functions import randn, rand
import pyspark.sql.functions as F

large_df = spark.range(1_000_000).select(
    (col("id") % 10000).alias("customer_id"),
    F.element_at(F.array(F.lit("food"), F.lit("electronics"), F.lit("clothing")),
                 ((col("id") % 3) + 1).cast("int")).alias("category"),
    (rand() * 1000 + 100).alias("amount")
)
large_df.cache()
large_df.count()  # прогреть кэш

# Неоптимизированный вариант
t0 = time.time()
large_df.groupBy("customer_id").pivot("category").sum("amount").count()
t1 = time.time()
print(f"Без явных значений: {t1 - t0:.1f}s")

# Оптимизированный вариант
t2 = time.time()
large_df.groupBy("customer_id").pivot("category", ["food", "electronics", "clothing"]).sum("amount").count()
t3 = time.time()
print(f"С явными значениями: {t3 - t2:.1f}s")

На реальных данных разница между вариантами - от 20% до 2× в зависимости от объёма данных и числа уникальных значений.

Практика: Time-series pivot (KPI-матрица)¶

Классическая задача для BI-дашборда: выручка по регионам помесячно.

from pyspark.sql.functions import date_format, sum as spark_sum

# Допустим, у нас есть таблица продаж с датой
sales_data = [
    ("North", "2024-01-15", 150000.0),
    ("North", "2024-01-28", 230000.0),
    ("North", "2024-02-10", 180000.0),
    ("South", "2024-01-05",  95000.0),
    ("South", "2024-02-20", 210000.0),
    ("South", "2024-03-01", 175000.0),
    ("East",  "2024-01-10", 320000.0),
    ("East",  "2024-02-14", 290000.0),
    ("East",  "2024-03-22", 410000.0),
]

sales = spark.createDataFrame(sales_data, ["region", "sale_date", "amount"])

# Добавляем колонку month для pivot
from pyspark.sql.functions import to_date

sales = sales.withColumn("sale_date", to_date("sale_date")) \
             .withColumn("month", date_format("sale_date", "MMM"))

# Pivot: регион × месяц
MONTHS_ORDER = ["Jan", "Feb", "Mar"]

kpi_matrix = sales \
    .groupBy("region") \
    .pivot("month", MONTHS_ORDER) \
    .agg(spark_round(spark_sum("amount") / 1000, 1).alias("k")) \
    .fillna(0) \
    .orderBy("region")

kpi_matrix.show()

+------+-----+-----+-----+
|region|  Jan|  Feb|  Mar|
+------+-----+-----+-----+
|  East|320.0|290.0|410.0|
| North|380.0|180.0|  0.0|
| South| 95.0|210.0|175.0|
+------+-----+-----+-----+

Эта таблица готова для передачи в BI-систему или рендеринга в Jupyter как HTML-таблица с display().

PIVOT vs альтернативы¶

PIVOT vs GROUPING SETS¶

GROUPING SETS (и его сахар ROLLUP, CUBE) - альтернативный инструмент для многомерной аналитики. Когда что выбрать:

Критерий	PIVOT	GROUPING SETS
Форма результата	Wide (одна строка на объект)	Long (одна строка на комбинацию)
Передача в BI	Удобно (Excel, Tableau)	Требует дополнительного форматирования
Количество измерений	Обычно одно	Несколько одновременно
Кардинальность	Ограничена (≤50)	Не ограничена
Производительность	Один shuffle при явном списке	Один shuffle
Динамические значения	Сложно (генерация SQL)	Легко (просто добавить в groupBy)

# PIVOT: хорошо когда нужен wide-формат для экспорта
df.groupBy("region").pivot("month", MONTHS).sum("revenue")

# GROUPING SETS: хорошо когда нужны subtotals и разные уровни агрегации
spark.sql("""
    SELECT region, month, SUM(revenue)
    FROM sales
    GROUP BY GROUPING SETS (
        (region, month),  -- детальный уровень
        (region),         -- итоги по региону
        ()                -- общий итог
    )
""")

PIVOT vs реализация в BI-слое¶

Иногда правильный ответ - не делать pivot в Spark вообще. Современные BI-инструменты отлично справляются с транспонированием:

Tableau: Cross-tabulation из long-формата нативно
Power BI: Matrix visual принимает long-формат и сам строит матрицу
Metabase: Pivot table из любого запроса

Если конечный потребитель - BI-система, передавайте данные в long-формате. Это:

Проще хранить и обновлять (append-only)
Лучше для партиционирования по дате
Более гибко при изменении набора измерений

Делайте pivot в Spark только если downstream-система не умеет его строить сама, или если wide-формат нужен для ML/пакетного экспорта.

Spark UI и диагностика¶

Как pivot выглядит в Spark UI¶

В Spark UI вкладки Jobs и Stages при pivot-запросе покажут:

Stage 1: Scan + partial HashAggregate (локально на каждом executor, без shuffle)
Stage 2: Exchange (shuffle по groupBy-ключу) + final HashAggregate

Если вы видите три стадии вместо двух - это признак неоптимизированного pivot без явных значений. Первые две стадии - скрытый DISTINCT, третья - сам pivot.

Диагностика в Spark UI¶

# Добавьте имя запросу для поиска в UI
spark.sparkContext.setJobDescription("pivot-kpi-matrix")

result = df.groupBy("user_id").pivot("month", MONTHS).sum("revenue")
result.write.mode("overwrite").parquet("/tmp/kpi_output")

В UI ищите задачу по описанию "pivot-kpi-matrix". В деталях стадии обратите внимание на:

Shuffle Read Size: сколько данных передалось через shuffle. Для pivot это весь датасет (все строки, кроме отброшенных предикатами)
Task Duration Histogram: равномерность выполнения задач. Если одна задача значительно дольше - partition skew
Spill (Memory): если агрегация не помещается в память executor, данные spill'ятся на диск

Оптимизация через AQE¶

При spark.sql.adaptive.enabled=true (по умолчанию в Spark 3.x) AQE автоматически:

Коалесцирует shuffle-партиции после агрегации (уменьшает количество мелких файлов)
Обнаруживает skew в partition и разбивает слишком большие партиции

Для pivot это особенно важно: агрегация часто создаёт сильно неравномерные партиции (некоторые groupBy-ключи имеют много данных, другие - мало).

Anti-patterns¶

Anti-pattern 1: Pivot по high-cardinality колонке¶

# НЕПРАВИЛЬНО: pivot по user_id, где 1 000 000 уникальных значений
# Создаст 1 000 000 колонок - план не скомпилируется
df.groupBy("product_id").pivot("user_id").count()  # 💥 OOM или час компиляции

Если нужна матрица "продукт × пользователь" - используйте sparse представление (user_id + product_id как строки) или специализированные библиотеки.

Anti-pattern 2: Pivot без fillna для ML¶

# НЕПРАВИЛЬНО для ML: null в feature matrix ломает большинство алгоритмов
feature_matrix = df.groupBy("customer_id").pivot("category", CATEGORIES).sum("amount")
model.fit(feature_matrix)  # 💥 алгоритм не принимает null

# ПРАВИЛЬНО: всегда fillna после pivot для ML
feature_matrix = df.groupBy("customer_id") \
    .pivot("category", CATEGORIES) \
    .sum("amount") \
    .fillna(0)

Anti-pattern 3: Динамический pivot без контроля cardinality¶

# ОПАСНО: если categories вырастет до 500 значений - план взорвётся
categories = df.select("category").distinct().rdd.flatMap(lambda x: x).collect()
# Нет проверки на разумность!

# БЕЗОПАСНО: добавьте assert
assert len(categories) <= 50, f"Too many pivot values: {len(categories)}"
result = df.groupBy("user_id").pivot("category", categories).sum("amount")

Anti-pattern 4: Wide result без партиционирования¶

# НЕПРАВИЛЬНО: широкая таблица без партиционирования плохо читается
feature_matrix.write.parquet("s3a://datalake/gold/features/")

# ПРАВИЛЬНО: партиционируй по дате или сегменту, даже если таблица wide
feature_matrix.withColumn("snapshot_date", current_date()) \
    .write \
    .partitionBy("snapshot_date") \
    .parquet("s3a://datalake/gold/features/")

Домашнее задание: IoT-датасет¶

У вас есть IoT-данные с показаниями датчиков, записанными построчно:

# Структура данных
iot_data = [
    # (device_id, sensor_type, value, timestamp)
    ("DEV-001", "temperature", 22.5, "2024-01-15 10:00:00"),
    ("DEV-001", "humidity",    65.2, "2024-01-15 10:00:01"),
    ("DEV-001", "pressure",  1013.1, "2024-01-15 10:00:02"),
    ("DEV-001", "temperature", 22.8, "2024-01-15 11:00:00"),
    # ... миллионы строк за сутки
]

Задание:

Агрегировать данные по часам: для каждого (device_id, hour) вычислить avg(value) по каждому типу датчика
Сделать pivot по sensor_type, получив для каждого устройства и часа одну строку с колонками temperature, humidity, pressure
Добавить колонки-аномалии: флаг temp_anomaly = (temperature > 30 OR temperature < 10), humidity_anomaly = (humidity > 90 OR humidity < 20)
Отфильтровать часы с аномалиями и сохранить результат в Parquet с партиционированием по дате
Сравнить время выполнения с явными и без явных значений в pivot (замерить через time.time())
Открыть Spark UI → Jobs → выбрать задачу → сравнить количество стадий в оптимизированном и неоптимизированном вариантах. Приложить скриншот DAG-графа.

Чеклист¶

[ ] Понимаю разницу между Long Format и Wide Format - когда какой уместен
[ ] Знаю синтаксис .groupBy().pivot().agg() и порядок обязательных вызовов
[ ] Всегда передаю явный список значений в pivot() - никогда не оставляю без второго аргумента
[ ] Понимаю почему pivot без явных значений делает два прохода по данным (скрытый DISTINCT)
[ ] Умею читать EXPLAIN-план pivot: вижу CASE WHEN вместо специального оператора
[ ] Знаю SQL-синтаксис PIVOT: FROM ... PIVOT (agg FOR col IN (val1, val2))
[ ] Умею генерировать динамический SQL для pivot с переменным списком колонок
[ ] Знаю нативный unpivot() / melt() (Spark 3.4+) и stack() для совместимости
[ ] Понимаю ограничение по кардинальности: не более 50-100 уникальных значений в pivot-колонке
[ ] После pivot для ML делаю fillna(0) перед передачей в модель
[ ] Знаю когда GROUPING SETS лучше pivot (many dimensions, subtotals)
[ ] Знаю когда pivot лучше делать в BI-слое, а не в Spark