12. Spark in Practice

• 7–11장에서 Scheduler, Shuffle, Memory, Catalyst, AQE의 내부를 봤음
• 이번 장은 그 지식을 장애나 성능 문제를 볼 때 쓰는 순서로 다시 엮음
  • 느릴 때: Spark UI에서 어디를 보나
  • 죽을 때: 어느 메모리가 부족했나
  • 잘 끝났는데 다음 단계가 느릴 때: 출력 파일이 몇 개인가

flowchart TD
  symptom["증상"] --> slow["느리다"]
  symptom --> dead["죽는다"]
  symptom --> files["읽는 쪽이 느리다"]
  slow --> ui["12.1 Spark UI<br/>Stage · Skew · Spill · Shuffle"]
  dead --> oom["12.2 OOM 부검<br/>Driver · Heap · Container"]
  files --> small["12.4 Small Files<br/>쓰기 직전 파티션 수"]
  ui --> fix["12.3 처방<br/>싼 것부터"]
  oom --> fix

12.1 Reading the Spark UI

탭 지도

Stage 상세에서 꼭 보는 세 가지

① Summary Metrics의 Max vs Median

Task Duration
Min     25th    Median   75th    Max
8s      10s     12s      14s     35min   ← Skew

• 75번째 백분위까지는 고른데 Max만 튀면 “파티션 하나가 문제”
• Duration뿐 아니라 Shuffle Read Size도 같은 모양이면 데이터 Skew로 확정
• Duration만 튀고 크기는 고르면 특정 노드 문제(느린 디스크, GC)일 수 있으니 해당 Task의 호스트를 확인

② Spill (Memory / Disk)

• Spill이 있다는 것은 Execution 메모리가 모자라 디스크로 흘렸다는 뜻 (8.2, 9.3)
• 죽지는 않지만 느려짐
• Spill이 크면 Task당 데이터가 메모리 몫보다 크다는 뜻이므로, 파티션 수를 늘리라는 신호

③ Shuffle Read / Write 크기

• “이 Stage가 왜 느린가"의 대부분이 여기서 설명됨
• GB 단위면 그게 병목이므로 Broadcast Join, 미리 Filter하기, 파티셔닝 재설계를 고민

SQL 탭에서 보는 것

• Exchange 노드는 Shuffle. 개수가 곧 비용 (10.7)
• AQEShuffleRead 노드의 coalesced, skewed 표시로 AQE가 무엇을 했는지 확인 (11.4, 11.5)
• 최종 계획이 처음 계획과 달라졌는지 (SMJ에서 BHJ로 전환 등, 11.6)
• 연산자 이름 앞의 *(N)으로 Whole-Stage Codegen 융합 여부 확인 (10.8)
• 각 노드의 number of output rows로 카디널리티 추정이 실제와 얼마나 달랐는지 확인

증상별 경로

팁

• skipped stage(회색)는 좋은 것: Shuffle 파일이나 캐시를 재사용해 건너뛴 것 (7.3)
• Stages 탭 Event Timeline에서 Task들이 계단처럼 차례로 실행되면 병렬도가 부족한 것
• History Server는 이벤트 로그를 기반으로 하므로, 실행 중에만 보이는 메트릭(현재 메모리 사용량 등) 일부는 없음

12.2 OOM Postmortem

9장의 결론에서 출발함.

Spark가 관리하는 메모리(Shuffle, 정렬, 집계)는 부족하면 Spill할 뿐 죽지 않는다.

그러니 OOM으로 죽었다면 대부분 그 관리 바깥에서 터진 것. “바깥이 어디냐"를 찾는 것이 부검.

누가 죽었나: 세 갈래

① Driver OOM

• 증상: Driver 로그에 java.lang.OutOfMemoryError가 찍히고 Job 전체가 죽음

• 대응: collect를 없애고 파일이나 테이블로 쓰기, Broadcast 임계값을 원래대로 되돌리기, 반복문에는 중간 Checkpoint 넣기. Driver 메모리 증설은 최후의 수단

② Executor Heap OOM

• 증상: Executor 로그에 OutOfMemoryError: Java heap space가 찍히고, Task를 재시도하다 결국 Stage가 실패함

• 대응: 병렬도를 올려 파티션당 데이터를 줄이는 것이 가장 싼 처방. Skew면 AQE Skew Join이 동작하는지 확인하고 Salting 적용. UDF 안의 자료구조와 groupByKey는 없앰

③ Container Kill: Heap OOM이 아닌데 죽음

가장 헷갈리는 경우.

• 증상: OutOfMemoryError 로그가 없는데 Executor가 사라짐
  • Exit code 137 (SIGKILL)
  • Kubernetes OOMKilled
  • YARN “Container killed … exceeding memory limits”
• 원인: JVM Heap은 괜찮은데 Heap과 Heap 밖을 합친 프로세스 전체 메모리가 컨테이너 한도를 넘음
• Heap 밖 메모리를 쓰는 것들
  • 네트워크 버퍼(Netty, Shuffle fetch), 직렬화 버퍼
  • Parquet·압축 라이브러리의 네이티브 메모리
  • PySpark Python Worker: JVM 밖 별도 프로세스라 Heap 설정과 무관
  • Tungsten off-heap (spark.memory.offHeap.size를 켰다면 한도 계산에 더해야 함, 9.4)
• 대응: spark.executor.memoryOverhead를 늘림 (기본 Executor 메모리의 10%, 최소 384MB. PySpark면 부족하기 쉬움). Heap을 줄이고 Overhead를 늘리도록 다시 나누거나, 파티션 크기를 줄임

flowchart TD
  died["Executor / Driver 사망"] --> where{"어디서?"}
  where -- "Driver" --> d["collect · Broadcast · 결과 누적 · 계획 폭발"]
  where -- "Executor" --> log{"로그에 OutOfMemoryError?"}
  log -- "있음: Java heap space" --> heap["Heap OOM<br/>Skew · 사용자 객체 · groupByKey"]
  log -- "없음: exit 137 · OOMKilled" --> container["Container Kill<br/>Overhead 부족 · Python Worker · 네이티브 메모리"]
  heap --> heapFix["파티션 늘리기 · Skew 해소 · 코드 수정"]
  container --> containerFix["memoryOverhead 증설 · Heap 재배분"]
  d --> dFix["collect 제거 · Broadcast 임계값 원복"]

증거 수집 순서

1. Exit code와 에러 문구부터: OutOfMemoryError(Heap)인지, 137이나 OOMKilled(Container)인지에 따라 처방이 갈림
2. 어느 Stage에서: 죽은 Task들이 속한 Stage의 Shuffle Read 크기와 Spill 여부 (12.1)
3. Skew 확인: Summary Metrics Max vs Median. Skew면 메모리 증설은 미봉책이고 분포를 고쳐야 함
4. 재현 범위 축소: 문제 파티션이나 Key만 WHERE로 걸러 단독 실행

12.3 Prescriptions, Cheapest First

1. 파티션 수 늘리기 (파티션당 데이터 줄이기)
   • 대부분 여기서 해결됨
   • SQL이면 initialPartitionNum을 크게 두고 AQE에 맡기기 (11.4)
2. 코드 수정: collect, groupByKey, UDF 안에서 쌓는 자료구조를 없앰
3. Skew 대응: AQE Skew Join 동작 확인, 안 되면 Salting (5.4, 11.5)
4. memoryOverhead 증설: Container Kill일 때만 해당
5. Executor 메모리 증설: 마지막
   • 원인을 모른 채 메모리만 올리면 데이터가 더 커지는 날 같은 문제가 재발함
   • Core 수를 그대로 두고 메모리만 늘리면 Task당 몫이 늘어남 (9.3)

경험칙:

• Task 수는 Core당 2–3개 정도
• Shuffle 파티션 하나는 수십–200MB 정도가 무난

Spill할 수 있는 것은 안 죽는다. 죽는 것은 Spill 못 하는 곳(Driver 수집, 사용자 객체, Heap 밖)에 있다. 그래서 부검의 첫 질문은 “메모리가 부족했나"가 아니라 “어느 메모리가 부족했나"다.

12.4 The Small Files Problem

“병렬도를 올려라"의 뒷면. 그 비용은 파일 시스템 쪽에서 치르게 됨.

왜 생기나

출력 파일 수 = 쓰기 직전의 파티션 수

1. Shuffle 기본 200 파티션: 결과가 1GB여도 파일 200개(파일당 5MB)
2. 실행 횟수 × 파티션 수: 매시간 도는 Job이 200개씩 쓰면 하루 4,800개, 1년이면 175만 개
3. Streaming 마이크로배치: 몇 초마다 커밋할 때마다 파일이 생김. 구조적으로 작은 파일을 계속 만들어 냄
4. 파티션 컬럼으로 쓰기: partitionBy("date", "region")이면 조합마다 디렉터리가 생기고 그 안에 파일이 생김. 카디널리티가 높은 컬럼으로 나누면 파일 수가 폭발함

왜 나쁜가

• 파일당 고정 비용: open과 close, 메타데이터 조회, Parquet footer 읽기
  • 5MB 파일 1만 개를 읽는 비용 ≫ 같은 양을 큰 파일 몇 개로 읽는 비용
• 목록 조회가 느려짐: 오브젝트 스토리지 LIST는 페이지당 1,000개라서, 파일이 수백만 개면 읽기 전에 목록을 뽑는 데만 분 단위가 걸림
  • HDFS 시절엔 NameNode 메모리를 갉아먹는 문제였음. 같은 문제가 모습만 바뀐 것
• Parquet의 장점이 사라짐: 컬럼 압축과 Row Group 건너뛰기는 파일이 수십–수백 MB는 돼야 의미가 있음
• Spark로 읽을 때도 손해: 작은 파일은 합쳐 읽기도 하지만(spark.sql.files.maxPartitionBytes, openCostInBytes), 파일마다 여는 비용은 그대로
• 다른 시스템으로 적재(load)할 때 Job당 파일 수 제한에 걸리기도 함

해법

쓰기 시점 (예방이 최선)

• 쓰기 직전에 coalesce(n) 또는 repartition(n)
  • 목표: 파일당 128MB–1GB, n ≈ 전체 크기 / 256MB
  • coalesce: Shuffle 없이 파티션을 병합함. 싸지만 너무 줄이면 앞 Stage의 병렬성까지 줄어듦 (Narrow라서 같은 Stage로 묶이기 때문)
  • repartition: Shuffle 하나를 추가해 고르게 나눔. Skew가 있거나 앞 Stage의 병렬성을 지켜야 하면 이쪽
• spark.sql.files.maxRecordsPerFile: 파일당 행 수 상한 (파일이 너무 커지는 반대 방향 안전장치)
• SQL이라면 쓰기 직전에 REPARTITION 힌트를 넣으면 AQE가 그 Shuffle도 병합해 줌

AQE 병합은 절반만 해결

• 11.4의 병합은 Shuffle 읽기 파티션에만 적용됨
• 마지막 Stage가 Shuffle 없이 끝나면(스캔 → Filter → 쓰기) 출력 파일 수에는 영향이 없음
• “AQE를 켰는데 왜 파일이 많지"에 대한 답이 이것

flowchart LR
  scan["Scan<br/>파일 1,000개"] --> filter["Filter"]
  filter --> write1["Write<br/>파일 1,000개"]
  scan2["Scan"] --> agg["groupBy<br/>Shuffle 200"]
  agg --> aqe["AQE 병합<br/>200 → 12"]
  aqe --> write2["Write<br/>파일 12개"]

사후 (Compaction)

• 작은 파일들을 주기적으로 읽어 큰 파일로 다시 쓰는 별도 Job
• Streaming Sink에는 사실상 필수
• Iceberg·Delta Lake 같은 Table Format은 Compaction을 기능으로 내장함. 실무에서 Table Format이 필요해지는 이유 중 하나
• 관리형 웨어하우스의 내부 스토리지는 보통 자동으로 최적화되므로, 문제는 레이크를 직접 다루는 구간(스테이징, export)에 남아 있음

감 잡기

• 건강한 상태: 파일당 128MB–1GB
• 진단: 출력 경로의 파일 수와 평균 크기부터 확인 (파일 수 × 평균 크기 vs 전체 / 256MB)
• 원칙: 읽기 병렬도는 크게, 쓰기 파일 수는 작게
  • 둘은 다른 문제이고, 쓰기 직전의 repartition이 그 경계선