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

자동차 조립 라인을 생각해 보세요. A 공정이 차체를 만들고, B 공정이 도장하고, C 공정이 마무리합니다. 만약 B 공정이 결과를 어딘가에 저장하지 않은 채 C 공정이 실패한다면, A부터 전부 다시 해야 합니다. 데이터 파이프라인도 똑같습니다. 원천 정제 → 차원 조인 → 집계 → 마트 분기, 이 여러 단계(잡)를 어떻게 엮느냐가 파이프라인의 안정성과 비용을 좌우합니다. 잘못 엮으면 같은 데이터를 반복 계산하거나, 한 단계 실패에 전체를 처음부터 다시 돌립니다.

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

이 글에서 배우는 것

• Spark 잡·스테이지·태스크가 어떻게 나뉘는지, 액션이 왜 재계산을 유발하는지
• cache / persist / checkpoint 를 언제, 어떻게 써야 하는지
• 한 Application vs 여러 Application — 언제 쪼개는 것이 유리한지
• Airflow 같은 외부 오케스트레이터와 Spark 의 역할을 어떻게 나누는지
• 멱등성과 부분 재시도 설계로 실패 비용을 줄이는 방법

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

"잡을 두 번 돌렸는데 같은 데이터를 왜 두 번 읽지?" — 이 의문의 뿌리가 여기 있습니다. 용어를 한 번만 정리하면 이후 모든 것이 명확해집니다. Spark 내부에서 잡(Job)은 액션 하나가 트리거하는 실행 단위입니다.

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 을 공유한다면, 첫 번째 계산 결과를 붙잡아 두는 것이 답입니다 — 바로 persist 로 캐시해서 재계산을 막는 것이죠.

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 끊기

cache 는 "어떻게 이 결과를 만들었는지(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

파이프라인 규모가 커질수록 "한 spark-submit 에 다 넣을까, 단계별로 나눌까"라는 갈림길을 만납니다. 정답은 상황에 따라 다르지만, 판단 기준은 명확합니다.

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

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

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

Spark 는 뛰어난 분산 실행 엔진이지만, "오늘 오전 2시에 실행", "Bronze 가 끝나야 Silver 시작", "실패 시 3회 재시도 후 슬랙 알림" 같은 요구는 Spark 의 영역 밖입니다. 잡들 사이의 의존성·스케줄·재시도는 외부 오케스트레이터(Airflow, Dagster, Argo 등)의 몫이고, Spark 는 "한 잡 안의 실행"을 최적화하는 데 집중해야 합니다. 역할을 섞지 마세요.

오케스트레이터(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. 부분 재시도 — 단계 경계를 테이블로

멱등한 단계가 준비됐다면, 이제 "어디서부터 다시 시작할까"를 결정해야 합니다. 긴 파이프라인의 핵심 설계는 "실패 시 처음부터가 아니라 실패 지점부터" 재개하는 것입니다. 비결은 간단합니다 — 단계 경계를 영속 테이블로 저장하는 것이죠.

한 문장으로: 각 단계가 자기 결과를 테이블에 저장하면, 어느 단계에서 실패해도 그 단계만 다시 돌리면 됩니다.

# 각 단계가 자기 입력을 테이블에서 읽고, 자기 출력을 테이블로 씀
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 는 단계 실행에 집중하게 하는 것. 여기에 각 단계를 멱등하게 만들고 경계를 영속 테이블로 두면 — 한 단계가 실패해도 그 단계만 다시 돌리는, 견고하고 비용 효율적인 파이프라인이 됩니다.

마치며 — 핵심 요약

• 액션마다 재계산: count(), write 같은 액션 하나하나가 새 잡을 만듭니다. 공통 DataFrame 은 반드시 persist 하세요.
• cache vs checkpoint: 재사용이 목적이면 cache/persist, 반복 알고리즘처럼 lineage 가 폭발하는 경우엔 checkpoint 로 계보를 끊어야 합니다.
• 역할 분담이 전부: Spark 는 단계 실행, 오케스트레이터(Airflow 등)는 의존성·스케줄·재시도. 이 경계를 흐리면 둘 다 복잡해집니다.
• 멱등성은 협상 불가: append 대신 파티션 overwrite 나 MERGE 로 적재해야 안전하게 재시도할 수 있습니다.
• 단계 경계는 테이블로: 중간 결과를 메모리에만 두면 끝부분 실패가 전체 재계산으로 이어집니다. Bronze → Silver → Gold 각 경계를 영속 테이블에 저장하세요.
• 클러스터 공유 시 FAIR 스케줄러: 독립 단계 병렬 실행엔 FAIR 모드와 동적 할당을 함께 쓰면 자원 경합을 크게 줄일 수 있습니다.

여기서 소개한 패턴들을 하나씩 실제 파이프라인에 적용해 보세요. 처음엔 작은 것부터 — persist 하나, 멱등 쓰기 하나 — 바꿔도 파이프라인의 안정성이 눈에 띄게 달라집니다.

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

— Data Dynamics 엔지니어링 팀