분석 window와 시간 구간

분석 window는 원래 행과 열을 유지하면서 같은 그룹의 집계나 순위 값을 붙입니다. 집계로 행을 줄이는 groupBy와 구분합니다.

집계 후 다시 조인하는 방식

customer와 spend 열이 있는 거래 표에서 고객별 평균을 모든 거래에 붙이려면 집계한 뒤 같은 고객으로 조인할 수 있습니다.

from pyspark.sql import functions as F


def average_with_join(transactions):
    averages = transactions.groupBy("customer").agg(
        F.avg("spend").alias("average_spend")
    )
    return transactions.join(averages, ["customer"], "left_outer")

고객별 최대 지출과 같은 거래에만 최대값을 붙이려면 고객과 지출 값을 함께 비교합니다. 아래는 원래 행을 유지하되 최대값과 일치하지 않는 행의 max_spend는 NULL이 되는 별도의 조건입니다.

from pyspark.sql import functions as F


def maximum_on_matching_rows(transactions):
    maximums = transactions.groupBy("customer").agg(
        F.max("spend").alias("max_spend")
    ).alias("m")
    rows = transactions.alias("t")
    condition = (
        (F.col("t.customer") == F.col("m.customer"))
        & (F.col("t.spend") == F.col("m.max_spend"))
    )
    return rows.join(maximums, condition, "left_outer").select(
        "t.*", F.col("m.max_spend")
    )

고객 전체의 평균과 최대값을 window로 붙이기

고객의 모든 거래에 같은 평균/최댓값을 붙이는 목적이면 전체 partition을 대상으로 계산합니다.

from pyspark.sql import functions as F
from pyspark.sql.window import Window


def customer_statistics(transactions):
    customer_window = Window.partitionBy("customer")
    return (
        transactions
        .withColumn("average_spend", F.avg("spend").over(customer_window))
        .withColumn("max_spend", F.max("spend").over(customer_window))
    )

정렬을 지정하지 않은 기본 frame은 partition 전체입니다. 정렬을 지정하면 기본값이 RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW가 되어 현재 정렬 값까지 누적하는 계산으로 바뀝니다. 전체 평균을 원하면서 무심코 orderBy를 추가하면 계산 범위가 달라집니다.

customer가 NULL인 행은 window에서 같은 NULL partition에 속합니다. 일반 등가 조인은 NULL 키끼리 일치시키지 않으므로, 집계 후 조인과 window를 비교할 때는 고객 키의 NULL 정책도 맞춰야 합니다.

정렬이 있으면서 전체 그룹을 사용하려면 rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)처럼 범위를 명시할 수 있습니다.

순서와 frame을 명시한 누적값

다음 예제의 event_time과 transaction_id는 NULL이 없고, 고객 안에서 두 열의 조합이 유일하다고 가정합니다. 동일 시각에서는 transaction_id 순서로 처리한다는 정책입니다.

from pyspark.sql import functions as F
from pyspark.sql.window import Window


def running_spend(transactions):
    ordered = Window.partitionBy("customer").orderBy("event_time", "transaction_id")
    running = ordered.rowsBetween(Window.unboundedPreceding, Window.currentRow)
    return (
        transactions
        .withColumn("running_total", F.sum("spend").over(running))
        .withColumn("previous_spend", F.lag("spend", offset=1).over(ordered))
        .withColumn("sequence", F.row_number().over(ordered))
    )

rowsBetween은 행 위치의 범위이며 rangeBetween은 정렬 값의 범위를 정의합니다. 두 frame의 시작/끝 경계는 포함됩니다. 동률을 해소할 수 없는 정렬로 row_number를 계산하면 그 동률 안의 순서는 보장되지 않습니다.

순위와 이전/다음 행

이 함수들은 정의한 WindowSpec에 .over()로 적용합니다. 지출 금액의 동률에 같은 순위를 주려면 해당 순위 window의 정렬 키에 transaction_id를 추가하지 않습니다. 반대로 행별 고정 순번이 필요하면 실제 업무 순서와 동률 처리 키를 포함합니다.

F.window는 시간 구간 묶기

F.window(timeColumn, windowDuration, slideDuration=None, startTime=None)는 위 분석 WindowSpec과 다른 API입니다. Timestamp 행을 시간 구간에 배정하는 struct 열을 만들고 groupBy 등에 사용합니다.

기간은 고정 길이이며 week/day/hour/minute/second/millisecond/microsecond 단위를 사용합니다. 월 단위는 지원하지 않고, 1 day는 달력의 일광절약시간과 무관하게 고정 24시간입니다.

from pyspark.sql import functions as F


def overlapping_time_counts(events):
    return events.groupBy(
        F.window("event_time", "10 minutes", "5 minutes").alias("time_window")
    ).agg(F.count("*").alias("event_count"))

결과 struct에는 start와 end가 있습니다. 시간 구간은 시작을 포함하고 끝을 제외하므로 12:05는 [12:05,12:10)에 속하고 [12:00,12:05)에는 속하지 않습니다. 겹치는 구간을 사용하면 한 사건이 여러 구간에 포함될 수 있습니다.