Пивот данных в PySpark

Пивот — это операция преобразования данных, при которой строки превращаются в колонки: «длинный» формат становится «широким». В PySpark это выполняется цепочкой .groupBy() + .pivot() + агрегация.

Синтаксис¶

pivot_df = df.groupBy("grouping_column") \
             .pivot("pivot_column", values=[...optional list...]) \
             .agg({"agg_column": "agg_function"})

• grouping_column: колонка(и) для группировки
• pivot_column: колонка, чьи уникальные значения станут новыми колонками
• values: необязательный список значений в pivot-колонке для ограничения вывода (иначе Spark определяет уникальные значения самостоятельно)
• agg_column / agg_function: колонка для агрегации и функция агрегации (sum, avg, count и др.)

Пример: базовый пивот¶

from pyspark.sql import SparkSession

spark = SparkSession.builder \
        .appName("PySpark Pivot Example") \
        .getOrCreate()

data = [
    ("A", "East", 100),
    ("A", "West", 150),
    ("B", "East", 200),
    ("B", "West", 250),
    ("C", "East", 300),
    ("C", "West", 350)
]
columns = ["Product", "Region", "Sales"]

df = spark.createDataFrame(data, columns)
df.show()

Вывод:

+-------+------+-----+
|Product|Region|Sales|
+-------+------+-----+
|      A|  East|  100|
|      A|  West|  150|
|      B|  East|  200|
|      B|  West|  250|
|      C|  East|  300|
|      C|  West|  350|
+-------+------+-----+

Теперь выполним пивот по Region, группируя по Product и агрегируя sum(Sales):

pivot_df = df.groupBy("Product") \
             .pivot("Region") \
             .sum("Sales")

pivot_df.show()

Ожидаемый результат:

+-------+----+----+
|Product|East|West|
+-------+----+----+
|      A| 100| 150|
|      B| 200| 250|
|      C| 300| 350|
+-------+----+----+

Каждый регион становится колонкой, а значения — суммой продаж.

Указание списка значений¶

Если в pivot-колонке много уникальных значений, но нужна только часть из них, используйте параметр values=:

pivot_df2 = df.groupBy("Product") \
              .pivot("Region", ["East"]) \
              .sum("Sales")

pivot_df2.show()

В результате будет только колонка "East" (без "West"). Удобно для контроля ширины выходного DataFrame.

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

from pyspark.sql.functions import sum, avg

agg_df = df.groupBy("Product") \
           .pivot("Region") \
           .agg(
             sum("Sales").alias("total_sales"),
             avg("Sales").alias("avg_sales")
           )

agg_df.show()

В результате получится по две колонки на каждый регион (например, East_total_sales, East_avg_sales, West_total_sales, West_avg_sales).

Применение и лучшие практики¶

Когда использовать пивот:

• Когда нужно создать перекрёстные таблицы (например, итоги по категориям в разрезе групп)
• Когда требуется широкий формат для отчётности или передачи в другие инструменты

На что обращать внимание:

• Если в pivot-колонке много уникальных значений, результирующий DataFrame станет очень широким — это может вызвать проблемы с памятью и производительностью
• Параметр values= помогает избежать избыточной ширины вывода
• NULL-значения: если для конкретной комбинации pivot-значения и группы данных нет, в этой колонке будет null — обрабатывайте через .fillna(0)
• Пивот вызывает перемешивание данных (shuffle); при работе с большими наборами данных планируйте партиционирование и ресурсы