[Spark] 스파크의 Executor Memory 구조

참고: https://community.cloudera.com/t5/Community-Articles/Spark-Memory-Management/ta-p/317794

 

Spark Executor Memory 구조

https://medium.com/analytics-vidhya/spark-memory-management-583a16c1253f

Executor Container의 메모리 구조는 이미지와 같이 크게

• Memory Inside of JVM: spark.executor.memory
• Memory Outside of JVM: spark.yarn.executor.MemroyOverHead

로 나뉜다.

Object의 읽기/쓰기 속도는 `on-heap > off-heap > disk` 순서로 빠르다.

Memory Inside of JVM (=On-Heap Memory = In-Memory)

java GC 관리 하에 있는 메모리 영역이다. 어플리케이션 대부분에 데이터가 on-heap 메모리에 저장된다.

`spark.executor.memory`는 스파크의 설정 옵션 중 하나로, executor 에 할당되는 메모리를 설정하는데에 사용되는 conf값이다. 메모리 값이 너무 작으면 작업이 충분한 메모리를 사용하지 못해 성능 저하 또는 OOM이 발생할 수 있고, 너무 크면 클러스터의 메모리 사용량이 높아져 다른 어플리케이션에 영향을 줄 수 있다.

https://community.cloudera.com/t5/Community-Articles/Spark-Memory-Management/ta-p/317794

on-heap 메모리에서는 세 가지 메모리 영역이 있는데 위 이미지와 같다.

Spark Memory

스파크에서 관리하는 메모리 영역으로, 스파크가 인메모리 컴퓨팅을 할 때 중간 상태를 spark memory에 저장한다.

(Java Heap — Reserved Memory) * spark.memory.fraction

첫 번째 이미지 사진에서 보면 spark memory 영역이 두 개로 나뉘어 있는 것을 볼 수 있다.

• `spark.memory.fraction - spark.memory.storageFraction`: Execution
• `spark.memory.storageFraction`: Storage

fraction 영역이 셔플, 조인, 정렬, 집계 등 연산 작업에 사용되고, storageFraction 영역으로 표시된 부분이 데이터 파티션을 캐시하는 데에 사용된다.

이 영역은 `spark.memory.stroageFraction` 값으로 조정할 수 있으며 기본값은 0.5 (50%) 이다.

a. Storage Memory

storage memory는 캐시 데이터, broadcast variable 등을 저장하는데에 사용된다. serialized data -> deserialized 로 변환하는데에 사용되는 프로세스인 " unroll" 도 storage memory에 올라가있다. 

(Java Heap — Reserved Memory) * spark.memory.fraction * spark.memory.storageFraction

b. Execution Memory

스파크 작업을 실행하는 동안 필요한 object를 저장하는데에 사용된다.

셔플, 조인, 소트, 집계 등 연산 시 중간 버퍼(임시 데이터)를 저장하거나 해시 테이블을 execution memory 에 저장한다. 

storage memory보다 수명이 짧아서 task가 끝나면 바로 비워지고 다음 task 작업을 위해서 공간을 만들어준다.

(Java Heap — Reserved Memory) * spark.memory.fraction * (1.0 - spark.memory.storageFraction)

 

 

User Memory

user memory는 user defined data structure,  스파크 내부 메타데이터, UDF, RDD 변환작업에 필요한 데이터(RDD 종속성 등)를 저장하는데에 사용되는 메모리이다.

(Java Heap — Reserved Memory) * (1.0 — spark.memory.fraction)

 

Reserved Memory

시스템용으로 예약된 메모리 영역으로, Spark의 내부 object를 저장하는데에 사용된다. 

300MB로 하드코딩 되어 있으며, 관련 값을 바꾸고 스파크를 다시 컴파일해서 변경된 값을 사용할 수는 있지만 운영 환경에서는 추천하지 않는다고 한다.

// https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala#L198
val RESERVED_SYSTEM_MEMORY_BYTES = 300 * 1024 * 1024

 

추가로 `spark.executor.memory` < `Reserved Memory * 1.5 ` 인 경우 session을 초기화하지 못하고 실패된다.

 

 

Memory Outside of JVM (=Off-Heap Memory = External-Memory)

* 참고

스파크 2.2 이하: `spark.yarn.executor.memoryOverhead`

스파크 2.3 이상: `spark.executor.memoryOverhead`

cluster manager가 yarn 뿐만 아니라 kubernetes 등 다양하게 있기 때문에 yarndp 종속된듯한 설정명을 일반적 설정명으로 변경했다고 함

 

리소스 매니저 관리 하의 메모리

JVM GC 관리 범위 밖에 있는 메모리 영역이다.

기본적으로 executor memory를 설정하면 memoryOverhead 크기는 아래와 같이 설정된다.

MAX(spark.executor.memory*0.1, 384MB)

 

`spark.executor.memoryOverhead` 값 지정을 통해 직접 설정해줄 수도 있다.

 

On-Heap과 Off-Heap

만약 executor-memory=5g로 설정했다면

• on-heap: 5G
• off-heap: min(5G*0.1, 384MB) = 500MB

총 5GB + 500MB를 할당받게 된다.

 

메모리가 부족하다는 메시지가 있을 때 어떤 메모리를 늘려야 할지 고민이 될 때가 있다.

아래의 경우에는 executor-memory를 늘려야 한다.

• GC가 자주 발생하는 경우: on heap 메모리가 부족해서 가비지 컬렉션이 자주 발생한다는 뜻이므로 on heap 메모리를 늘려야한다.

반대로 아래의 경우에는 memoryOverhead를 늘려야 한다.

• GC는 자주 발생하지 않지만 cluster manager에 의해 executor 가 자꾸 죽을 경우
  

  Reason: Container killed by YARN for exceeding memory limits.
  5.5 GB of 5.5 GB physical memory used.
  Consider boosting spark.yarn.executor.memoryOverhead.​

Spark Core 관리 하의 메모리

Off-heap 메모리 관련 설정 값이 아래와 같이 추가로 더 있다.

• spark.memory.offHeap.enabled: 오프힙 메모리 사용 여부 (디폴트는 false)
• spark.memory.offHeap.size: 오프 힙 사이즈 (디폴트 0)

spark.yarn.executor.memoryOverhead 와의 차이점은 아래와 같다:

• spark.yarn.executor.memoryOverhead: yarn과 같은 리소스 매니저에 의해 관리됨
• spark.memory.offHeap.size: spark core에 의해 관리됨 (tungsten 엔진)

spark.yarn.executor.memoryOverhead 와 spark.memory.offHeap.size 의 관계는 다음과 같다:

• spark 1.x and 2.x: 전체 Off-heap 메모리 = spark.yarn.executor.memoryOverhead
   spark.yarn.executor.memoryOverhead 가 spark.memory.offHeap.size를 포함함
• spark 3.x: 전체 Off-heap 메모리 = spark.yarn.executor.memoryOverhead + spark.memory.offHeap.size

참고: https://stackoverflow.com/questions/58666517/difference-between-spark-yarn-executor-memoryoverhead-and-spark-memory-offhea

 

Spark의 Tungsten 엔진은 가비지 컬렉션 오버헤드를 최소화하는데 기여한다. GC 오버헤드를 어떻게 최소화한다는 것일까,,

일반적으로 JVM에서 GC 가 발생할 때 stop-the-world 라고 표현하는 어플리케이션 중단 현상이 일어난다. 이 과정은 어플리케이션 실행을 지연시키고 전체 실행 시간을 늘린다.

Tungsten에서는 (c의 malloc 같은) sun.misc.Unsafe 기반으로 off heap에 메모리를 할당할 수 있도록 한다. 

Off heap 메모리는 JVM 바깥에 존재하기 때문에 GC 관리 밖의 메모리를 사용하면 stop-the-world 를 최소화할 수 있다는 점에서 성능 향상을 볼 수 있을 것 같다.

단점은 아래와 같다.

• on-heap 메모리보다 읽고 쓰기가 느리다.
• 메모리를 직접 수동으로 관리해주어야 한다.

실행 중 GC 단계가 너무 길어지면 찾아볼만하다.