PySpark As-of Join — "그 시점의 값"을 붙이는 Point-in-Time 조인

주식 거래 시스템을 만든다고 상상해 보세요. 오전 10시 0분 5초에 발생한 거래에 "그 순간 유효했던 환율"을 붙여야 합니다. 그런데 환율 테이블에는 10:00:00, 10:00:10, 10:00:20 … 이렇게 갱신 시각만 기록돼 있어서, 거래 시각과 환율 시각이 정확히 일치하는 경우가 거의 없습니다. 이런 상황에서 필요한 것이 바로 as-of join(point-in-time join)입니다.

"각 거래 시점에 유효했던 환율을 붙여라." "센서 이벤트마다 그 직전 보정값을 매칭하라." "피처 생성 시점의 고객 등급은?" — 이런 요구는 모두 as-of join 의 대표 사례죠. 키가 정확히 일치하는 조인이 아니라, "기준 시각 이전의 가장 가까운 값" 을 찾는 시간 기반 조인입니다.

문제는 PySpark(Spark SQL)에 as-of join 이 1급 구문으로 없다는 것입니다. 잘못 구현하면 range join 으로 데이터가 폭발하거나, 미래 정보가 새어 들어가 ML 파이프라인을 망칩니다. 이 글은 올바른 구현 패턴과 함정을 정리합니다.

이 글에서 배우는 것

• as-of join 이 무엇이고 왜 일반 equi-join 으로는 안 되는지
• range join 이 왜 데이터 폭발을 일으키는지
• union + window 패턴으로 깔끔하게 구현하는 방법
• 동시각(타이) 처리 시 주의할 타이브레이커
• ML 파이프라인에서 미래 정보 누수를 막는 방법

1. As-of Join 이란

한 줄로 설명하면, 두 타임스탬프가 정확히 맞지 않아도 "가장 최근 이전 값"을 찾아 붙여주는 조인입니다. 아래 예시를 보면 바로 감이 올 거예요.

거래(trades)                환율(rates, 시각별로 변함)
time      symbol            time      rate
10:00:05  KRW               10:00:00  1330
10:00:25  KRW               10:00:10  1331
                            10:00:20  1332
 
→ 10:00:05 거래에는 10:00:00 의 1330 (그 시점 이전 최신)
→ 10:00:25 거래에는 10:00:20 의 1332

일반 equi-join 은 time 이 정확히 같아야 매칭됩니다. 하지만 거래 시각과 환율 갱신 시각은 절대 정확히 안 맞습니다. "≤ 기준시각 중 최대" 를 찾아야 합니다 — 이것이 as-of 의 핵심입니다.

한 문장으로: "기준 시각 이전에 가장 마지막으로 알려진 값"을 찾아 붙이는 것이 as-of join입니다.

2. 왜 일반 조인으로 안 되나

직관적으로 떠오르는 두 가지 방법이 있는데, 둘 다 실전에서 문제가 생깁니다. 차례로 살펴봅시다.

시도 ① equi-join — 매칭 실패

trades.join(rates, "time")   # 시각이 정확히 같은 경우만 → 대부분 누락

시도 ② range join — 폭발

# 거래시각 이전의 모든 환율과 조인 → 거래당 수많은 행
joined = trades.join(rates,
    (trades.symbol == rates.symbol) & (rates.time <= trades.time))
# 그 후 거래별로 가장 최근 1건만 남기기 → 중간 결과가 거대해짐

range join 은 <= 조건이라 거래 하나가 그 이전 모든 환율과 매칭됩니다. 환율이 하루 수만 건이면 거래당 수만 행이 생겨 셔플·메모리가 폭발합니다. Spark 는 range join 을 잘 최적화하지 못해 사실상 cross join 에 가까워집니다.

⚠️ range join 을 "나중에 필터링하면 되겠지"라고 생각하면 큰 코 다칩니다. 중간에 폭발적으로 생성된 행들이 먼저 메모리와 셔플 비용을 삼켜버립니다.

3. 올바른 패턴 — Union + Window

두 방법 모두 함정이 있다면, 어떻게 해야 할까요? 검증된 PySpark as-of 패턴은 두 테이블을 union 한 뒤 window 로 직전 값을 끌어오는 것입니다. 마치 두 타임라인을 하나로 합쳐서 시간순으로 쭉 훑어 내려가는 것처럼 — range join 의 폭발 없이 한 번의 정렬로 해결합니다.

from pyspark.sql import functions as F
from pyspark.sql.window import Window
 
# 1) 두 데이터셋을 공통 스키마로 union (출처 구분 + 값 컬럼)
t = trades.select("symbol", "time",
                  F.lit("trade").alias("src"),
                  F.col("trade_id"),
                  F.lit(None).cast("double").alias("rate"))
 
r = rates.select("symbol", "time",
                 F.lit("rate").alias("src"),
                 F.lit(None).cast("long").alias("trade_id"),
                 F.col("rate"))
 
unioned = t.unionByName(r)
 
# 2) symbol 별 시간순으로 정렬하고, 직전까지의 마지막 rate 를 끌어옴
w = (Window.partitionBy("symbol")
            .orderBy("time")
            .rowsBetween(Window.unboundedPreceding, Window.currentRow))
 
filled = unioned.withColumn(
    "asof_rate",
    F.last("rate", ignorenulls=True).over(w))   # 핵심: 직전 최신 rate 채우기
 
# 3) 거래 행만 남기면 각 거래에 as-of rate 가 붙어 있음
result = filled.where("src = 'trade'").select("trade_id", "symbol", "time", "asof_rate")

핵심은 F.last("rate", ignorenulls=True) 입니다. 시간순으로 보면서 마지막으로 본 NULL 이 아닌 환율을 각 행에 채웁니다. 거래 행에는 원래 rate 가 NULL 이므로, 직전 환율 행의 값이 들어옵니다.

4. 동시각 처리 — 타이브레이커

거래와 환율이 같은 시각일 때 어느 것을 먼저 볼지 정해야 합니다. "거래 시점 이전(같은 시각 포함)"이라면 환율을 거래보다 먼저 정렬해야 합니다. 이 부분은 작은 디테일처럼 보이지만, 놓치면 조용히 틀린 결과가 나오는 함정이 됩니다.

# src 에 순서를 부여: 같은 time 이면 rate(0) 가 trade(1) 보다 먼저
w = (Window.partitionBy("symbol")
            .orderBy("time", F.when(F.col("src") == "rate", 0).otherwise(1)))

이 디테일을 놓치면 "같은 시각의 환율"이 거래에 안 붙는 미묘한 버그가 생깁니다.

5. 가장 위험한 함정 — 데이터 누수(미래 정보)

성능보다 더 무서운 함정이 있습니다. ML 피처 생성에서 as-of join 의 존재 이유가 바로 데이터 누수 방지입니다. 피처 시점 이후의 정보가 학습에 들어가면, 검증에서는 좋아 보이지만 실전에서 무너집니다.

방어 규칙:

• 조건은 반드시 rate.time <= trade.time (strict 하게 미래 배제). < vs <= 를 비즈니스 정의에 맞게 명확히.
• window 정렬에서 동시각 처리를 점검(4장).
• "현재값 테이블(SCD2 의 is_current)"을 그냥 조인하지 말 것 — 그건 미래값입니다. 반드시 유효기간 기반으로.

팁: as-of join 을 쓰는 진짜 이유의 절반은 성능이고 절반은 정확성(미래 정보 차단) 입니다. 특히 피처 스토어·백테스팅에서 이 누수는 치명적입니다.

6. SCD2 차원과의 as-of 조인

이력 차원(SCD2)처럼 "유효 시작일~종료일"로 관리되는 테이블이 있다면, as-of 는 유효기간 조인으로 자연스럽게 표현됩니다. 이벤트 시각이 어느 유효기간 안에 들어가는지 찾으면 되거든요.

# dim_users: valid_from, valid_to (SCD2)
result = (events.alias("e")
    .join(dim.alias("d"),
        (F.col("e.user_id") == F.col("d.user_id")) &
        (F.col("e.event_time") >= F.col("d.valid_from")) &
        (F.col("e.event_time") <  F.col("d.valid_to")))
)

유효기간이 겹치지 않게 관리되면(별도 글 "PySpark 대규모 중복 제거와 SCD Type 2" 참고) 이 조인은 이벤트당 정확히 한 행을 매칭합니다. 다만 이것도 범위 조인이라, 큰 데이터에서는 valid_from 으로 파티션 프루닝이 되도록 설계하거나 union+window 로 전환을 고려합니다.

7. 성능 팁

as-of 패턴에서 가장 자주 발생하는 성능 문제는 파티션 스큐입니다. 아래 표를 체크리스트처럼 활용해 보세요.

symbol(파티션 키)이 적으면 한 파티션이 거대해져 window 가 OOM 납니다. 키 분포를 확인하세요(스큐는 별도 글 참고).

8. 정리

As-of join 은 "Spark 에 없는 구문을 직접 구현"해야 하는 대표적 난제입니다. range join 으로 정면 돌파하면 폭발하지만, union + window 의 last(ignorenulls) 패턴이면 한 번의 정렬로 깔끔히 풀립니다. 그리고 성능만큼 중요한 것이 미래 정보 차단 — 피처·백테스팅에서 as-of 를 쓰는 진짜 이유가 여기에 있다는 점을 잊지 마세요.

마치며 — 핵심 요약

• as-of join 은 "기준 시각 이전에 가장 최신인 값"을 찾는 시간 기반 조인으로, 환율·가격·고객 등급 등 시간에 따라 변하는 모든 데이터에 적용됩니다.
• equi-join 은 시각이 정확히 일치해야 하므로 대부분 누락이 발생하고, range join 은 중간 결과가 폭발적으로 늘어납니다.
• union + last(ignorenulls) over window 패턴이 정석입니다. 두 테이블을 합쳐 시간순으로 한 번만 훑으면 됩니다.
• 같은 시각의 이벤트가 섞이면 타이브레이커(정렬 우선순위)를 반드시 명시하세요. 작은 디테일이지만 결과가 달라집니다.
• 미래 정보 누수는 ML 파이프라인에서 가장 치명적인 함정입니다. <= 조건과 SCD2 유효기간 기반 조인을 습관처럼 점검하세요.
• 파티션 키 카디널리티가 낮으면 window 가 OOM 을 일으킬 수 있으니, 키 분포를 미리 확인하고 AQE 를 켜두는 것을 권장합니다.

패턴 하나를 제대로 익혀두면, 시계열 피처 엔지니어링과 백테스팅 파이프라인에서 여러분이 마주할 대부분의 "시점 매칭" 문제를 자신 있게 풀 수 있을 겁니다.

이 글은 Spark 3.5 기준으로 작성되었습니다. 시계열 피처 엔지니어링이나 point-in-time 정합성이 중요한 파이프라인 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀