PySpark Structured Streaming 상태 관리와 Exactly-Once 적재

택배 시스템을 상상해 봅시다. 배치 처리는 하루 한 번 몰아서 배달하는 방식이라, 실패하면 다음 날 다시 보내면 됩니다. 하지만 스트리밍은 24시간 쉬지 않고 실시간으로 배달하는 시스템이죠. 장애가 나도 중복 배달도, 분실도 없어야 합니다. PySpark Structured Streaming 에서 가장 어려운 두 가지가 바로 이것입니다 — 무한히 커지는 상태(state) 와 exactly-once 정합성.

이 글은 워터마크로 상태를 길들이는 법, 늦게 도착한(late) 데이터 처리, 그리고 foreachBatch + Lakehouse 멱등 MERGE 로 exactly-once 적재를 구현하는 실전 패턴을 정리합니다.

이 글에서 배우는 것

• 스트리밍 상태가 왜 무한히 커지는지, 그리고 워터마크로 어떻게 정리하는지
• 상태를 쌓는 연산(집계·조인·dedup)에 워터마크가 왜 필수인지
• 커스텀 상태를 관리할 때 타임아웃 설정이 빠지면 어떤 일이 생기는지
• 체크포인트 + 멱등 싱크로 exactly-once 를 달성하는 원리
• foreachBatch + Iceberg/Delta MERGE 로 스트리밍에서 안전한 upsert 를 구현하는 법

1. 두 가지 핵심 난제

스트리밍 파이프라인을 운영하다 보면 결국 두 가지 문제에 부딪힙니다. 먼저 어떤 문제인지 구체적으로 살펴봅시다.

난제 ①  상태 무한 증가
  스트리밍 집계·조인·dedup 은 상태를 메모리/체크포인트에 보관
  → 정리 안 하면 상태가 영원히 커져 OOM
 
난제 ②  Exactly-once
  장애 → 재시작 → 마지막 배치 재처리
  → 잘못하면 같은 데이터를 두 번 쓰거나(중복) 빠뜨림(유실)

이 둘을 각각 워터마크와 멱등 싱크로 해결합니다.

한 문장으로: 상태는 워터마크로 정리하고, 정합성은 멱등 싱크로 보장한다는 것이 Structured Streaming 운영의 핵심입니다.

2. 워터마크 — 상태를 길들이는 핵심

워터마크는 "이 시각보다 더 늦게 오는 데이터는 버린다"는 경계입니다. 강물에 선을 하나 그어 "이 선보다 오래된 부분은 이제 신경 쓰지 않겠다"고 선언하는 것과 같죠. 이 경계 덕분에 Spark 는 그보다 오래된 상태를 안전하게 정리할 수 있습니다.

from pyspark.sql import functions as F
 
agg = (stream
    .withWatermark("event_time", "10 minutes")    # 10분까지 지각 허용
    .groupBy(
        F.window("event_time", "5 minutes"),       # 5분 텀블링 윈도우
        "user_id")
    .agg(F.count("*").alias("cnt")))

동작:

• 윈도우 종료 시각 + 워터마크 지연(10분)이 지나면, 그 윈도우의 상태를 메모리에서 제거합니다.
• 워터마크 없이 집계/조인/dedup 을 하면 상태가 무한히 쌓여 결국 OOM 입니다.

워터마크 지연은 "실제로 데이터가 얼마나 늦게 오는가"의 분포로 정합니다. 너무 짧으면 정상 데이터를 유실하고, 너무 길면 상태가 비대해집니다.

3. 상태가 쌓이는 연산들

모든 연산이 상태를 필요로 하는 건 아닙니다. 하지만 아래 표의 연산들은 반드시 상태를 관리해야 합니다 — 그리고 워터마크 없이 쓰면 조용히 메모리를 잠식하다가 결국 터집니다.

# 스트리밍 dedup — 워터마크 없으면 상태 폭발
dedup = (stream
    .withWatermark("event_time", "1 hour")
    .dropDuplicates(["event_id", "event_time"]))

4. 커스텀 상태 — flatMapGroupsWithState / transformWithState

윈도우·dedup 으로 표현 안 되는 세션화·이상탐지 같은 로직은 직접 상태를 관리해야 합니다. PySpark 에서는 applyInPandasWithState, 최신 버전에서는 transformWithStateInPandas 를 사용하면 됩니다.

# 개념: 그룹(키)별로 상태를 들고, 이벤트마다 갱신, 타임아웃으로 만료
# - 세션 윈도우: 30분 비활동 시 세션 종료·출력
# - 상태에 last_seen 보관, 타임아웃 처리로 만료

핵심은 상태에 만료(timeout)를 반드시 설정하는 것입니다. 만료 없는 커스텀 상태는 가장 흔한 스트리밍 OOM 원인입니다.

⚠️ 커스텀 상태에 타임아웃을 설정하지 않으면, 오래전에 비활성화된 키의 상태가 영원히 메모리에 남습니다. 처음엔 괜찮아 보이지만 며칠·몇 주가 지나면 결국 OOM 으로 터집니다.

5. Exactly-Once — 체크포인트 + 멱등 싱크

Structured Streaming 의 exactly-once 는 두 조건의 조합으로 성립합니다.

① 재생 가능한 소스(replayable): Kafka offset 처럼 어디까지 읽었는지 복원 가능
② 멱등/트랜잭션 싱크(idempotent sink): 같은 배치를 다시 써도 결과 동일
        +
   체크포인트(checkpoint): 진행 상황(offset, 상태)을 신뢰 스토리지에 저장

query = (agg.writeStream
    .option("checkpointLocation", "s3://bucket/checkpoints/agg")  # 필수
    .outputMode("update")
    .trigger(processingTime="1 minute")
    .foreachBatch(upsert_to_iceberg)      # 멱등 싱크 (아래)
    .start())

체크포인트는 "어디까지 처리했나"를 기록합니다. 장애 후 재시작하면 체크포인트에서 이어가되, 마지막 배치는 다시 실행될 수 있습니다. 그래서 싱크가 멱등이어야 중복이 안 생기는 거죠.

6. foreachBatch + Iceberg/Delta 멱등 MERGE

파일 싱크(.format("parquet"))는 append-only 라 정교한 upsert 가 안 됩니다. foreachBatch 로 각 마이크로배치를 일반 DataFrame 처럼 다뤄, Iceberg/Delta 의 트랜잭션 MERGE 로 멱등하게 적재합니다.

def upsert_to_iceberg(micro_batch_df, batch_id):
    # 1) 배치 내 키 중복 제거 (MERGE source 는 키당 1행이어야 함)
    from pyspark.sql.window import Window
    w = Window.partitionBy("id").orderBy(F.col("updated_at").desc())
    deduped = (micro_batch_df
        .withColumn("rn", F.row_number().over(w))
        .where("rn = 1").drop("rn"))
 
    deduped.createOrReplaceTempView("updates")
 
    # 2) 멱등 MERGE: 더 최신일 때만 갱신 → 같은 배치 재실행해도 결과 동일
    micro_batch_df.sparkSession.sql("""
        MERGE INTO analytics.events AS t
        USING updates AS s
          ON t.id = s.id
        WHEN MATCHED AND s.updated_at > t.updated_at THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)
 
query = (stream.writeStream
    .option("checkpointLocation", "s3://bucket/checkpoints/events")
    .foreachBatch(upsert_to_iceberg)
    .start())

왜 exactly-once 가 되는가:

• 재시작으로 같은 batch_id 가 재처리돼도, MERGE 의 s.updated_at > t.updated_at 조건 덕에 두 번 써도 결과가 동일(멱등)합니다.
• MERGE 는 트랜잭션이라 부분 쓰기로 인한 깨진 상태가 없습니다.

이 패턴(foreachBatch + 멱등 MERGE)이 PySpark 스트리밍 exactly-once 적재의 사실상 표준입니다. 멱등성의 비결은 배치 단위 중복 제거 + updated_at 비교라는, 배치 파이프라인과 똑같은 원리입니다(별도 글 "PySpark 대규모 중복 제거와 SCD Type 2" 참고).

7. 늦게 도착한 데이터(Late Data)

데이터가 항상 제시간에 오지는 않습니다. 모바일 앱이 오프라인 상태에서 쌓아둔 이벤트를 뒤늦게 전송하는 경우가 대표적이죠. 워터마크 경계 안에 도착한 지각 데이터는 해당 윈도우에 반영됩니다. 경계를 넘은 데이터는 버려집니다(드롭).

# 드롭된 지각 데이터를 따로 보관하고 싶다면 (감사·재처리용)
# foreachBatch 안에서 워터마크 기준 이전 데이터를 별도 테이블로 분기

정책 결정:

• 비즈니스가 지각을 얼마나 허용하나 → 워터마크 지연 결정.
• 드롭 데이터를 버릴지, 별도 보관 후 배치 재처리할지(람다/카파 아키텍처).

8. 운영 체크리스트

스트리밍 파이프라인을 프로덕션에 올리기 전에 아래 항목을 모두 확인해 보세요.

• 상태 연산(집계/조인/dedup)에 워터마크 설정
• 커스텀 상태에 타임아웃(만료) 설정
• checkpointLocation 을 신뢰 스토리지(S3/HDFS)에
• 싱크는 멱등/트랜잭션(foreachBatch + MERGE)
• MERGE source 배치 내 키 중복 제거
• 트리거 간격으로 작은 파일 제어(+ 정기 컴팩션)
• 상태 크기·처리 지연(lag) 모니터링
• 체크포인트 스키마 호환성(쿼리 변경 시 주의)

9. 정리

스트리밍의 두 난제는 결국 두 도구로 수렴합니다 — 상태는 워터마크로, 정합성은 멱등 싱크로. 파일 append 로 대충 쓰면 중복·작은 파일·정합성 문제가 한꺼번에 터지지만, foreachBatch + 멱등 MERGE 패턴을 쓰면 배치 파이프라인과 동일한 정합성 보장을 스트리밍에서도 얻습니다. 24시간 멈추지 않는 파이프라인일수록, 이 기본기가 안정성을 결정합니다.

마치며 — 핵심 요약

• 워터마크는 "이 선보다 오래된 데이터는 버린다"는 경계로, 상태 연산(집계·조인·dedup) 에서 필수입니다.
• 워터마크 없는 집계·dedup 은 상태가 무한히 쌓여 OOM 으로 이어집니다.
• 커스텀 상태(flatMapGroupsWithState 계열) 에는 반드시 타임아웃(만료)을 설정해야 합니다.
• Exactly-once 의 두 기둥: 재생 가능한 소스(Kafka offset) + 멱등 싱크(MERGE) + 체크포인트.
• foreachBatch + Iceberg/Delta MERGE 는 스트리밍 exactly-once 적재의 사실상 표준 패턴입니다.
• 늦게 도착하는 데이터는 워터마크 경계 안이면 반영되고, 경계를 넘으면 드롭되며 — 드롭 정책은 비즈니스 요구에 맞게 설계하면 됩니다.

이 두 가지 도구(워터마크 + 멱등 싱크)만 제대로 이해하면, 24시간 스트리밍 파이프라인도 자신 있게 운영할 수 있습니다.

이 글은 Spark 3.5 + Iceberg/Delta 기준으로 작성되었습니다. 실시간 적재 파이프라인의 정합성·상태 관리 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀