5. Memory, Caching, and Performance Basics

5.1 Executor Memory Model

• Executor 메모리는 하나의 덩어리가 아니라 용도별로 구획되어 있음

flowchart TD
  container["Executor Container Memory"]
  container --> heap["JVM Heap<br/>spark.executor.memory"]
  container --> overhead["Overhead<br/>spark.executor.memoryOverhead"]
  heap --> unified["Unified Memory<br/>spark.memory.fraction"]
  heap --> user["User Memory<br/>사용자 자료구조 · 내부 메타데이터"]
  unified --> execution["Execution Memory<br/>Shuffle · Sort · Aggregation · Join"]
  unified --> storage["Storage Memory<br/>Cache · Broadcast"]
  overhead --> external["Python Worker · Native Libraries<br/>Serialization Buffers"]

• Unified Memory Model: Execution과 Storage는 고정 경계가 아니라 서로 밀고 당김
  • Execution이 부족하면 Storage의 Cache를 Evict하고 빌려옴
  • 반대로 Storage는 Execution이 쓰고 있는 메모리를 빼았지 못함 (Execution 우선)
• Spill: Execution Memory가 부족하면 중간 결과를 디스크로 내림

Aggregation / Sort 중 메모리 부족
→ 디스크에 Spill
→ 느려지지만 죽지는 않음 (Spark UI의 Spill Metric으로 확인)

• OOM은 두 가지 방식으로 남 — 구분이 중요함

JVM OOM (java.lang.OutOfMemoryError)
→ Heap 부족: 거대한 파티션, groupByKey로 모은 큰 Key, Driver collect
→ executor.memory 증설 또는 파티션 수 조정으로 해결

Container Kill (K8s OOMKilled / YARN memory limit exceeded)
→ Heap 밖(Overhead) 부족: PySpark의 Python Worker, 네이티브 메모리
→ executor.memory를 늘려도 해결 안 됨 — memoryOverhead를 늘려야 함

• 실무 연결: EMR·K8s 에서 executor 사양을 잡을 때 조정하는 것이 바로 이 값들 (executor.memory / memoryOverhead / cores)

5.2 Cache, Persist, and Storage Levels

• cache()는 persist()의 기본값 버전 — 그리고 API마다 기본값이 다름 (고전 함정)

RDD.cache()        = persist(MEMORY_ONLY)
DataFrame.cache()  = persist(MEMORY_AND_DISK)

• 언제 캐시하나: 같은 중간 결과를 두 번 이상 사용할 때만
  • 한 번 쓰고 버리는 데이터를 캐시하면 오히려 손해: 저장 비용 + Storage 메모리 압박 + GC 부담
  • 다 썼으면 unpersist()로 명시적으로 해제
• Cache와 Checkpoint의 차이(Lineage 유지 여부)는 3.4 참고

5.3 Broadcast

• Broadcast Variable: Driver의 읽기 전용 데이터를 각 Executor에 한 번씩만 배포

Task Closure 방식
→ Task마다 복사본 전송 (1000 Task = 1000번)

Broadcast 방식
→ Executor당 1번 (10 Executor = 10번), Task들이 공유

• 4.9의 Task Closure 예시에서 large_lookup을 Broadcast로 바꿐야 했던 이유가 이것
• Broadcast Join: 작은 테이블을 모든 Executor에 배포해 Shuffle 없이 조인

SortMergeJoin (기본)
→ 양쪽 테이블 모두 Shuffle + Sort

BroadcastHashJoin
→ 작은 쪽만 전체 배포, 큰 쪽은 제자리에서 조인 — Shuffle 없음

• spark.sql.autoBroadcastJoinThreshold (기본 10MB) 이하면 자동 선택, broadcast() 힌트로 강제 가능
• 주의: 브로드캐스트할 데이터는 Driver에 모였다가 배포됨 — 크면 Driver OOM (2.6)

5.4 Data Skew

• 증상: 대부분의 Task는 금방 끝나는데 소수 Task가 Stage 전체를 붙잡음

Task Duration

Task 0   : 10s
Task 1   : 12s
...
Task 199 : 45min ← 특정 Key가 몰린 파티션

• 원인: Shuffle은 Key 기준으로 파티션을 나누는데, 특정 Key(null, 기본값, 인기 사용자·상품)에 데이터가 몰림
• 진단: Spark UI Stage 탭에서 Task의 Duration·Shuffle Read Size 분포 확인 — Max ≫ Median이면 Skew
• 대처법
  1. Skew Key 분리: null·기본값은 미리 제거하거나 따로 처리
  2. Salting: Key에 랜덤 접미사를 붙여 여러 파티션으로 분산 → 부분 집계 → 접미사 제거 후 재집계
  3. AQE Skew Join 자동 분할 (5.5)
  4. groupByKey 대신 reduceByKey·집계 함수 — Shuffle 전에 map-side 부분 집계로 이동량 감소

Salting 예시

seoul → seoul_0, seoul_1, seoul_2 (랜덤 분산)
→ 각각 부분 집계
→ 접미사 제거 후 최종 집계

5.5 Partition Count and AQE

• Shuffle 후 파티션 수: spark.sql.shuffle.partitions (기본 200)

너무 적으면
→ 파티션 하나가 커서 OOM·Spill, 병렬성 부족

너무 많으면
→ Task 스케줄링 오버헤드, 작은 파일 양산

• AQE (Adaptive Query Execution): 실행 중 실제 통계로 계획을 수정 (Spark 3.2부터 기본 활성화)

Static Plan
→ 실행 전 추정 통계로 확정

AQE
→ Stage가 끝날 때마다 실제 통계로 다음 Stage 계획을 재조정

• AQE가 하는 일 세 가지
  1. Shuffle 파티션 자동 병합: 작은 파티션들을 합쳐 Task 수 감소 — shuffle.partitions를 정밀 튜닝할 필요가 줄어듦
  2. Skew Join 자동 분할: 비정상적으로 큰 파티션을 쪼개서 여러 Task로 처리 (5.4)
  3. Join 전략 런타임 변경: 실제 크기가 작으면 SortMerge → Broadcast로 전환 (5.3)