📒 Spark(16)

📌 Driver & Executor

Driver의 역할

• Spark Application = (1 Driver) + (1 + Executor)
• Driver는 다음 역할을 수행한다.
  • main 함수를 실행하고 SparkSession/SparkContext를 생성한다.
  • 코드를 Task로 변환하여 DAG를 생성한다.
  • 이를 execution/logical/physical plan으로 변환한다.
  • 리소스 매니저의 도움을 받아 Task들을 실행하고 관리한다.
    -> Task의 수가 너무 많아지면 Driver 메모리 에러가 발생한다.
  • 위 정보들을 모두 Web UI로 표현한다.(Port : 4040)

Driver 메모리 구성

spark.driver.memory = 4GB
spark.driver.cores = 4
spark.driver.memoryOverhead = 0.1
# max : spark.driver.memory의 10% -> 384MB

Executor 메모리 구성

spark.executor.memory = 8GB
spark.executor.cores = 4
spark.executor.memoryOverhead = 0.1
# max : spark.executor.memory의 10% -> 384MB
yarn.nodemanager.resource.memory-mb

Heap 메모리 구성

• Reserved Memory : 300MB로 Spark Engine 전용이다.(필수)
• Spark Memory : 데이터 프레임 관련 작업과 캐싱에 사용된다. (8000mb - 300mb) * 0.6
• User Memory : UDF, UDAF등등이 여기에 존재한다. (8000mb - 300mb) * 0.4

User Memory

• User-defined data structure

• Spark internal metadata

• UDF

• RDD conversion operation

• RDD lineage and dependency

• RDD와 관련된 작업을 직접 하는 경우 사용한다.

• 이 부분이 적다면 spark.memory.fraction을 증가시키면 된다.

Spark Memory

spark.executor.memory = 8GB
spark.executor.cores = 4
spark.memory.storageFraction = 0.5

Executor CPU

spark.executor.memory = 8GB
spark.executor.cores = 4

• 4 Task가 병렬로 실행 가능하다.
• Slot들은 사실 같은 JVM안의 Thread를 말한다.

Executor Resource

• 4개의 슬롯은 JVM 힙 메모리를 어떻게 공유하는가.
  

Executor Memory Pool Management

• Static Memory Management

  • Spark 1.6이전에는 슬롯들 끼리 공평하게 나누어 가졌다.

• 지금은 Unified Memory Manager를 사용한다.

  • 동작중인 Task 대상으로 Fair Allocation이 기본 동작이다.
    -> 즉 실행중인 Task가 모든 메모리를 가져가는 구조이다.

Unified Memory Management

• Executor 메모리가 부족해지기 시작하면 생기는 일

  • 메모리가 부족해지면 Storage Memory Pool에 남는 메모리를 사용한다.
  • spark.memory.storageFraction로 지정된 비율은 시작 비율이다.
    -> 하지만 양쪽의 메모리가 차기 시작하면 이 경계선은 지켜지면서 eviction이 발생한다.

• 반대로 Storage메모리가 부족해지면 생기는 일

  • DataFrame 캐싱을 하기 위한 메모리가 부족해지면 Executor Memory Pool에 남는 메모리를 사용한다.
  • 하지만 결국 spark.memory.storageFraction 이상은 넘어가지 않는다.

• 더 이상 쓸 메모리가 없다면?

  • 데이터를 메모리에서 디스크로 옮긴다.(Disk Spill)
  • Disk로 Spill을 할 수 없다면 OOM(Out of Memory)가 발생한다.

Spark Executor Memory Configuration

이름	Default	Explain
spark.driver.memory	1G
spark.driver.cores	1
spark.executor.memoryOverhead	executorMemory * spark.executor.memoryOverheadFactor와 384중 더 큰 값(MB)	Non-JVM 작업에 부여되는 메모리
spark.executor.memory	1G	JVM에 부여되는 메모리
spark.memory.fraction	0.6	Spark 메모리 : JVM에 부여되는 메모리 중 데이터프레임 관련 메모리 비율(캐실 포함)
spark.memory.storageFraction	0.5	Spark 메모리 중 캐싱에 사용되는 메모리의 비율
spark.executor.cores	1(YARN)	보통 executor당 1~5개의 CPU를 할당하는 것이 일반적이다.
spark.memory.offHeap.enabled	Fasle	Non-JVM 작업에 부여되는 별도 메모리 활성화 여부
spark.memory.offHeap.size	0	offHeap에 부여되는 메모리
spark.executor.pyspark.memory	없음	Python 프로세스에 지정되는 메모리, 지정되지 않으면 Non-JVM메모리(오버헤드 메모리)를 사용한다. 지정된다면 이 값만을 사용한다.
spark.python.worker.memory	512	Py4J에 지정되는 메모리(MB)

• 하나의 Executor에 할당되는 메모리
  -> spark.executor.memoryOverhead + spark.executor.memory + spark.memory.offHeap.size + spark.executor.pyspark.memory (+ spark.python.worker.memory)

Off Heap Memory

• Spark은 On Heap 메모리에서 가장 잘 작동한다.
• 하지만 JVM Heap은 Garbage collection의 대상이 된다.
• JVM Heap의 크기가 클수록 Garbe collection으로 인한 비용이 증가한다.
• 이때 같이 사용할 수 있는 것이 JVM 밖에 있는 메모리이다.
  • Overhead 메모리
  • Off Heap 메모리
    • spark.memory.offHeap.enabled를 True로 설정
    • spark.memory.offHeap.size에 원하는 크기를 지정

Spark 3.x의 Off heap memory 정리

• Spark 3.x는 Off Heap Memory 작업에 최적화 되어있다.
  -> JVM 없이 직접 메모리 관리 가능하다.

• Spark 3.x는 Off Heap 메모리를 DataFrame용으로 사용한다.
  -> GC의 발생을 줄일 수 있다.

• 즉 Off Heap 메모리의 크기는 아래와 같다.

  • spark.executor.memoryOverhead + spark.offHeap.size
  • spark.offHeap.size를 사용해서 executor memory의 증가 없이 off heap 메모리를 증가 시키는 것이 가능하다.