[Spark] Spark tuning, job에 따른 최적의 Partition 크기, 개수 조정하기, core의 역할 정리

이번 포스트에는 Partition 수 조정에 관한 spark tuning을 알아 보겠습니다.

 

Partition 개념에 대해 잘 모르시다면 아래의 링크를 참조해주세요.

https://spidyweb.tistory.com/312

[Spark Tuning] Spark의 Partition 개념, spark.sql.shuffle.partitions, coalesce() vs repartition(), partitionBy()와의 차이

Partition 세팅 전에 고려해보아야 할 Executor와 core 수 memory 양 세팅에 관한 포스트는 아래의 링크를 참조해주세요.

2021.12.20 - [BigData/Spark & Spark Tuning] - [Spark Tuning] (spark on YARN)yarn container, spark core, executor 개수 Memory 용량 계산법 및 최적화

[Spark Tuning] (spark on YARN)yarn container, spark core, executor 개수 Memory 용량 계산법 및 최적화

---

Partition의 내용을 이해하기 위해 Shuffle과 memory spill, disk spill에 대해 알아 보겠습니다.

---

Spark에서 Shuffle이 일어나는 이유, 언제 일어나는 지

1. Shuffle이 일어나는 이유

• 파티션 사이에서 데이터가 재배치 되어야 할 때
• 특정 transforamtion 실행 시 다른 파티션에서 정보가 필요하기 때문에 셔플을 함으로써 정보를 찾음

---

2. shuffle은 언제 일어나는 지

• join, union, groupBy, sort, coalesce, repartition, aggregate 과 같이 파티션과 관련된 조작이 있을 때
• 즉, MapReduce의  reduce()에 해당하는 operation을 실행하기전에 물리적인 데이터의 이동이 있어야 하기 때문에 shuffle이 일어난다.
• ex) 컬럼의 모든 값을 더할 때

---

shuffle spill(memory)란? shuffle spill(disk)란?

1. shuffle spill이란?

데이터를 직렬화하고 스토리지에 저장, 처리이후에 역직렬화하여 연산 재개하는 행위

data를 shuffle할 때 자원이 충분하지 못할 때 발생

→ partition size는 크지만 연산에 쓰이는 memory가 부족할 때

---

shuffle spill(Memory)

• 의미:
  • spill될 때 메모리에서 역직렬화된 형태의 데이터 크기 
  • spark 내부 메모리에서 한 작업이 너무 많은 메모리를 차지해 다른 작업으로 일부 데이터를 밀어내는 현상
• 원리:
  • Spark Executor는 Execution Memory와 Storage Memory로 나뉘어 메모리를 사용
  • 작업 중 메모리 사용이 증가하면, Spark는 Execution Memory 내에서 처리하려고 시도
  • 하지만 이 메모리가 부족하면 일부 데이터를 Storage Memory 또는 GC(Garbage Collection)를 통해 밀어내려고 함
• 발생 상황:
  • 작업 메모리가 제한된 크기를 초과할 때 발생(작업에서 생성되는 중간 데이터가 execution memory를 초과(spark.executor.memory))
  • 데이터셋이 매우 크고, map, flatMap, filter 같은 Transformation 연산이 많을 때
  • Caching 또는 Broadcast 메모리가 많아 Storage Memory가 부족할 때
  • 작업 중 Execution Memory와 Storage Memory 간의 충돌로 인해 일부 데이터가 Spill

---

shuffle spill(Disk)

• 의미:
  • spill된 이후 디스크에서 직렬화된 형태의 데이터 크기
  • Spark 작업 중 메모리가 부족하여 데이터를 디스크에 임시로 기록하는 현상
• 원리:
  • 작업이 메모리를 초과하여 데이터를 더 이상 메모리 내에서 처리할 수 없는 경우, 데이터를 디스크로 기록(중간 데이터가 메모리에 저장될 수 없을 만큼 크면 디스크로 Spill이 발생)
    • Shuffle 작업 중 중간 데이터가 생성되고 파티션별로 정렬 및 병합이 필요
    • 이 데이터를 모두 메모리에 저장할 수 없으면 디스크로 Spill
  • 특히 Shuffle 작업에서 대규모 데이터를 처리할 때 발생 가능
• 발생 상황:
  • groupByKey, reduceByKey, join 같은 연산 중, 중간 데이터가 Executor 메모리 한계를 초과할 때
  • Shuffle Read/Write 크기가 클 때
  • 전체 메모리가 적게 설정되었거나 데이터셋이 비정상적으로 클 때

---

•  

---

2. shuffle spill이 성능에 미치는 영향

Task가 지연되고, 에러가 발생

Hadoop 클러스터의 사용률이 높다면, 연달아 에러가 발생되고 최악의 경우에는 Spark가 강제종료

---

3. Shuffle spill을 방지 하는 방법

• 쿼리 최적화(skew 현상 제거)
  • 필요한 쿼리만select or wide transformation 전에 조건 연산자 추가
• Partition 수 증대 -> partition 크기 감소(task 당 필요한 memory 감소)
• core당 메모리 양 증가
  • spark.executor.memory: Executor에 할당된 메모리 크기를 늘림.
  • spark.memory.fraction: 전체 메모리에서 Execution Memory와 Storage Memory 비율을 조정.
  • spark.memory.storageFraction: Storage Memory 비율을 조정해 메모리 부족 문제 완화.

---

Core 의 역할

spark job을 처리하는 자원 중 memory와 core 중 core의 역할 정리

1) Task 실행의 기본 단위

• Spark 작업은 Task 단위로 실행
• Core는 Task를 실행할 수 있는 CPU 리소스를 제공
• 즉, 1 Task는 1 Core를 사용하여 실행 됨

2) 병렬 처리

• Core는 병렬 처리를 가능하게 함
  예를 들어, 8 Core가 있다면 최대 8개의 Task를 병렬로 실행할 수 있음
  이로 인해 데이터 처리 속도가 향상

3) 작업 스케줄링

• Core는 Spark 클러스터의 리소스를 스케줄링할 때 기준으로 사용
• Executor는 Core 수에 따라 한 번에 실행할 수 있는 Task 수를 결정

4) 파티션과의 관계

• 데이터는 Partition 단위로 나뉘며, 각 Partition은 하나의 Task로 처리
• 1core = 1task = 1partition으로 어림 잡으면 됨

5) 적정 갯수

• executor 당 3~5코어정도를 권장
• 위의 범위를 벗어나서 많은 코어를 할당 할 경우 아래와 같은 문제 발생
  1. Core가 많아지면 여러 Task가 Executor의 CPU를 공유하면서 컨텍스트 스위칭이 발생
     • 컨텍스트 스위칭: CPU가 한 Task에서 다른 Task로 전환할 때 발생하는 오버헤드
     • Core 수가 많을수록 Task 간 전환 비용이 누적되어 성능이 저하 
  2. 네트워크 I/O 병목
     • Core 수가 많아지면 많은 Task가 동시에 데이터를 읽거나 쓸 수 있음
     • 그러나 Executor는 네트워크 대역폭을 공유하므로, 동시에 많은 Task가 실행되면 네트워크 I/O 병목이 발생

 

---

Shuffle Partition 수 최적화

관련 config: Spark.sql.shuffle.partitions

• Spark 성능에 가장 크게 영향을 미치는 Partition으로, Join,groupBy 등의 연산을 수행할 때 Shuffle Partition이 쓰임
• join, groupBy와 같은 연산을 수행 시 Partition의 수(Task 수)가 결정

Shuffle Partition은 일반적으로 core 수 = partition 수 라고는 되어 있지만,

1. core의 2~3배 숫자를 지정하는 것이 적합,

2. shuffle Partition 1개당 size를 100Mb ~ 200Mb 사이의 크기로 맞추어 Partition 숫자를 산정

---

Partition 크기를 아는 방법

Partition의 크기는 shuffle read size와 Partition 수에 따라서 결정

 

*shuffle read size:
모든 executor에 직렬화된 읽기 데이터의 총합

---

Partition 크기 및 개수 최적화 예시

1) shuffle read size 240GB라 가정 partition 수가 300일 시, Partition당 크기는 800MB(partition당 읽어 들이는 양)

 

2) spill(memory) size가 없다면, 설정한 partition 수로 자원의 부족함 없이 task가 실행되고 있다고 봄

 

3) spill(memory) size가 있다면, 예를들면 spill(memory) size가 840GB라 가정하면 Partition당 spill(memory) size는 약 2.8GB~이므로 1core당 2GB 메모리의 자원으로는 수행이 불가

 

4) Partition당 크기를 160MB가 되도록 설정한다면, 800MB/5 = 160MB ,즉 Partition의 수는 기존 300개 보다 5배 많은 300 * 5 = 1500으로 증가

300개의 partition수 일 때 일어났던 840GB의 spill(memory) size가 Partition당 840G/1500 =약 0.5xGB~정도로 1core당 2GB는 물론, 심지어 놀고 있는 메모리가 생겨 1core당 1GB를 사용해도 가능

 

메모리 설정 팁

• shuffle read size + shuffle write size < executor 수 * executor당 memory 양
• Shuffle Size가 600GB에 가깝거나 그 이상일 경우, Core당 메모리를 증가시키는 것이 좋음. 또한 1코어당 4GB를 고려하기를 권장

---

결론:

1. 총 코어의 개수보다 적은 수의 파티션 수는 일부 resource가 놀게 되므로 비효율적이다.
2. 일반적으로 파티션수를 늘리는 것은 오버헤드가 너무 많아지기 전까지는 성능을 높인다(2~3배가 적당)
3. 파티션을 늘리게 되면 파티션의 크기가 작아지고 각 executor에서 spark가 처리하는 partition(task)의 양이 적어지므로 메모리 부족 오류를 줄이게 된다.
4. 메모리 부족 오류 및 spill size가 줄어들면 지연이 사라지고 성능이 올라간다.
5. 너무 많은 파티션은 driver memory error, driver overhead error를 유발하며, 작은 사이즈의 파일들을 생성하기위한 I/O도 많이 발생한다(특히 block store)

---

*큰 파일을 가지고 작업하거나 그에 맞는 EMR을 혹은 하둡클러스터를 구축할 만한 실습환경이 없어서 이후에 더 좋은 자료가 생기거나, 관련 실습,테스트를 할 상황이 생기면 추가적으로 수정하여 포스트하겠습니다.

               

참조:

https://jaemunbro.medium.com/apache-spark-partition-%EA%B0%9C%EC%88%98%EC%99%80-%ED%81%AC%EA%B8%B0-%EC%A0%95%ED%95%98%EA%B8%B0-3a790bd4675d

https://tech.kakao.com/2021/10/08/spark-shuffle-partition/

https://developer.ibm.com/blogs/spark-performance-optimization-guidelines/