PySpark 멀티잡 오케스트레이션 — 의존성, 캐시 재사용, 재시도

실무 데이터 파이프라인은 잡 하나로 끝나지 않습니다. 원천을 정제하고, 여러 차원과 조인하고, 집계하고, 여러 마트로 분기합니다. 이 여러 단계(잡)를 어떻게 엮느냐가 파이프라인의 안정성과 비용을 좌우합니다. 잘못 엮으면 같은 데이터를 반복 계산하거나, 한 단계 실패에 전체를 처음부터 다시 돌립니다.

이 글은 Spark 내부의 잡 경계 이해, cache 로 공통 결과 재사용, 외부 오케스트레이터(Airflow 등)와의 역할 분담, 그리고 멱등·부분 재시도 설계를 정리합니다.

1. 먼저 — Spark 의 "잡"은 무엇인가

용어를 정리해야 합니다. Spark 내부에서 잡(Job)은 액션 하나가 트리거하는 실행 단위입니다.

Application (spark-submit 1회)
  └─ Job (액션마다 1개: count, write, collect, ...)
        └─ Stage (셔플 경계로 분할)
              └─ Task (파티션 단위 병렬)

df2 = df.filter(...).groupBy(...).agg(...)   # 변환 — 잡 아님(lazy)
df2.write.parquet("a")    # 액션 → Job 1
df2.count()               # 액션 → Job 2 (df2 를 또 계산!)

핵심 함정: 액션마다 전체 변환이 다시 실행됩니다. df2 를 write 하고 count 하면, df2 가 두 번 계산됩니다. 이게 "왜 같은 걸 두 번 읽지?"의 정체입니다.

2. cache / persist — 공통 결과 재사용

여러 액션이 같은 DataFrame 을 쓴다면, 캐시해서 재계산을 막습니다.

from pyspark import StorageLevel
 
shared = df.filter(...).join(dim, "key")   # 여러 곳에서 쓰는 중간 결과
shared.persist(StorageLevel.MEMORY_AND_DISK)
 
shared.write.parquet("out_a")              # Job 1 (여기서 계산 + 캐시)
shared.groupBy("k").count().write...       # Job 2 (캐시 재사용 — 재계산 안 함)
shared.filter("x>0").write...              # Job 3 (캐시 재사용)
 
shared.unpersist()                          # 다 쓰면 즉시 해제

(캐시가 execution 메모리를 굶기는 문제는 별도 글 "PySpark Executor OOM 정복" 참고.)

3. 체크포인트 — 긴 lineage 끊기

캐시는 lineage(계보)를 유지하지만, 반복·복잡한 파이프라인에서는 계보가 너무 길어져 플래너가 느려지고 실패 시 재계산이 비쌉니다. checkpoint 는 결과를 디스크에 저장하고 계보를 잘라냅니다.

spark.sparkContext.setCheckpointDir("s3://bucket/checkpoints")
 
# 반복/복잡 단계 후 계보 절단
result = complex_iterative_step(df)
result = result.checkpoint()   # 디스크 저장 + lineage truncate

반복 알고리즘(그래프·계층)에서 checkpoint 는 필수입니다(별도 글 "PySpark GraphFrames", "재귀·계층 데이터").

4. 한 Application 안에서 여러 잡 vs 여러 Application

선택 기준:
  강결합·중간결과 재사용 多  → 한 Application 안에서
  단계 독립·재시도 단위 분리 → 여러 Application(스토리지로 연결)

보통 단계 간 경계는 테이블(Iceberg/Delta)로 두고, 각 단계를 독립 잡으로 만드는 편이 운영(재시도·모니터링)에 유리합니다.

5. 외부 오케스트레이션 — Airflow 등과 역할 분담

Spark 는 "한 잡 안의 실행"을 최적화하지만, 잡들 사이의 의존성·스케줄·재시도는 외부 오케스트레이터(Airflow, Dagster, Argo 등)의 몫입니다. 역할을 섞지 마세요.

오케스트레이터(Airflow)         Spark
  - DAG 의존성(A→B→C)            - 한 잡 내 분산 실행
  - 스케줄(매일 2시)             - 셔플·조인·집계 최적화
  - 재시도·알림·SLA              - 파티션·메모리
  - 백필 트리거                  - (단계 내부)

# Airflow 개념: 각 Spark 잡을 태스크로, 의존성을 DAG 로
# bronze >> silver >> gold  (각각 SparkSubmitOperator)

안티패턴: 한 거대한 Spark Application 안에 모든 단계를 넣고 try/except 로 흐름 제어. 단계별 재시도·관측이 어려워집니다. 의존성과 재시도는 오케스트레이터에게, Spark 는 단계 실행에 집중하게 하세요.

6. 멱등성 — 재시도의 전제

오케스트레이터가 실패한 단계를 재시도할 수 있으려면, 각 단계가 멱등해야 합니다(재실행해도 결과 동일). append 로 적재하면 재시도가 중복을 만듭니다.

# ❌ append: 재시도 시 중복
result.write.mode("append").save(table)
 
# ✅ 멱등: 파티션 교체 또는 MERGE
(result.write.format("delta").mode("overwrite")
    .option("replaceWhere", f"dt = '{run_date}'").save(table))

(멱등 백필·파티션 교체는 별도 글 "PySpark 동적 파티션 덮어쓰기와 멱등 백필", MERGE 는 "대규모 중복 제거와 SCD Type 2" 참고.) 각 단계가 멱등이면, 실패한 단계만 다시 돌려도 안전합니다.

7. 부분 재시도 — 단계 경계를 테이블로

긴 파이프라인의 핵심 설계는 "실패 시 처음부터가 아니라 실패 지점부터" 재개하는 것입니다. 단계 경계를 영속 테이블로 두면 가능합니다.

Bronze ──저장──> Silver ──저장──> Gold
   각 단계 결과가 테이블로 영속 → Silver 실패 시 Bronze 부터 안 함, Silver 만 재실행

# 각 단계가 자기 입력을 테이블에서 읽고, 자기 출력을 테이블로 씀
def silver_step(run_date):
    bronze = spark.read.table("bronze.events").where(f"dt='{run_date}'")
    cleaned = transform(bronze)
    write_idempotent("silver.events", cleaned, run_date)

중간 결과를 메모리에만 들고 한 잡으로 처리하면, 끝부분 실패가 전체 재계산을 부릅니다. 비싼 단계 사이에는 영속 경계를 두세요.

8. 동시 실행과 자원 — 한 클러스터의 여러 잡

여러 잡이 한 클러스터를 공유하면 자원 경합이 생깁니다.

# 한 Application 안에서 여러 액션을 스레드로 병렬 실행 시 FAIR 권장
spark.conf.set("spark.scheduler.mode", "FAIR")

9. 정리

멀티잡 파이프라인의 핵심은 두 가지 분리입니다. 첫째, Spark 내부의 잡 경계를 이해해 액션마다의 재계산을 캐시·체크포인트로 다스리는 것. 둘째, 단계 간 의존성·재시도는 오케스트레이터에 맡기고 Spark 는 단계 실행에 집중하게 하는 것. 여기에 각 단계를 멱등하게 만들고 경계를 영속 테이블로 두면 — 한 단계가 실패해도 그 단계만 다시 돌리는, 견고하고 비용 효율적인 파이프라인이 됩니다.

이 글은 Spark 3.5 기준으로 작성되었습니다. 멀티스테이지 데이터 파이프라인·오케스트레이션 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀