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

공장 센서 데이터를 받아 보면 타임스탬프가 중간중간 툭툭 끊겨 있는 경우가 많죠. 네트워크가 잠깐 끊겼거나, 센서가 값을 안 보낸 거거든요. 금융 거래 로그도 마찬가지 — 거래가 없는 시각엔 행이 아예 없습니다. 그런데 머신러닝 모델이나 분석 도구들은 대부분 규칙적인 시계열을 전제로 설계되어 있어요. 빠진 구간을 채우고(gap fill), 간격을 재조정하고(resampling), 그러면서도 큰 파티션이 메모리를 터뜨리지 않게 하는 것 — 이것이 PySpark 시계열 처리의 핵심 난제입니다.

이 글에서 배우는 것

• 불규칙한 시계열을 규칙적 간격으로 리샘플링하는 방법
• sequence + explode 로 빠진 시각의 행을 통째로 생성하는 갭필 패턴
• last(ignorenulls) 윈도우 함수로 NULL 을 직전 값으로 채우는 forward fill
• 큰 파티션 윈도우가 OOM 을 일으키는 이유와 실전 대응 전략
• 세션화·성능 체크리스트까지 한 번에

1. 시계열의 세 가지 과제

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

세 과제는 서로 다른 도구를 씁니다. 각각이 어떻게 다른지 먼저 확인해 봅시다.

2. 리샘플링 — 시간 버킷 집계

시계열 작업 중 여러분이 가장 자주 마주칠 패턴이에요. 초 단위로 쌓인 세밀한 데이터를 1분·1시간 단위로 묶어 집계하는 거죠.

한 문장으로: F.window() 로 시간 버킷을 만들고 groupBy 로 집계하면 됩니다.

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")

date_trunc 로도 가능합니다(겹치지 않는 단순 버킷).

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

리샘플링은 데이터를 줄이는 방향이라 안전합니다. 문제는 다음 단계 — 줄였더니 중간에 빈 버킷이 생기는 갭입니다.

3. 갭필 — 없는 행 만들기

리샘플링을 마치고 나면 "센서가 1분 동안 아무 값도 안 보낸 시각"은 행 자체가 존재하지 않아요. 데이터가 없는 게 아니라, 행이 아예 없는 겁니다. 분석을 위해 모든 시각 격자를 먼저 만들고 left join 해서 빈 행을 확보해야 합니다.

# 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")

sequence(start, stop, interval) + explode 가 시각 격자를 만드는 핵심입니다. crossJoin 은 센서 수가 많을수록 결과가 커지므로, 센서별로 시간 범위가 다르다면 범위를 최대한 좁혀서 만드세요.

팁: 센서가 수백 개를 넘어가면 crossJoin 대신 센서별 (min_ts, max_ts) 를 구해 sequence 를 개별 생성하는 방식이 훨씬 효율적입니다.

4. Forward Fill (LOCF) — 빈 값을 직전 값으로

갭필로 행은 만들었지만 값은 아직 NULL 이에요. 시계열에서는 "마지막으로 측정된 값이 그 이후에도 유효하다"고 보고 직전 관측값으로 채우는 것이 일반적입니다(Last Observation Carried Forward, LOCF).

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))

F.last(..., ignorenulls=True) 가 forward fill 의 핵심입니다. (이 패턴은 as-of join 과 같은 원리입니다 — 별도 글 "PySpark As-of Join" 참고.)

Backward fill / 보간

# 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))
 
# 선형 보간은 직전·직후 값과 시간 거리로 직접 계산

5. 가장 위험한 함정 — 큰 파티션 윈도우 OOM

여기서부터가 실전에서 가장 많이 당하는 부분입니다. 윈도우 함수는 partitionBy 키 하나에 해당하는 모든 행을 단일 익스큐터 메모리에 올려야 합니다. 센서 종류가 적고 데이터 기간이 길면, 한 센서의 수억 포인트가 한 익스큐터로 몰려 OOM 이 납니다.

⚠️ 파티션 키의 카디널리티가 낮을수록(예: 센서 10개) OOM 위험이 높아집니다. 항상 파티션당 행 수를 먼저 추정해 보세요.

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

특히 orderBy 가 있는 윈도우 + unbounded 프레임은 정렬을 위해 파티션 전체를 메모리에 올려둬야 해서 위험합니다.

대응 방법은 상황에 따라 달라집니다:

# 파티션을 잘게: 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시간 이내만

핵심 트레이드오프: 파티션을 잘게 쪼개면 OOM 은 피하지만, fill 이 청크 경계를 넘지 못합니다(날짜가 바뀌면 직전 값을 못 가져옴). 경계를 넘는 fill 이 필요하면 청크 끝값을 다음 청크로 전파하는 추가 처리가 필요합니다.

6. 세션화 (시계열 패턴)

"30분 이상 비활동이면 새 세션"처럼 이벤트 간 간격을 기준으로 세션을 나누는 것도 시계열 윈도우의 대표적인 응용입니다. lag 으로 직전 이벤트 시각을 가져오고, 간격이 임계치를 넘으면 새 세션 플래그를 세우는 방식이에요.

한 문장으로: lag 으로 간격을 계산하고 sum 누적으로 세션 ID 를 부여하면 됩니다.

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))))

7. 성능 체크리스트

• 리샘플링으로 먼저 데이터 줄이기(윈도우 전에)
• 윈도우 partitionBy 키 카디널리티 확인(낮으면 OOM)
• unbounded 프레임 남발 금지 — 필요한 범위로 제한
• forward fill 거리 제한(rangeBetween 시간)
• 시각 격자 crossJoin 범위 최소화
• AQE 켜기, 스큐 점검

8. 정리

시계열 처리의 핵심은 "리샘플링으로 줄이고, 격자로 채우고, last 로 메우되, 파티션 크기를 항상 의식하는 것"입니다. 윈도우 함수는 강력하지만 partitionBy 한 파티션을 통째로 메모리에 올린다는 사실을 잊으면, 센서 몇 개짜리 데이터에서도 OOM 을 만납니다. 채움 거리와 청크 경계의 트레이드오프를 이해하면, 수십억 포인트의 시계열도 안정적으로 규칙화할 수 있습니다.

마치며 — 핵심 요약

• 리샘플링은 안전하다: F.window() 나 date_trunc 로 시간 버킷을 만들어 집계하면 데이터가 줄어들기만 해서 문제가 없습니다.
• 갭필은 sequence + explode + left join: 없는 행을 직접 만들어야 NULL 이 아닌 빈 슬롯이 생깁니다. crossJoin 범위는 최소화하세요.
• forward fill 은 last(ignorenulls) over window: 직관적이지만, 파티션 전체를 메모리에 올린다는 사실을 잊지 마세요.
• OOM 의 주범은 낮은 카디널리티 파티션: 파티션 키에 날짜를 추가하거나 fill 거리를 rangeBetween 으로 제한해 파티션 크기를 줄이세요.
• 청크 경계를 넘는 fill 은 추가 처리가 필요: 파티션을 잘게 쪼개면 OOM 은 피하지만, 경계를 넘는 직전 값을 전파하는 로직을 별도로 구현해야 합니다.
• 순서를 기억하세요: 리샘플링 → 갭필 → forward fill → OOM 점검 — 이 순서로 접근하면 대부분의 시계열 문제를 해결할 수 있습니다.

이 글은 Spark 3.5 기준으로 작성되었습니다. 대규모 시계열·IoT 데이터 파이프라인 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀