PySpark 시계열 처리 — 갭필, 리샘플링, 그리고 큰 파티션 윈도우

① 갭필(Gap Fill)   : 빠진 시각의 행을 만들어 채우기 (없는 행 생성)
② 리샘플링         : 불규칙/세밀한 간격 → 규칙적 간격으로 (1초 → 1분)
③ Forward Fill     : 빠진 값을 직전 값으로 채우기 (LOCF)

from pyspark.sql import functions as F
 
# 1분 단위로 리샘플 (텀블링 윈도우)
resampled = (df
    .groupBy(
        "sensor_id",
        F.window("event_time", "1 minute"))
    .agg(
        F.avg("value").alias("avg_value"),
        F.max("value").alias("max_value"),
        F.count("*").alias("n")))
 
# window 구조체를 평탄화
resampled = resampled.select(
    "sensor_id",
    F.col("window.start").alias("ts"),
    "avg_value", "max_value", "n")

df.withColumn("minute", F.date_trunc("minute", "event_time")) \
  .groupBy("sensor_id", "minute").agg(F.avg("value"))

# 1) 전체 시간 격자 생성: 시작~끝을 1분 간격으로
bounds = df.agg(
    F.min("event_time").alias("t0"),
    F.max("event_time").alias("t1")).collect()[0]
 
grid = (spark.sql(f"""
    SELECT explode(sequence(
        timestamp '{bounds.t0}',
        timestamp '{bounds.t1}',
        interval 1 minute)) AS ts
"""))
 
# 2) 센서 × 시간격자 교차 (모든 센서가 모든 시각을 갖도록)
sensors = df.select("sensor_id").distinct()
full_grid = sensors.crossJoin(grid)        # 센서 수 × 시각 수
 
# 3) 실제 데이터를 격자에 left join → 없는 시각은 NULL
filled = full_grid.join(resampled, ["sensor_id", "ts"], "left")

from pyspark.sql.window import Window
 
w = (Window.partitionBy("sensor_id")
            .orderBy("ts")
            .rowsBetween(Window.unboundedPreceding, Window.currentRow))
 
# 직전의 NULL 아닌 값으로 채우기
filled = filled.withColumn(
    "value_ffill",
    F.last("value", ignorenulls=True).over(w))

# backward fill: 역순 정렬 후 last, 또는 first(ignorenulls) with following 프레임
w_back = Window.partitionBy("sensor_id").orderBy("ts") \
    .rowsBetween(Window.currentRow, Window.unboundedFollowing)
filled = filled.withColumn("value_bfill", F.first("value", ignorenulls=True).over(w_back))
 
# 선형 보간은 직전·직후 값과 시간 거리로 직접 계산

partitionBy("sensor_id")  +  센서 10개 × 각 1억 포인트
→ 한 파티션 = 1억 행이 한 익스큐터 메모리에 → OOM 💥

# 파티션을 잘게: sensor_id + 날짜로 (단, fill 이 날짜 경계를 못 넘는 점 주의)
w = Window.partitionBy("sensor_id", F.to_date("ts")).orderBy("ts")...
 
# 또는 forward fill 거리를 제한 (최근 60분만 채움)
w = (Window.partitionBy("sensor_id").orderBy(F.col("ts").cast("long"))
     .rangeBetween(-3600, 0))   # 직전 1시간 이내만

w = Window.partitionBy("user_id").orderBy("ts")
 
sessionized = (df
    .withColumn("prev_ts", F.lag("ts").over(w))
    .withColumn("gap_min",
        (F.col("ts").cast("long") - F.col("prev_ts").cast("long")) / 60)
    .withColumn("is_new",
        F.when((F.col("gap_min") > 30) | F.col("prev_ts").isNull(), 1).otherwise(0))
    .withColumn("session_id",
        F.sum("is_new").over(w.rowsBetween(Window.unboundedPreceding, Window.currentRow))))