Window Functions: frame specification, ROWS vs RANGE, UNBOUNDED

Зачем нужны оконные функции¶

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

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

Оконная функция вычисляет значение для каждой строки, «заглядывая» в соседние строки той же логической партиции. Число строк в результате равно числу строк во входных данных - никакого схлопывания.

Применение оконных функций шире, чем кажется:

• BI и отчётность: running total, moving average, cumulative share
• Feature engineering: lag-фичи для временных рядов в ML-пайплайнах
• Дедупликация: row_number() + фильтр для выбора последней версии записи
• Sessionization: группировка событий пользователя в сессии
• Ранжирование: top-N покупателей по регионам, best sellers по категориям

Анатомия WindowSpec¶

Любая оконная функция в PySpark работает через объект WindowSpec. Он описывает три вещи:

1. partitionBy - как разбить данные на независимые окна
2. orderBy - как отсортировать строки внутри каждого окна
3. frame - какие строки из окна включить в вычисление

from pyspark.sql import Window
from pyspark.sql.functions import row_number, sum, lag

# Полная спецификация окна
window = (
    Window
    .partitionBy("customer_id")          # одно окно на клиента
    .orderBy("order_date")               # хронологический порядок
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)  # от начала до текущей строки
)

# Применение
df = orders.withColumn("running_revenue", sum("amount").over(window))

Как Spark физически выполняет оконную функцию¶

Window functions фундаментально отличаются от groupBy по механике выполнения. При groupBy Spark может частично агрегировать данные прямо на узле-источнике (Partial Aggregation / map-side combine) и отправлять по сети уже уменьшенные итоги. При window functions этого сделать нельзя: Spark обязан собрать все строки одной партиции в одном месте, сохранив каждую строку нетронутой - ведь результат нужен именно на уровне строк, не агрегатов.

Выполнение window function проходит четыре этапа:

Этап 1: Exchange - shuffle без частичной агрегации¶

partitionBy("customer_id") запускает Exchange hashPartitioning. Spark вычисляет хэш ключа для каждой строки и переносит её на соответствующий Executor. В отличие от groupBy, где mapper-узел сначала агрегирует локальные данные и отправляет только итоги, в window functions по сети летят все исходные строки.

Это означает: объём трафика при shuffle для window function равен полному объёму входных данных, а не размеру агрегата. На 100 GB входных данных shuffle передаст ~100 GB.

Риск Data Skew: если один ключ (например, user_id = 'anonymous') встречается в 90% строк, весь этот объём уйдёт на один Executor. Это вызывает straggler-задачу и потенциальный OOM. В отличие от groupBy, где skew «смягчается» partial aggregation, в window functions горячий ключ полностью ложится на один узел.

Этап 2: Sort - сортировка внутри партиции¶

После shuffle каждый Executor сортирует свои строки по колонке orderBy. Это обязательный шаг - без него frame specification (rowsBetween, rangeBetween) не имеет физического смысла: «2 строки назад» определяется только в отсортированном порядке.

В физическом плане вы увидите SortExec. Сортировка занимает O(N log N) времени и требует O(N) памяти внутри партиции. При нехватке памяти Spark делает spill на диск - производительность резко падает.

Важно: если orderBy не задан (только partitionBy), этапа Sort нет - данные используются в произвольном порядке. Это допустимо только для функций, нечувствительных к порядку (например, sum или count по всей партиции).

Этап 3: WindowExec - скользящий буфер в памяти¶

После сортировки WindowExec запускает итератор, который проходит по строкам линейно. Для каждой строки итератор:

1. Определяет границы frame: для rowsBetween(-2, 0) - две строки назад и текущая
2. Вычисляет функцию по строкам в этих границах
3. Записывает результат в выходную строку
4. Сдвигается на одну строку вперёд

Оптимизация буфера для агрегатов: для функций вроде sum или avg Spark не пересчитывает агрегат заново при каждом сдвиге. Вместо этого при сдвиге окна на одну строку вперёд он вычитает значение ушедшей строки и прибавляет значение вошедшей. Это делает скользящее суммирование O(1) на каждый шаг вместо O(frame_size).

Например, для rowsBetween(-2, 0):

• На строке 4 буфер: [200, 150, 300], sum = 650
• Вместо 200 + 150 + 300 Spark делает 450 - 100 + 300 = 650 (вычел строку 1, добавил строку 4)

Не все функции поддерживают это: rank, row_number, lead/lag не являются агрегатными - для них буфер работает иначе.

Этап 4: ROWS vs RANGE - разница в реализации¶

ROWS frame - физический счётчик строк. Итератор просто отсчитывает N позиций назад/вперёд в массиве. Это O(1) операция - никаких сравнений значений, только индексная арифметика. Максимально дешёвый режим.

RANGE frame - семантическое сравнение. Для каждой текущей строки итератор проверяет: «находится ли значение orderBy-колонки этой строки в диапазоне [current - N, current + M]?» Это требует сравнения значений при каждом шаге. При наличии дубликатов в orderBy-колонке в «текущую» позицию могут попасть несколько строк одновременно - итератор должен их все обработать.

Практическое следствие: для rangeBetween с временными данными в orderBy должна быть числовая колонка (unix_timestamp), а не строка-дата - иначе Spark не сможет выполнить арифметическое сравнение.

Чтение физического плана¶

df.withColumn("rn", row_number().over(
    Window.partitionBy("customer_id").orderBy("order_date")
)).explain()

# == Physical Plan ==
# AdaptiveSparkPlan isFinalPlan=false
# +- Window [row_number() windowspecdefinition(
#      customer_id, order_date ASC NULLS FIRST,
#      specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$()))],
#    [customer_id], [order_date ASC NULLS FIRST]
#    +- Sort [customer_id ASC NULLS FIRST, order_date ASC NULLS FIRST], false, 0
#       +- Exchange hashpartitioning(customer_id, 200), ENSURE_REQUIREMENTS
#          +- Scan parquet ...

Цепочка снизу вверх: Scan → Exchange → Sort → Window. Каждый оператор - отдельный этап выполнения. Exchange - граница Stage (shuffle boundary).

Project Tungsten: оптимизация памяти¶

Tungsten - движок управления памятью в Spark, который хранит данные в компактном бинарном формате (UnsafeRow) напрямую в off-heap памяти, минуя Java Heap и Garbage Collector.

Для window functions это означает:

• Буферы frame хранятся как массивы байтов фиксированного размера, а не как Java-объекты
• GC-паузы не прерывают вычисление скользящего окна
• Данные партиции (после shuffle + sort) лежат компактно в памяти Executor'а, что ускоряет линейный проход итератора

При spill на диск Tungsten сериализует те же бинарные буферы - это быстрее, чем Java-сериализация объектов. Именно поэтому Spark предпочитает HashAggregate с Tungsten-буфером SortAggregate с Java-объектами.

Следствие: оконные функции дорогие операции. Они требуют полного shuffle + sort без оптимизации partial aggregation. При неправильном partitionBy (мало уникальных ключей или вовсе без partitionBy) все данные уходят на один Executor и возникает OOM.

PartitionBy: логические независимые окна¶

partitionBy определяет, как датасет разбивается на логические окна. Внутри каждого окна функция вычисляется независимо.

# Окно на клиента
Window.partitionBy("customer_id")

# Окно на комбинацию региона и категории
Window.partitionBy("region", "category")

# Окно на год и месяц (вычисленные колонки)
Window.partitionBy(year("order_date"), month("order_date"))

Главная ошибка: Window без partitionBy - это одно окно на весь датасет. Spark переместит все данные на один Executor:

# ОПАСНО: одно окно = весь датасет на одном Executor = OOM
bad_window = Window.orderBy("order_date")

# ПРАВИЛЬНО: разбить на разумные partition
good_window = Window.partitionBy("customer_id").orderBy("order_date")

Если бизнес-задача требует ранжирования по всему датасету (например, global rank), сначала оцените объём данных и рассмотрите альтернативу через groupBy + join.

OrderBy внутри окна¶

orderBy задаёт порядок строк внутри каждой партиции окна. Он нужен для двух типов функций:

• Ranking functions (row_number, rank, dense_rank): без порядка ранг не определён
• Cumulative calculations (running sum, cumulative avg): порядок определяет, что считается «накопленным»

# По возрастанию даты (от ранних к поздним)
Window.partitionBy("customer_id").orderBy("order_date")

# По убыванию (от последних к ранним)
Window.partitionBy("customer_id").orderBy(col("order_date").desc())

# Составной порядок: сначала по дате, при равенстве - по сумме
Window.partitionBy("customer_id").orderBy("order_date", col("amount").desc())

Важно: orderBy в WindowSpec влияет на frame по умолчанию. Если orderBy задан и frame не указан явно, Spark использует RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW - это не то же самое, что ROWS BETWEEN .... Об этом подробнее в разделе про frame.

Frame Specification: точные границы вычисления¶

Frame - это подмножество строк внутри окна, которые участвуют в вычислении для конкретной текущей строки. Для разных строк frame разный.

Граничные значения¶

Spark предоставляет специальные константы для задания границ frame:

# Примеры задания границ
Window.rowsBetween(Window.unboundedPreceding, Window.currentRow)  # от начала до текущей
Window.rowsBetween(-2, 2)             # 2 строки назад и 2 вперёд
Window.rowsBetween(-3, 0)             # последние 3 строки включая текущую
Window.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)  # вся партиция

ROWS vs RANGE: ключевое отличие¶

Spark поддерживает два типа frame: rowsBetween и rangeBetween. Они ведут себя идентично при уникальных значениях в orderBy, но расходятся при дубликатах.

ROWS frame: физическое смещение¶

rowsBetween(N, M) - отступить ровно N строк назад и M строк вперёд в отсортированном порядке. Это физическое позиционирование: строго по счётчику строк.

# Скользящее среднее за последние 3 транзакции (включая текущую)
sliding_avg = Window \
    .partitionBy("customer_id") \
    .orderBy("order_date") \
    .rowsBetween(-2, Window.currentRow)  # 2 назад + текущая = 3 строки

df.withColumn("rolling_3tx_avg", avg("amount").over(sliding_avg))

При rowsBetween(-2, 0) для каждой строки берутся ровно 2 предыдущие строки в отсортированном порядке, независимо от значений дат.

RANGE frame: логическое смещение по значению¶

rangeBetween(N, M) - включить все строки, где значение orderBy-колонки попадает в диапазон [current_value + N, current_value + M]. Это семантическое смещение по значению, а не по позиции.

# Сумма за последние 7 дней (семантически)
from pyspark.sql.functions import unix_timestamp

df_with_ts = df.withColumn("ts", unix_timestamp("order_date") * 1000)  # в миллисекундах

seven_days_ms = 7 * 24 * 60 * 60 * 1000
range_window = Window \
    .partitionBy("customer_id") \
    .orderBy("ts") \
    .rangeBetween(-seven_days_ms, Window.currentRow)

df_with_ts.withColumn("revenue_last_7d", sum("amount").over(range_window))

Критическое отличие при дубликатах:

# Данные: customer=1, даты [2024-01-01, 2024-01-01, 2024-01-01, 2024-01-05]
# orderBy("date"), суммируем amount=[100, 200, 150, 300]

# ROWS: строго физические позиции
# Для строки 1 (2024-01-01, 100):  rowsBetween(0, 0) → только строка 1 → 100
# Для строки 2 (2024-01-01, 200):  rowsBetween(0, 0) → только строка 2 → 200
# Для строки 3 (2024-01-01, 150):  rowsBetween(0, 0) → только строка 3 → 150

# RANGE: все строки с тем же значением date попадают в CURRENT ROW
# Для строки 1 (2024-01-01, 100):  rangeBetween(0, 0) → строки 1,2,3 → 450!
# Для строки 2 (2024-01-01, 200):  rangeBetween(0, 0) → строки 1,2,3 → 450!
# Для строки 3 (2024-01-01, 150):  rangeBetween(0, 0) → строки 1,2,3 → 450!

При использовании RANGE без явного frame (это поведение по умолчанию при orderBy без frame!), все строки с одинаковым значением в orderBy объединяются в «текущий» диапазон. Это часто приводит к неожиданным результатам.

Практическое правило:

• Используйте rowsBetween для скользящего окна по числу строк (последние N транзакций)
• Используйте rangeBetween для семантического временного окна (последние N дней/часов)
• При rangeBetween в orderBy должна быть только одна числовая колонка

Frame по умолчанию: скрытая ловушка¶

Если frame не указан явно:

• Без orderBy: ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING (вся партиция)
• С orderBy: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW (накопленный итог с RANGE-семантикой)

Второй случай - источник неожиданных результатов при дубликатах в orderBy. Всегда задавайте frame явно, если поведение по умолчанию неочевидно.

Ranking Functions: ранжирование строк¶

row_number: уникальный номер для каждой строки¶

row_number() присваивает уникальный последовательный номер каждой строке в партиции. При дубликатах в orderBy порядок произвольный, но номера всегда уникальны.

from pyspark.sql.functions import row_number

# Нумерация заказов клиента по дате
row_w = Window.partitionBy("customer_id").orderBy("order_date")
df.withColumn("order_seq", row_number().over(row_w))

# Дедупликация: взять последний заказ для каждого продукта
dedup_w = Window.partitionBy("product_id").orderBy(col("updated_at").desc())
df.withColumn("rn", row_number().over(dedup_w)) \
  .filter(col("rn") == 1) \
  .drop("rn")

Классическое применение row_number - дедупликация с выбором «лучшей» версии записи (latest update, highest confidence score, etc). Это самый распространённый паттерн использования window functions в ETL.

rank и dense_rank: ранги с дубликатами¶

from pyspark.sql.functions import rank, dense_rank

rank_w = Window.partitionBy("region").orderBy(col("revenue").desc())

df.withColumn("rank",       rank().over(rank_w)) \
  .withColumn("dense_rank", dense_rank().over(rank_w))

# Данные: revenues = [1000, 1000, 800, 700]
# rank:        [1, 1, 3, 4]   - после двух 1-х следует 3-е место (пропуск)
# dense_rank:  [1, 1, 2, 3]   - после двух 1-х следует 2-е место (без пропуска)

Когда что выбрать:

• row_number - дедупликация, нумерация (нужна уникальность)
• rank - спортивный рейтинг (при ничьей следующий номер пропускается)
• dense_rank - ранжирование без пробелов (удобно для top-N фильтрации)

ntile: разбивка на квантили¶

ntile(N) делит партицию на N равных групп (как децили, квартили). Если строки не делятся ровно, первые группы получают на одну строку больше.

from pyspark.sql.functions import ntile

# Разбить клиентов на 4 квартиля по выручке
quartile_w = Window.partitionBy("region").orderBy("total_spent")
df.withColumn("quartile", ntile(4).over(quartile_w))

# 10 децилей по сумме чека внутри категории
df.withColumn("decile", ntile(10).over(
    Window.partitionBy("category").orderBy("amount")
))

percent_rank и cume_dist: относительное положение в партиции¶

percent_rank() и cume_dist() - две функции, которые выражают позицию строки как долю от размера партиции. Они полезны для нормализации, построения CDF и анализа распределения.

percent_rank() возвращает относительный ранг от 0.0 до 1.0 по формуле (rank - 1) / (total_rows - 1). Первая строка всегда получает 0.0, последняя - 1.0.

cume_dist() (cumulative distribution) возвращает долю строк в партиции, значение которых меньше или равно текущему. Диапазон - от 1/N до 1.0. Это дискретный аналог функции накопленного распределения (CDF).

from pyspark.sql.functions import percent_rank, cume_dist

pr_w = Window.partitionBy("region").orderBy("amount")

df = orders.withColumn("pct_rank",  percent_rank().over(pr_w)) \
           .withColumn("cum_dist",  cume_dist().over(pr_w))

# Данные в партиции: amounts = [100, 200, 200, 300, 500]
# percent_rank:  [0.0,  0.25,  0.25,  0.75,  1.0]
# cume_dist:     [0.2,   0.6,   0.6,   0.8,  1.0]
#
# Для amount=200 (строки 2 и 3 из 5):
#   percent_rank = (2-1)/(5-1) = 0.25  - ранг первой из двух одинаковых
#   cume_dist    = 3/5 = 0.6            - 3 строки из 5 имеют amount ≤ 200

Практическое применение:

# Найти «дорогие» заказы - те, что выше 80-го перцентиля в своей категории
expensive = orders \
    .withColumn("pct", percent_rank().over(
        Window.partitionBy("category").orderBy("amount")
    )) \
    .filter(col("pct") >= 0.8)

# Нормализовать score клиента в диапазон [0, 1] внутри сегмента
normalized = customers \
    .withColumn("score_pct", percent_rank().over(
        Window.partitionBy("segment").orderBy("ltv_score")
    ))

Top-N на группу через dense_rank + filter¶

Частая задача: «топ-3 покупателя по выручке в каждом регионе».

# Шаг 1: ранжировать
ranked = orders.groupBy("region", "customer_id") \
    .agg(F.sum("amount").alias("revenue")) \
    .withColumn(
        "rnk",
        dense_rank().over(Window.partitionBy("region").orderBy(col("revenue").desc()))
    )

# Шаг 2: отфильтровать топ-3
top3 = ranked.filter(col("rnk") <= 3)

Это эффективнее, чем LIMIT или Python-sort, потому что фильтрация после window function компилируется в оптимальный физический план.

Analytical Functions: доступ к соседним строкам¶

lag: обращение к предыдущим строкам¶

lag(col, n, default) возвращает значение колонки из строки, стоящей на n позиций раньше в отсортированном окне. При n=1 это предыдущая строка, при n=3 - третья позиция назад.

Третий аргумент - значение по умолчанию для строк, у которых нет предыдущей (первые n строк партиции).

from pyspark.sql.functions import lag

time_w = Window.partitionBy("customer_id").orderBy("order_date")

df = orders \
    .withColumn("prev_amount",    lag("amount", 1).over(time_w)) \
    .withColumn("prev_amount_2",  lag("amount", 2).over(time_w)) \
    .withColumn("prev_amount_3",  lag("amount", 3, 0.0).over(time_w))  # default=0

# Изменение суммы по сравнению с предыдущим заказом
df = df.withColumn(
    "amount_delta",
    col("amount") - col("prev_amount")
).withColumn(
    "days_since_prev",
    F.datediff(col("order_date"), lag("order_date", 1).over(time_w))
)

Множественные lag-фичи для ML: в feature engineering временных рядов часто нужны несколько лагов одновременно. Это позволяет модели видеть паттерн за несколько периодов назад:

# Генерация lag-фичей для forecasting модели
lag_w = Window.partitionBy("product_id").orderBy("week")

sales_features = weekly_sales.select(
    "product_id", "week", "sales",
    *[
        lag("sales", i).over(lag_w).alias(f"sales_lag_{i}w")
        for i in range(1, 8)   # lags 1..7 недель назад
    ]
)

# Результат: sales_lag_1w, sales_lag_2w, ..., sales_lag_7w
# Строки первых 7 недель каждого продукта будут иметь NULL в части колонок

Заполнить NULL в lag-фичах можно через fillna или coalesce:

# Заполнить пустые лаги нулём (допустимо для продаж)
sales_features = sales_features.fillna(0, subset=[f"sales_lag_{i}w" for i in range(1, 8)])

lead: обращение к следующим строкам¶

lead(col, n, default) - зеркало lag: возвращает значение из строки на n позиций вперёд. Последние n строк партиции получают default (по умолчанию NULL).

lead незаменим для forward-looking паттернов: когда нужно знать, что произошло после текущего события, не делая self-join.

from pyspark.sql.functions import lead

order_w = Window.partitionBy("customer_id").orderBy("order_date")

df = orders \
    .withColumn("next_order_date",   lead("order_date", 1).over(order_w)) \
    .withColumn("next_order_amount", lead("amount", 1, 0.0).over(order_w))

# Количество дней до следующего заказа
df = df.withColumn(
    "days_to_next_order",
    F.datediff(col("next_order_date"), col("order_date"))
)

# Флаг: этот заказ - последний у клиента (lead вернул NULL)
df = df.withColumn(
    "is_last_order",
    col("next_order_date").isNull()
)

Lookahead pattern: определить, произойдёт ли конверсионное событие в следующих N шагах воронки:

event_w = Window.partitionBy("user_id").orderBy("event_time")

# Проверить: был ли purchase в течение следующих 5 событий
events = events \
    .withColumn("next_1_type", lead("event_type", 1).over(event_w)) \
    .withColumn("next_2_type", lead("event_type", 2).over(event_w)) \
    .withColumn("next_3_type", lead("event_type", 3).over(event_w)) \
    .withColumn(
        "converts_soon",
        col("next_1_type").isin("purchase", "checkout") |
        col("next_2_type").isin("purchase", "checkout") |
        col("next_3_type").isin("purchase", "checkout")
    )

Сравнение lag и lead:

NULL в orderBy: управление позицией пустых значений¶

Когда колонка orderBy содержит NULL, Spark по умолчанию ставит их первыми при asc и последними при desc. Это влияет на lag/lead и ranking функции.

# По умолчанию: nulls first при asc
Window.partitionBy("cust").orderBy("order_date")             # NULL идут первыми

# Явное управление: nulls last при asc (обычно желаемое поведение)
Window.partitionBy("cust").orderBy(col("order_date").asc_nulls_last())

# Для desc: nulls first (NULL как «самые новые» - осмысленно в некоторых задачах)
Window.partitionBy("cust").orderBy(col("order_date").desc_nulls_first())

# Или nulls last при desc
Window.partitionBy("cust").orderBy(col("order_date").desc_nulls_last())

Это критично для row_number при дедупликации: если updated_at содержит NULL и вы делаете .orderBy(col("updated_at").desc()), строки с NULL уйдут в конец при desc_nulls_last или окажутся первыми при дефолтном поведении (desc → nulls last - в PySpark это противоположно SQL Server, но совпадает с PostgreSQL).

# Безопасная дедупликация при наличии NULL в sort-колонке
dedup_w = Window.partitionBy("record_id").orderBy(
    col("updated_at").desc_nulls_last()  # NULL-записи не получат rn=1
)

df.withColumn("rn", row_number().over(dedup_w)).filter(col("rn") == 1)

first и last: первое и последнее значение в партиции¶

from pyspark.sql.functions import first, last

full_w = Window.partitionBy("customer_id").orderBy("order_date") \
    .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)

df.withColumn("first_order_date", first("order_date").over(full_w)) \
  .withColumn("last_order_date",  last("order_date").over(full_w))

first и last при задании orderBy без полного frame возвращают первое/последнее значение в accumulated frame, а не в полной партиции. Всегда указывайте явный frame unboundedPreceding..unboundedFollowing если нужна первая/последняя строка всей партиции.

Aggregate Functions в окне¶

Все стандартные агрегатные функции (sum, avg, min, max, count) работают внутри over(). Их поведение зависит от frame.

import pyspark.sql.functions as F

# Без frame и без orderBy: сумма по всей партиции (одно значение на группу)
total_w = Window.partitionBy("customer_id")
df.withColumn("total_customer_revenue", F.sum("amount").over(total_w))

# С orderBy и без явного frame: накопленная сумма (running total)
cum_w = Window.partitionBy("customer_id").orderBy("order_date")
df.withColumn("running_revenue", F.sum("amount").over(cum_w))

# Явный ROWS frame: скользящая сумма за 3 транзакции
sliding_w = Window.partitionBy("customer_id").orderBy("order_date") \
    .rowsBetween(-2, Window.currentRow)
df.withColumn("rolling_3tx_sum", F.sum("amount").over(sliding_w))

# Явный RANGE frame: сумма за 30 дней
import datetime
thirty_days = 30 * 24 * 60 * 60  # в секундах

range_w = Window.partitionBy("customer_id") \
    .orderBy(col("order_date").cast("long")) \
    .rangeBetween(-thirty_days, Window.currentRow)
df.withColumn("revenue_last_30d", F.sum("amount").over(range_w))

Cumulative (Running) Total¶

Накопленный итог - классический кейс для window aggregate:

# Running total: для каждого заказа - сумма всех предыдущих заказов клиента + текущий
running_w = Window.partitionBy("customer_id").orderBy("order_date") \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)

df = orders.withColumn("cumulative_revenue", F.sum("amount").over(running_w)) \
           .withColumn("cumulative_orders",  F.count("*").over(running_w))

Почему rowsBetween вместо дефолтного RANGE: при orderBy("order_date") без явного frame Spark использует RANGE. Если несколько заказов имеют одну дату, они все входят в «текущую» позицию и накопленный итог для них одинаковый. С rowsBetween каждый заказ получает свой уникальный накопленный итог.

Доля от итога группы¶

# Процент выручки каждого заказа от общей выручки клиента
total_w  = Window.partitionBy("customer_id")
order_w  = Window.partitionBy("customer_id").orderBy("order_date")

df = orders \
    .withColumn("customer_total", F.sum("amount").over(total_w)) \
    .withColumn("order_share",    col("amount") / col("customer_total"))

Практика: Sessionization¶

Sessionization - разбивка потока событий пользователя на сессии. Сессия начинается, если пауза между событиями превышает порог (например, 30 минут).

from pyspark.sql.functions import lag, col, sum as spark_sum, unix_timestamp

SESSION_GAP_SECONDS = 30 * 60  # 30 минут

events = spark.read.parquet("/data/events/")

# Шаг 1: вычислить разрыв между событиями
event_w = Window.partitionBy("user_id").orderBy("event_time")

events_with_gap = events.withColumn(
    "prev_time",
    lag(unix_timestamp("event_time"), 1).over(event_w)
).withColumn(
    "gap_seconds",
    unix_timestamp("event_time") - col("prev_time")
).withColumn(
    "is_new_session",
    # Первое событие (prev_time=NULL) или разрыв > порога
    col("prev_time").isNull() | (col("gap_seconds") > SESSION_GAP_SECONDS)
)

# Шаг 2: пронумеровать сессии накопленной суммой флагов
events_with_gap.withColumn(
    "session_id",
    spark_sum(col("is_new_session").cast("int")).over(event_w)
)

# Шаг 3: вычислить метрики сессии
session_w = Window.partitionBy("user_id", "session_id")

sessions = events_with_gap.withColumn(
    "session_start",  F.first("event_time").over(session_w)
).withColumn(
    "session_events", F.count("*").over(session_w)
).withColumn(
    "session_duration_sec",
    F.last(unix_timestamp("event_time")).over(session_w) -
    F.first(unix_timestamp("event_time")).over(session_w)
)

Практика: Time-Series аналитика заказов¶

import pyspark.sql.functions as F
from pyspark.sql.functions import col, lag, lead, row_number, rank, sum as spark_sum
from pyspark.sql import Window

orders = spark.read.parquet("/data/silver/orders/").filter(col("status") == "COMPLETED")

# --- Window Specs ---
customer_time_w = Window.partitionBy("customer_id").orderBy("order_date")
customer_full_w = Window.partitionBy("customer_id")
region_rev_w    = Window.partitionBy("region").orderBy(col("amount").desc())

# --- Аналитика по клиенту ---
enriched = orders \
    .withColumn("order_seq",        row_number().over(customer_time_w)) \
    .withColumn("prev_order_date",  lag("order_date", 1).over(customer_time_w)) \
    .withColumn("days_since_prev",  F.datediff(col("order_date"), col("prev_order_date"))) \
    .withColumn("running_revenue",  spark_sum("amount").over(
        customer_time_w.rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )) \
    .withColumn("rolling_3tx_avg",  F.avg("amount").over(
        customer_time_w.rowsBetween(-2, Window.currentRow)
    )) \
    .withColumn("customer_total",   spark_sum("amount").over(customer_full_w)) \
    .withColumn("order_share",      col("amount") / col("customer_total")) \
    .withColumn("region_rank",      rank().over(region_rev_w))

# --- Дедупликация: один заказ на клиента в день ---
latest_per_day = orders \
    .withColumn("rn", row_number().over(
        Window.partitionBy("customer_id", "order_date")
              .orderBy(col("updated_at").desc())
    )) \
    .filter(col("rn") == 1) \
    .drop("rn")

Производительность и anti-patterns¶

Множественные window specs: минимизация shuffle¶

Каждый уникальный WindowSpec - потенциально отдельный shuffle. Если несколько функций используют одну спецификацию, Spark умеет объединить их в один WindowExec. Если спецификации разные - будет несколько shuffle.

# ПЛОХО: три разных WindowSpec → три shuffle
w1 = Window.partitionBy("customer_id").orderBy("order_date")
w2 = Window.partitionBy("customer_id").orderBy("amount")
w3 = Window.partitionBy("region").orderBy("order_date")

df = df \
    .withColumn("seq_by_date",   row_number().over(w1)) \
    .withColumn("rank_by_amount", rank().over(w2)) \
    .withColumn("region_rank",   row_number().over(w3))

# ЛУЧШЕ: если логика позволяет - объединить в один select
# Разные partitionBy/orderBy всё равно дадут разные shuffle,
# но хотя бы явно выражено намерение

# Минимизировать количество уникальных WindowSpec в одном pipeline

Skew в window функциях¶

Если partitionBy создаёт сильно неравные окна (например, один пользователь с миллионом событий), Executor с большой партицией станет узким местом.

Решения:

# 1. Добавить соль к горячему ключу (только если функция позволяет разбиение)
# 2. Предварительно агрегировать данные для горячих ключей
# 3. Отдельно обработать VIP-пользователей

vip_users = ["VIP_001", "VIP_002"]

# Обработать VIP отдельно с другой стратегией
regular = events.filter(~col("user_id").isin(vip_users))
vip     = events.filter(col("user_id").isin(vip_users))

result = regular.union(vip)  # merge после отдельной обработки

Explain план window функции¶

# Смотрим физический план
df.withColumn("rn", row_number().over(Window.partitionBy("customer_id").orderBy("order_date"))) \
  .explain("extended")

# В Physical Plan ищите:
# WindowExec - исполнитель оконной функции
# SortExec   - сортировка внутри партиции
# Exchange   - shuffle по ключу partitionBy

Anti-patterns window functions¶

Best practices production window pipeline¶

Группировать функции с одинаковым WindowSpec в одном select:

w = Window.partitionBy("customer_id").orderBy("order_date")

# Один проход - несколько функций
df = orders.select(
    "*",
    row_number().over(w).alias("order_seq"),
    lag("order_date", 1).over(w).alias("prev_order_date"),
    F.sum("amount").over(w.rowsBetween(Window.unboundedPreceding, Window.currentRow))
      .alias("running_revenue"),
    F.avg("amount").over(w.rowsBetween(-2, Window.currentRow))
      .alias("rolling_avg_3tx"),
)

Явно задавать frame для aggregate функций:

# Вместо неявного RANGE
Window.partitionBy("cust").orderBy("date")

# Явный ROWS
Window.partitionBy("cust").orderBy("date").rowsBetween(Window.unboundedPreceding, Window.currentRow)

Фильтровать до window function:

# Уменьшить объём данных до shuffle
orders.filter(col("status") == "COMPLETED") \
      .filter(col("order_date") >= "2024-01-01") \
      .withColumn("rn", row_number().over(w))

Контролировать shuffle.partitions для window:

# Window shuffle подчиняется тому же параметру, что и groupBy
spark.conf.set("spark.sql.shuffle.partitions", "100")

Когда использовать window, а когда groupBy¶

Задания для практики¶

Задание 1: Дедупликация¶

Дан DataFrame с историей изменений (record_id, version, updated_at, data). Оставьте только последнюю версию каждой записи (максимальный updated_at). Используйте row_number().

from pyspark.sql import Window
from pyspark.sql.functions import row_number, col

# Данные: несколько версий одной записи
data = [
    (1, 1, "2024-01-01", "v1"),
    (1, 2, "2024-01-05", "v2"),
    (1, 3, "2024-01-10", "v3"),   # ← самая свежая для record_id=1
    (2, 1, "2024-01-03", "va"),
    (2, 2, "2024-01-08", "vb"),   # ← самая свежая для record_id=2
]
df = spark.createDataFrame(data, ["record_id", "version", "updated_at", "data"])

# Окно: для каждого record_id, от новых к старым
dedup_w = Window.partitionBy("record_id").orderBy(col("updated_at").desc())

result = (
    df
    .withColumn("rn", row_number().over(dedup_w))
    .filter(col("rn") == 1)
    .drop("rn")
)

result.show()
# +---------+-------+----------+----+
# |record_id|version|updated_at|data|
# +---------+-------+----------+----+
# |        1|      3|2024-01-10|  v3|
# |        2|      2|2024-01-08|  vb|
# +---------+-------+----------+----+

Задание 2: Running total и rolling average¶

По датасету транзакций клиентов (customer_id, order_date, amount):

• Накопленную выручку клиента (running_revenue) через rowsBetween(unbounded, current)
• Скользящее среднее по последним 5 транзакциям через rowsBetween(-4, 0)
• Сравните результаты при использовании RANGE вместо ROWS при наличии дат-дубликатов

from pyspark.sql import Window
from pyspark.sql.functions import sum as spark_sum, avg, col

data = [
    (1, "2024-01-01", 100.0),
    (1, "2024-01-03", 200.0),
    (1, "2024-01-03", 150.0),   # дубликат даты!
    (1, "2024-01-07", 300.0),
    (1, "2024-01-10", 50.0),
    (1, "2024-01-12", 400.0),
]
df = spark.createDataFrame(data, ["customer_id", "order_date", "amount"])

base_w = Window.partitionBy("customer_id").orderBy("order_date")

# ROWS: физические позиции - каждая строка уникальна
rows_w = base_w.rowsBetween(Window.unboundedPreceding, Window.currentRow)

# RANGE: по значению даты - дубликаты объединяются в один "текущий" диапазон
range_w = base_w.rangeBetween(Window.unboundedPreceding, Window.currentRow)

result = df \
    .withColumn("running_rows",   spark_sum("amount").over(rows_w)) \
    .withColumn("running_range",  spark_sum("amount").over(range_w)) \
    .withColumn("rolling_5tx",    avg("amount").over(base_w.rowsBetween(-4, 0)))

result.show()
# customer_id | order_date | amount | running_rows | running_range | rolling_5tx
# 1           | 2024-01-01 |  100.0 |        100.0 |         100.0 |       100.0
# 1           | 2024-01-03 |  200.0 |        300.0 |         450.0 | ← RANGE: 100+200+150
# 1           | 2024-01-03 |  150.0 |        450.0 |         450.0 |       150.0
# 1           | 2024-01-07 |  300.0 |        750.0 |         750.0 |       190.0
# 1           | 2024-01-10 |   50.0 |        800.0 |         800.0 |       160.0
# 1           | 2024-01-12 |  400.0 |       1200.0 |        1200.0 |       220.0

Задание 3: Топ-3 по регионам¶

Найдите трёх лучших клиентов по суммарной выручке в каждом регионе. Используйте dense_rank(). Добавьте колонку regional_share - доля этого клиента в выручке региона.

from pyspark.sql import Window
from pyspark.sql.functions import dense_rank, sum as spark_sum, col
import pyspark.sql.functions as F

# Сначала агрегируем по клиенту и региону
customer_revenue = orders.groupBy("region", "customer_id").agg(
    spark_sum("amount").alias("revenue")
)

# Окно для ранжирования: внутри каждого региона, от большой выручки к малой
rank_w = Window.partitionBy("region").orderBy(col("revenue").desc())

# Окно для суммарной выручки региона (весь раздел)
total_w = Window.partitionBy("region")

result = (
    customer_revenue
    .withColumn("rnk",             dense_rank().over(rank_w))
    .withColumn("region_total",    spark_sum("revenue").over(total_w))
    .withColumn("regional_share",  F.round(col("revenue") / col("region_total"), 4))
    .filter(col("rnk") <= 3)
    .orderBy("region", "rnk")
    .select("region", "customer_id", "revenue", "rnk", "regional_share")
)

result.show()
# +------+-----------+--------+---+--------------+
# |region|customer_id| revenue|rnk|regional_share|
# +------+-----------+--------+---+--------------+
# |   EKB|       C_07|  8500.0|  1|        0.3214|
# |   EKB|       C_12|  6200.0|  2|        0.2344|
# |   EKB|       C_03|  5800.0|  3|        0.2193|
# |   MSK|       C_01| 45000.0|  1|        0.1823|
# ...

Задание 4: Sessionization¶

Реализуйте sessionization для потока событий (user_id, event_time, event_type). Сессия начинается заново, если пауза > 30 минут. Для каждой сессии вычислите: длительность, количество событий, первое и последнее событие.

from pyspark.sql import Window
from pyspark.sql.functions import (
    lag, col, unix_timestamp, sum as spark_sum,
    count, first, last, round as spark_round
)

SESSION_GAP = 30 * 60  # 30 минут в секундах

events = spark.createDataFrame([
    (1, "2024-01-10 10:00:00", "page_view"),
    (1, "2024-01-10 10:05:00", "click"),
    (1, "2024-01-10 10:08:00", "purchase"),
    (1, "2024-01-10 11:00:00", "page_view"),  # новая сессия: разрыв 52 мин
    (1, "2024-01-10 11:10:00", "click"),
    (2, "2024-01-10 09:00:00", "page_view"),
    (2, "2024-01-10 09:15:00", "click"),
], ["user_id", "event_time", "event_type"])

# Конвертируем время в unix-секунды для вычислений
events = events.withColumn("ts", unix_timestamp("event_time"))

# Шаг 1: найти разрыв между событиями через lag
user_w = Window.partitionBy("user_id").orderBy("ts")

events = events.withColumn("prev_ts", lag("ts", 1).over(user_w)) \
               .withColumn("gap_sec", col("ts") - col("prev_ts")) \
               .withColumn(
                   "is_new_session",
                   col("prev_ts").isNull() | (col("gap_sec") > SESSION_GAP)
               )

# Шаг 2: нумеровать сессии через cumulative sum флагов
events = events.withColumn(
    "session_id",
    spark_sum(col("is_new_session").cast("int")).over(user_w)
)

# Шаг 3: агрегировать метрики сессии
session_w = Window.partitionBy("user_id", "session_id")
full_session_w = session_w.rowsBetween(
    Window.unboundedPreceding, Window.unboundedFollowing
)

result = events.withColumn(
    "session_start",    first("event_time").over(full_session_w)
).withColumn(
    "session_end",      last("event_time").over(full_session_w)
).withColumn(
    "session_events",   count("*").over(session_w)
).withColumn(
    "session_dur_min",  spark_round(
        (last("ts").over(full_session_w) - first("ts").over(full_session_w)) / 60,
        1
    )
)

result.select(
    "user_id", "session_id", "event_time", "event_type",
    "session_start", "session_end", "session_events", "session_dur_min"
).show(truncate=False)

Задание 5: lead и lookahead¶

По датасету событий (user_id, event_time, event_type) с типами page_view, add_to_cart, checkout, purchase:

• Для каждого add_to_cart-события определите, было ли purchase в следующих 3 событиях пользователя
• Добавьте days_to_next_purchase - сколько дней прошло до ближайшей покупки (если есть)
• Выведите конверсию: доля add_to_cart, которые привели к purchase в следующих 3 шагах

from pyspark.sql import Window
from pyspark.sql.functions import lead, col, datediff, when, avg

event_w = Window.partitionBy("user_id").orderBy("event_time")

events = spark.createDataFrame([
    (1, "2024-01-01", "page_view"),
    (1, "2024-01-02", "add_to_cart"),
    (1, "2024-01-03", "checkout"),
    (1, "2024-01-04", "purchase"),     # через 2 шага после add_to_cart
    (2, "2024-01-01", "page_view"),
    (2, "2024-01-02", "add_to_cart"),
    (2, "2024-01-05", "page_view"),    # нет purchase → не конвертировал
    (3, "2024-01-01", "add_to_cart"),
    (3, "2024-01-02", "page_view"),
    (3, "2024-01-03", "page_view"),
    (3, "2024-01-04", "page_view"),    # purchase только на 5-м шаге → не в окне
    (3, "2024-01-05", "purchase"),
], ["user_id", "event_time", "event_type"])

# Смотрим вперёд на 3 события
tagged = events \
    .withColumn("next_1", lead("event_type", 1).over(event_w)) \
    .withColumn("next_2", lead("event_type", 2).over(event_w)) \
    .withColumn("next_3", lead("event_type", 3).over(event_w)) \
    .withColumn("next_1_date", lead("event_time", 1).over(event_w)) \
    .withColumn(
        "converts_in_3",
        col("next_1").isin("purchase") |
        col("next_2").isin("purchase") |
        col("next_3").isin("purchase")
    ) \
    .withColumn(
        "days_to_purchase",
        when(col("next_1") == "purchase", datediff("next_1_date", "event_time"))
        .when(col("next_2") == "purchase", datediff(lead("event_time", 2).over(event_w), "event_time"))
        .when(col("next_3") == "purchase", datediff(lead("event_time", 3).over(event_w), "event_time"))
    )

# Конверсия среди add_to_cart
atc = tagged.filter(col("event_type") == "add_to_cart")
atc.select("user_id", "event_time", "converts_in_3", "days_to_purchase").show()

conversion_rate = atc.agg(avg(col("converts_in_3").cast("double")).alias("conversion_rate"))
conversion_rate.show()
# +---------------+
# |conversion_rate|
# +---------------+
# |           0.33|  ← 1 из 3 add_to_cart привёл к purchase в 3 шагах
# +---------------+

Задание 6: percent_rank и сегментация¶

По датасету заказов (customer_id, region, total_spent) вычислите percent_rank по выручке внутри каждого региона. Добавьте колонку value_segment:

• VIP - top 20% (percent_rank >= 0.8)
• HIGH - следующие 30% (percent_rank >= 0.5)
• MID - следующие 30% (percent_rank >= 0.2)
• LOW - нижние 20%

Выведите распределение клиентов по сегментам в каждом регионе.

from pyspark.sql import Window
from pyspark.sql.functions import percent_rank, when, col, count

pr_w = Window.partitionBy("region").orderBy("total_spent")

# Агрегируем total_spent по клиенту и региону
customer_totals = orders.groupBy("region", "customer_id").agg(
    F.sum("amount").alias("total_spent")
)

segmented = customer_totals \
    .withColumn("pct", percent_rank().over(pr_w)) \
    .withColumn(
        "value_segment",
        when(col("pct") >= 0.8, "VIP")
        .when(col("pct") >= 0.5, "HIGH")
        .when(col("pct") >= 0.2, "MID")
        .otherwise("LOW")
    )

# Распределение сегментов по регионам
distribution = segmented \
    .groupBy("region", "value_segment") \
    .agg(count("*").alias("customers")) \
    .orderBy("region", "value_segment")

distribution.show()
# +------+-------------+---------+
# |region|value_segment|customers|
# +------+-------------+---------+
# |   MSK|          VIP|       48|
# |   MSK|         HIGH|       71|
# |   MSK|          MID|       73|
# |   MSK|          LOW|       48|
# ...

Задание 7: Explain план¶

Напишите pipeline с двумя разными WindowSpec (разный partitionBy). Вызовите explain("extended"). Убедитесь, что план содержит два Exchange (два shuffle). Переформулируйте задачу так, чтобы использовался один WindowSpec.

from pyspark.sql import Window
from pyspark.sql.functions import row_number, rank, col

orders = spark.read.parquet("/data/orders/")

# --- Версия с двумя WindowSpec → два Exchange ---
w_customer = Window.partitionBy("customer_id").orderBy("order_date")
w_region   = Window.partitionBy("region").orderBy(col("amount").desc())

bad = orders \
    .withColumn("customer_seq",  row_number().over(w_customer)) \
    .withColumn("region_rank",   rank().over(w_region))

bad.explain("extended")
# В Physical Plan увидите два Exchange (shuffle):
# Exchange hashpartitioning(customer_id, ...)
# Exchange hashpartitioning(region, ...)

# --- Версия с одним WindowSpec (если задача это позволяет) ---
# Например, если нужна нумерация внутри региона (а не по клиенту):
w_region_time = Window.partitionBy("region").orderBy("order_date")

good = orders \
    .withColumn("region_seq",   row_number().over(w_region_time)) \
    .withColumn("region_rank",  rank().over(
        Window.partitionBy("region").orderBy(col("amount").desc())
    ))
# Два WindowSpec с одинаковым partitionBy("region") —
# Spark может объединить их в один shuffle.
# Проверьте explain: Exchange должен быть один.

good.explain("extended")