PySpark 대규모 중복 제거와 SCD Type 2 — MERGE 로 정합성 잡기

# "어떤" 행이 남는지 보장되지 않음 — 비결정적!
df.dropDuplicates(["user_id"])

from pyspark.sql import functions as F
from pyspark.sql.window import Window
 
# user_id 별, updated_at 최신순. 동률이면 op_seq 로 타이브레이크
w = Window.partitionBy("user_id").orderBy(
    F.col("updated_at").desc(),
    F.col("op_seq").desc(),
)
 
latest = (df
    .withColumn("rn", F.row_number().over(w))
    .filter(F.col("rn") == 1)
    .drop("rn"))

# 워터마크 기반 dedup (상태 무한 증가 방지)
dedup = (stream
    .withWatermark("event_time", "1 hour")
    .dropDuplicates(["event_id", "event_time"]))

# source(변경분)에서 최신 레코드만 추린 뒤 MERGE
latest.createOrReplaceTempView("updates")
 
spark.sql("""
MERGE INTO analytics.users AS t
USING updates AS s
  ON t.user_id = s.user_id
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
""")

# MERGE 의 source 에 같은 키가 2개 이상이면 에러 또는 비결정적 결과!
# → 2장의 window 패턴으로 키당 1행 보장한 뒤 MERGE

user_id | tier   | valid_from | valid_to   | is_current
--------+--------+-----------+------------+-----------
123     | silver | 2026-01-01 | 2026-05-01 | false      ← 과거
123     | gold   | 2026-05-01 | 9999-12-31 | true       ← 현재

from pyspark.sql import functions as F
 
target = spark.table("analytics.dim_users")     # is_current=true 가 현재 버전
src = latest                                     # 최신 소스(키당 1행)
 
# 현재 버전과 비교해 '실제로 바뀐' 키만 추림
current = target.filter("is_current = true")
changed = (src.alias("s")
    .join(current.alias("t"), "user_id")
    .where("s.tier <> t.tier OR s.region <> t.region")   # 변경 감지 컬럼
    .select("s.*"))
 
# (A) 신규 키 + (B) 변경 키 → 새 버전으로 삽입
# (C) 변경된 기존 행은 valid_to 마감 + is_current=false
 
# 삽입용: 변경 키 + 신규 키
to_insert = changed.unionByName(
    src.join(current, "user_id", "left_anti")    # 아예 없던 신규 키
)
 
to_insert.createOrReplaceTempView("scd_updates")
 
spark.sql("""
MERGE INTO analytics.dim_users AS t
USING scd_updates AS s
  ON t.user_id = s.user_id AND t.is_current = true
WHEN MATCHED AND (t.tier <> s.tier OR t.region <> s.region) THEN
  UPDATE SET t.is_current = false, t.valid_to = current_date()
""")
 
# 새 버전 행 append (valid_from=오늘, valid_to=9999, is_current=true)
new_versions = to_insert.select(
    "user_id", "tier", "region",
    F.current_date().alias("valid_from"),
    F.lit("9999-12-31").cast("date").alias("valid_to"),
    F.lit(True).alias("is_current"),
)
new_versions.writeTo("analytics.dim_users").append()

1. 소스에서 키당 최신 1행 확보 (window dedup)
2. 현재 버전과 비교 → 실제 변경된 키 식별
3. MERGE: 변경된 현재 행을 is_current=false + valid_to 마감
4. append: 변경 키 + 신규 키를 새 현재 버전으로 삽입

-- 현재 값
SELECT * FROM analytics.dim_users WHERE is_current = true;
 
-- 특정 시점의 값 (point-in-time)
SELECT * FROM analytics.dim_users
WHERE '2026-03-15' BETWEEN valid_from AND valid_to;