Apache Spark Deep Dive: Memory

Apache Spark의 메모리 할당은 Spark 작업(Job)의 성능을 결정짓는 핵심적인 요소입니다. Spark의 메모리는 연산(Computations), 캐싱(Caching), 그리고 시스템 오버헤드(System Overhead)를 처리하기 위해 여러 구성 요소로 나뉩니다.

• Driver Memory:
  태스크 스케줄링(Task Scheduling) 및 Spark 작업 전반의 관리를 담당하는 드라이버 프로그램에 할당되는 메모리입니다.
  Driver는 SparkSession/SparkContext를 생성하고, Job을 submit하고,
  Worker 간의 Task 실행을 조정하는 주요 프로세스입니다.
• Executor Memory:
  워커 노드(Worker Node)에서 실행되는 각 Executor에 할당되는 메모리입니다. 실제 데이터 처리와 저장(캐싱)이 일어나는 공간입니다.

Executor 메모리 아키텍처

Spark Executor(JVM)가 실행될 때, 메모리는 크게 JVM Heap과 Off-Heap(JVM 외부), 그리고 Overhead 영역으로 나뉩니다.

JVM Heap Memory (Inside of JVM)

Spark 1.6 이후 도입된 Unified Memory Manager는 메모리를 정적으로 나누지 않고, 필요에 따라 유연하게 경계를 조정합니다. Executor 컨테이너 내부의 메모리는 크게 세 가지 영역으로 나뉩니다.

Spark 1.6.0부터 통합 메모리 관리자가 Spark의 기본 메모리 관리자로 설정되었으며, Spark 1.6 이상에서는 spark.memory.useLegacyMode=true 매개변수 를 통해 정적 메모리 관리를 활성화할 수 있습니다.

1. Reserved Memory (Hardcoded to 300MB as of spark 1.6)

Spark 1.6 이후 300MB로 하드코딩 되어 있습니다.
이 공간은 사용자나 데이터 처리에 사용되지 않습니다.

힙 크기가 이 예약된 메모리의 1.5배(450MB)보다 작으면 Spark는 시작조차 되지 않고 종료됩니다.

2. User Memory

Spark가 직접 관리하지 않는 영역입니다. 사용자가 코드에서 정의한 커스텀 객체(Custom Objects), UDF, 전역 변수(Broadcast Variables 등)가 저장됩니다.

• 공식: (Java Heap — 300MB) X (1 — spark.memory.fraction)
• 기본값: 전체 힙의 약 40%

이 영역을 너무 줄이면 UDF 실행 중 OOM이 발생할 수 있습니다.

3. Spark Memory (Unified Region)

실제 Spark의 핵심 활동이 일어나는 공간으로, 다시 두 개의 하위 영역으로 나뉘어 상호 공유됩니다.
기본값: spark.memory.fraction (기본 0.6, 즉 60%)

• Execution Memory:
  Join, Shuffle, Aggregation, Sort 등 실제 연산(Computation)을 수행하는 공간입니다. Shuffle 시 발생하는 임시 버퍼(Buffer)들이 이곳에 할당됩니다.
• Storage Memory:
  cache(), persist() 된 RDD나 DataFrame이 저장되는 공간입니다. Shuffle 데이터나 RDD의 혈통(Lineage) 정보를 저장하여 내결함성(Fault Tolerance)을 보장합니다.

Off-Heap Memory (Outside of JVM)

Spark는 GC(Garbage Collection) 오버헤드를 피하기 위해 JVM 힙 외부의 메모리를 적극적으로 활용합니다.

• 설정: spark.memory.offHeap.enabled=true 및 spark.memory.offHeap.size로 활성화.
• Tungsten Engine: Spark의 최적화 엔진인 Tungsten은 데이터를 Java 객체가 아닌 바이너리(Binary) 형태로 변환하여 Off-Heap에 직접 저장합니다. 이는 GC의 간섭을 받지 않으며, C++ 수준의 메모리 관리 효율을 제공합니다.

Off-Heap 메모리는 기본적으로 Serialized Data Storage(직렬화된 데이터 저장)나 네트워크 전송 버퍼(Network Buffer)로 활용되지만, spark.memory.offHeap.enabled=true 옵션을 활성화하면 Shuffle & Sorting 단계의 실행 버퍼(Execution Buffer)까지 Off-Heap 영역에 할당되어 GC 오버헤드를 제거할 수 있습니다.

Unified Memory Manager의 작동 원리

Unified Memory 아키텍처에서 Execution과 Storage 영역은 고정된 파티션(Hard Partition) 없이 단일 메모리 풀을 공유하며, 런타임 워크로드의 요구량에 따라 서로의 유휴 자원을 유연하게 차용하는 동적 점유(Dynamic Occupancy) 메커니즘을 따릅니다.

동적 점유 정책 (Dynamic Occupancy Policy)

Unified Memory Manager는 Execution과 Storage 영역 간의 상호 대여(Mutual Borrowing) 및 선점(Preemption) 메커니즘을 통해 메모리 효율을 극대화합니다

• 동적 차용 (Dynamic Borrowing):
  Execution 영역에 유휴 공간이 존재할 경우 Storage가 이를 차용하여 캐싱에 활용하며, 반대의 경우 Execution이 Storage의 유휴 공간을 점유하여 연산에 활용합니다.
• Execution의 선점 우위 (Preemptive Eviction):
  - 연산(Execution) 수행 중 메모리가 부족할 경우, Storage가 차용 중인 메모리 블록을 즉시 축출(Evict)하여 공간을 확보합니다.

캐시된 데이터의 손실은 재계산 비용을 발생시킬 뿐이지만, 연산 중인 데이터의 손실은 Task 실패(Failure)로 직결되기 때문입니다.
따라서 Spark는 연산의 안정성을 최우선으로 보장합니다.

• Storage의 면역 임계값 (Immunity Threshold):
  spark.memory.storageFraction(기본값 0.5)으로 설정된 영역은 면역 지대(Immune Zone)로 간주됩니다. 이 임계값 내에 저장된 캐시 데이터는 Execution 메모리가 부족하더라도 강제로 축출되지 않습니다

Disk Spill의 발생 메커니즘과 비용

Task 실행 시 각 스레드(Active Slot)는 파티션 처리를 위해 Execution Memory를 점유합니다. 가용 메모리가 임계치에 도달하면 Spill(디스크 유출) 메커니즘이 트리거됩니다

Spill 처리 프로세스 (Step-by-Step)

Execution Memory가 500MB이고 실행 중인 코어가 2개(N = 2)라면,
각 태스크는 최소 125MB(500 / 4(2N))에서 최대 250MB(500 / 2) 사이의 메모리 내에서만 동작해야 합니다.

최소한의 메모리를 확보하지 못하면 해당 태스크는 대기하거나 acquireMemory()를 통해 공간이 날 때까지 기다립니다.

1. In-Memory Buffer 할당:
   SortMergeJoin 등의 연산 수행 시, Spark는 ExternalAppendOnlyMap 혹은 ExternalSorter와 같은 인메모리 페이지(Page) 버퍼에 데이터를 적재합니다.
2. 리소스 요청 및 부족 감지:
   버퍼가 포화 상태에 이르면 MemoryManager에 추가 페이지 할당을 요청(acquireMemory())합니다.
3. Spill 트리거 (Spill Trigger):
   Unified Memory 내의 가용 공간이 고갈되고, Storage로부터 축출할 수 있는 공간조차 없을 경우 Spill이 발동됩니다.
4. 직렬화 및 디스크 쓰기 (Serialization & I/O Overhead):
   메모리에 상주하던 데이터를 정렬(Sort) 후 직렬화(Serialize)하여 로컬 디스크의 임시 디렉토리(spark.local.dir)에 기록합니다.
   - 성능 영향: 이 과정에서 직렬화로 인한 CPU 부하와 Disk I/O Latency가 동시에 발생하여 처리 속도가 급격히 저하됩니다.
5. 병합 (Merge Sort):
   Task 완료 시점에 디스크에 분산된 Spill 파일들과 메모리의 잔존 데이터를 다시 읽어들이며 병합 정렬(Merge Sort)을 수행합니다.

Spill은 OOM(Out Of Memory)을 방지하기 위한 방어 기제이나, 과도한 발생은 심각한 병목(Bottleneck)이 발생할 수 있습니다.
Spark UI의 Stage 탭에서 Spill (Memory) 및 Spill (Disk) 지표를 모니터링할 수 있습니다.

OOM 분석 및 성능 최적화

Case 1: Executor OOM

java.lang.OutOfMemoryError: Java heap space

• 원인 A (Data Skew): 특정 파티션 키(Key)에 데이터가 편중되어, 단일 Task가 처리해야 할 데이터 양이 Executor의 힙 메모리 한계를 초과함
  → Key Salting 기법 적용 또는 repartition()을 통한 데이터 재분배.
• 원인 B (High Memory Footprint): User Memory 영역에 고비용 객체(대용량 Map, List 등)가 다수 생성됨
  → 어플리케이션 로직 최적화 또는 spark.memory.fraction 하향 조정을 통한 User Memory 공간 확보

Case 2: Container killed by YARN for exceeding memory limits

• 원인: JVM Heap 외적으로 사용되는 Off-Heap Overhead(Netty Buffer, Thread Stack 등)가 컨테이너 설정 한도를 초과함
  → spark.executor.memoryOverhead 파라미터 상향 조정

기준: Max(spark.executor.memory * 0.10, 384MB)

Get Seonggil Jeong’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

conf spark.executor.memoryOverhead=2048 등으로 여유 공간 확보.

Case 3: Driver OOM

• 원인: collect() 호출로 인해 Executor의 대용량 결과 데이터가 Driver로 일시에 전송됨
  → collect() 대신 toLocalIterator()를 사용하여 데이터 스트리밍 처리.
  → spark.driver.maxResultSize 설정 변경(기본 1GB)
  → Driver 메모리 리소스 증설 (spark.driver.memory)

Case 4: 과도한 Spill로 인한 Latency 증가

• 원인: 데이터 규모 대비 파티션 수가 부족하여, 파티션 당 크기가 Executor 메모리 용량을 상회함.
  → spark.sql.shuffle.partitions (기본 200) 값을 상향 조정하여 파티션 당 데이터 크기를 적정 수준(128MB~200MB)으로 유지

Shuffle 발생 시 메모리 점유 및 할당 메커니즘

셔플링은 크게 Shuffle Write (Map Side)와 Shuffle Read (Reduce Side) 두 단계로 나뉩니다. 이 과정에서 Spark는 Execution Memory 영역을 사용하며, Tungsten 엔진의 페이지 기반 관리가 본격적으로 작동합니다.

1. Map Side: Shuffle Write 단계의 메모리 할당

Map Task는 데이터를 처리한 후, 다음 스테이지의 Reducer가 가져갈 수 있도록 파티션별로 데이터를 분류하고 정렬해야 합니다. 이 과정에서 In-Memory Buffer가 핵심적인 역할을 합니다.

1–1. 데이터 구조: PartitionedAppendOnlyMap & PartitionedPairBuffer

Spark는 들어오는 레코드(Record)를 즉시 디스크에 쓰지 않고, 먼저 메모리 버퍼에 채웁니다. 연산 종류에 따라 두 가지 자료구조 중 하나를 할당받습니다.

• PartitionedAppendOnlyMap:
  reduceByKey나 aggregateByKey처럼 Aggregation(집계)이 필요한 경우 사용합니다. 해시 테이블과 유사하게 Key가 같으면 값을 갱신(Update)하며 메모리를 효율적으로 사용합니다.
• PartitionedPairBuffer:
  sortByKey나 groupByKey처럼 단순 정렬이나 그룹핑만 필요할 때 사용합니다. 단순히 데이터를 차곡차곡 쌓는 배열(Array) 형태입니다.

1–2. 메모리 점유 방식 (Pointer + Page)

이 버퍼들은 단순히 자바 객체를 저장하는 것이 아니라, Tungsten 방식으로 메모리를 점유합니다.

1. Pointer Array (포인터 배열):
   - 데이터의 주소값(Pointer)과 파티션 ID만을 저장하는 Long 타입 배열을 할당합니다.
   - 정렬(Sorting)이 필요할 때 실제 무거운 데이터를 옮기지 않고, 이 **가벼운 포인터 배열만 퀵 소트(Quick Sort)**하여 속도를 높입니다.
2. Data Pages (데이터 페이지):
   - 실제 Key와 Value 데이터는 OS의 Page 단위(기본 64MB 등)로 할당된 메모리 블록에 바이너리 형태로 연속해서 기록됩니다.
   - 이 방식은 메모리 단편화(Fragmentation)를 막고 CPU 캐시 적중률(Cache Locality)을 높입니다.

1-3. 할당과 Spill의 임계점

데이터가 계속 들어오면 메모리 매니저는 다음과 같은 과정을 반복합니다.

1. Check: 현재 버퍼(Pointer Array + Data Pages)의 크기가 임계치를 넘었는지 확인합니다.
2. Request: UnifiedMemoryManager에게 Execution Memory 추가 할당을 요청(acquireMemory())합니다.
3. Spill: 요청이 거절되거나(메모리 부족), 컬렉션 내부의 포인터 배열이 한계에 도달하면 Spill이 트리거됩니다.
4. 메모리에 있는 데이터를 파티션 ID 기준으로 정렬하여 디스크에 쓰고, 메모리를 비웁니다(Free).

2. Reduce Side: Shuffle Read 단계의 메모리 할당

Reducer는 여러 Mapper로부터 자신에게 할당된 파티션의 데이터 블록을 네트워크를 통해 가져옵니다.

2–1. Fetch & Buffer (가져오기 및 버퍼링)

• ExternalAppendOnlyMap: Reduce 단계에서도 데이터 집계나 정렬이 필요하다면 Map Side와 유사한 ExternalAppendOnlyMap을 할당합니다.
• 네트워크 버퍼: 네트워크를 통해 들어오는 데이터는 일시적으로 네트워크 버퍼(Netty Buffer, Off-Heap)를 거쳐 Executor의 Heap(또는 Off-Heap) Execution Memory로 복사됩니다.

2–2. Merge Sort (병합 정렬)와 메모리

Spill 없이 모든 데이터가 메모리에 들어올 수 있으면 좋지만, 대용량 처리 시에는 Memory-Disk 하이브리드 방식이 작동합니다.

1. In-Memory Merge:
   Mapper에서 가져온 데이터 블록들을 메모리 맵(Map)에 넣으면서 실시간으로 집계(Merge)를 수행합니다.
2. Spill to Disk:
   Map Side와 마찬가지로 Execution Memory가 꽉 차면, 현재까지 집계된 맵의 내용을 디스크로 Spill 합니다
3. Global Merge:
   최종적으로 디스크에 Spill된 여러 파일들과, 현재 메모리에 남아있는 데이터를 하나로 합칩니다.