Apache Spark ( Spark 최적화 실습 )

간단한 Spark 최적화를 실습해보자!

---

1. Spark 서브밋 옵션 핵심 정리

스크립트를 사용해 Spark 애플리케이션 제출
→ 작업 상황에 따라 편하게 옵션을 조정하며 + 실행로그 까지 남기는 기본 템플릿

######### spark-submit.sh #########
#!/usr/bin/bash
JARS=""
JOB_NAME="SY_TSET"
SCRIPT="${1}"
MASTER="local[*]"

echo "Running script: $SCRIPT"

# 스파크 애플리케이션 제출
spark-submit --name ${JOB_NAME} --master ${MASTER} \
  --conf spark.eventLog.enabled=true \
  --conf spark.eventLog.dir=file:///work/jsy/meta_code_spark/spark-events \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.dynamicAllocation.executorIdleTimeout=2m \
  --conf spark.dynamicAllocation.minExecutors=1 \
  --conf spark.dynamicAllocation.maxExecutors=2 \
  --conf spark.dynamicAllocation.initialExecutors=1 \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=1G \
  --conf spark.shuffle.service.enabled=true \
  --conf spark.executor.memory=1G \
  --conf spark.driver.memory=1G \
  --conf spark.driver.maxResultSize=0 \
  --num-executors 1 \
  --executor-cores 1 \
  ${SCRIPT}
######### spark-submit.sh #########

• 사용법 : ./spark-submit.sh slow_submit.py 이런식으로 실행하고 옵션을 조정하며 튜닝 가능

---

1-1. Event Log ( 디버깅 / 튜닝용 )

--conf spark.eventLog.enabled=true
--conf spark.eventLog.dir=file:///work/jsy/meta_code_spark/spark-events

• 왜 쓰냐?
  • 실행 기록 저장
• 왜 중요?
  • Spark UI는 실행 끝나면 → 작업 정보 날아감
  • Event Log 있으면 → History Server 에서 다시 분석 가능

즉, 성능 비교 분석 or 튜닝 시 중요한 옵션

---

1-2. Dynamic Allocation ( 익스큐터 오토 스케일링 )

--conf spark.dynamicAllocation.enabled=true
--conf spark.dynamicAllocation.executorIdleTimeout=2m
--conf spark.dynamicAllocation.minExecutors=1
--conf spark.dynamicAllocation.maxExecutors=2
--conf spark.dynamicAllocation.initialExecutors=1

1. spark.dynamicAllocation.enabled=true
→ 동적 익스큐터 활성화

2. spark.dynamicAllocation.executorIdleTimeout=2m
→ 익스큐터 일 안 하면 죽임 ( 2분 )
짧으면 → 계속 재생성 ( 오버헤드 )
길면 → 리소스 낭비

3. spark.dynamicAllocation.minExecutors=1 | spark.dynamicAllocation.maxExecutors=2
→ 동적 익스큐터 최소 Or 최대 갯수

4. spark.dynamicAllocation.initialExecutors
→ 익스큐터 초기 갯수

 

[ Spark.shuffle.service.enabled=true ]

--conf spark.shuffle.service.enabled=true
→ 셔플 데이터를 저장하는 옵션

[ 옵션 OFF시 동작 ]
1. 익스큐터 A → 셔플 파일 생성 ( 로컬 DISK )
2. 익스큐터 A 죽음 → 파일도 같이 삭제
3. 다음 스테이지에서 "파일 어디감?" → 실패

[ 옵션 ON 하면 ]
1. 익스큐터 A → 셔플 파일 생성
2. 익스큐터 A 죽음 → External Shuffle Service는 살아있음 → 파일 계속 유지

> 주로 동적 익스큐터 옵션 ( Dynamic Allocation ) 사용 시 필수적으로 사용

• Dynamic Allocation은 혼자 못씀
• 왜??
  • 익스큐터 죽음 → 셔플 데이터 날아감
• 해결
  • External Shuffle Service가 대신 저장
• External Shuffle Service 위치
  • 각 워커 노드마다 1개
  • 노드 내부 여러 익스큐터는 → 같은 Shuffle Service 공유

---

1-3. 메모리 위치 튜닝

--conf spark.memory.offHeap.enabled=true
--conf spark.memory.offHeap.size=1G
→ 익스큐터 마다 off-heap 1기가 메모리 사용 활성화
→ 즉, 바이너리 배열을 네이티브 메모리 저장 허용 ( 기본은 jvm heap만 사용 )


[ 익스큐터 메모리 구조 ]
1) 기본
┌─────────────────┐
│   JVM Heap (8GB)  │ ← GC 영향 받음
└─────────────────┘

2) off-heap 적용
┌───────────────┐
│   Heap (8GB)  │ ← GC 영향 받음
├───────────────┤
│ Off-heap (2GB)│ ← GC 영향 없음
└───────────────┘
→ 스파크는 자동으로 ( jvm heap | off-heap ) 선택해서 사용

 

[ 왜 Jvm heap, Off heap 둘다 씀? ]

• jvm heap
  • 장점: 빠름 + JVM 최적화 가능
  • 단점: GC 퍼즈 발생 가능 ( 객체 많이 만들거나 + 데이터 큼 + 반복 연산 많을 때 )
• off-heap
  • 장점: GC 퍼즈 영향 없음 ( 큰 데이터 or 반복 연산 안정적 )
  • 단점: 직접 메모리 관리 비용 있음

즉, off-heap의 핵심음 GC를 줄이려 데이터 위치를 네이티브 메모리로 빼는 것

 

[ Off-heap은 언제 좋냐? ]

1. 객체가 많이 생성되는 연산
2. 메모리 압박이 클때 ( 큰 데이 )

• join | groupBy | 셔플 많음
• 데이터 큼

---

1-4. 익스큐터 메모리 튜닝

--conf spark.executor.memory=1G
--executor-cores 1
--num-executors 1
→ 익스큐터 1개
→ 메모리 1기가
→ 동시에 task 1개 처리

동적 익스큐터 옵션과 같이 사용시 →  Dynamic Allocation이 우선권 가짐

---

1-5. 드라이버 메모리 튜닝

--conf spark.driver.memory=1G
--conf spark.driver.maxResultSize=0
→ 드라이버 메모리 1기가
→ 드라이버로 가져오는 결과 크기 제한 ( 0 = 무제한 )

 

[ Driver 역할 ]

• DAG 생성 + Task 스케줄링 + 결과 수집

---

 

2. Spark 코드상 성능 최적화

우선, Spark는 Lazy Evaluation ( 지연 실행 )

df = spark.read.parquet(..)
df = df.filter(..)
df = df.groupBy(..).agg(..)
→ 여기까지는 아무것도 안 함

df.show() → 액션 수행 시
이때 한 번에 전체 DAG 실행

### 문제 상황 ###
df.show()
df.count()
df.first()
→ 같은 DAG 3번 실행됨
→ 파일 다시 읽음 + filter 다시 함 + groupBy 다시 함 ( 비효율.. )

---

2-1. cache() → "같은 계산 여러 번 쓸 때"

한 번 계산한 결과를 메모리에 저장해서 재사용

 

[ 동작 흐름 ]

df = spark.read.parquet(...)
df = df.filter(...).groupBy(...)

df.cache()        # ← 여기 중요

df.show()         # ← 이때 처음 계산 + 캐시 저장
df.count()        # ← 캐시에서 가져옴
df.first()        # ← 캐시에서 가져옴


### 포인트 ###
- cache()는 즉시 저장 아님 X
- 첫 액션 실행 시 캐싱됨

### 어디 저장? ###
1) DAG 실행 Start
2) 각 익스큐터가 자기 파티션 계산
3) 계산 결과를 메모리에 저장 ( 여기서 cache )
4) 결과를 드라이버로 가져와 show

즉, 각 익스큐터는 본인 파티션만 결과만 캐시함

 

[ 언제 쓰냐? ]

• 같은 데이터프레임 반복 사용시 → 액션 여러 번 호출할 때

 

[ 주의점 ]

• 메모리 부족하면 날아감 ( 재계산 함 )
• 너무 남발하면 메모리 터짐

---

2-2. persist() / StorageLevel → "cache의 상위 개념"

캐시를 어디에 저장할지 직접 선택 ( 메모리 or 디스크 or Off_Heap )

cache() = persist(MEMORY_ONLY)

 

[ 주요 옵션 ]

옵션	의미
MEMORY_ONLY	메모리에만 (빠름, 터지면 재계산)
MEMORY_AND_DISK	메모리 부족하면 디스크
MEMORY_ONLY_SER	직렬화해서 메모리 절약
DISK_ONLY	디스크만
OFF_HEAP	JVM 밖 메모리

• 유연한 캐시 저장 위치 적용할 때 사용하며
• 데이터 큼 → MEMORY_AND_DISK
• 메모리 부족 → MEMORY_ONLY_SER
• 안정성 필요 → DISK_ONLY

 

[ Cache vs Persist ]

cache	persist
간단	세밀 제어
MEMORY_ONLY 고정	전략 선택 가능

---

2-3. checkpoint → "DAG 끊기" ( 중요 )

DAG 계산 이력을 날리고 결과만 남김

[ 왜 CheckPoint 필요? ]
→ 스파크는 계속 연결됨 ( DAG )
df0 → df1 → df2.. → 반복하면?

 

[ DAG 길어지면 문제 ]

• 스파크는 연산이 길어질수록 계획 수립 시간 증가 + 최적화 비용 증가
• 재계산 비용 증가

 

[ checkpoint 효과 ]

df = df.checkpoint() # 체크포인트 생성
→ 결과 저장(disk / hdfs) + 이전 DAG 삭제 ( DAG 끊음 )

df0 → df10 → checkpoint
              ↓
           df10 (새 시작)

• 즉, checkpoint는 중간 결과를 저장하고 이전 구조를 리셋하는 기능

 

[ checkpoint 언제 쓰냐? ]

• 반복 알고리즘 ( ML학습 / Graph 처리 )
• 안정성 필요할 시 ( 익스큐터 장애 대비 )

하지만, 단순 ETL 또는 성능 최적화 목적이면 Cache가 훨씬 빠름..
체크포인트는 I/O 병목 발생 ( 체크포인트는 중간 결과를 디스크 / HDFS 저장 )

---

 

3. 브로드캐스트 조인 → "셔플 제거"

작은 테이블을 복사해서 셔플 없이 조인

---

3-1. 일반 Join 문제

A join B

• 셔플 발생 → 디스크 I/O + 네트워크 이동
• 키 단위 데이터 이동의 → 분산처리 병목 발생

---

3-2. 브로드캐스트 조인

from pyspark.sql.functions import broadcast

df = big_df.join(broadcast(small_df), "id")

1. 작은 테이블 → 모든 익스큐터에 복사
2. 각 노드에서 로컬 Join

즉, 셔플 없음 → 네트워크 병목 제거

 

[ 브로드캐스트 조인 자동 ]

spark.sql.autoBroadcastJoinThreshold
→ 일정 크기 이하 → 최적화시 자동 브로드캐스트

 

[ 브로드캐스트 조인 주의 사항 ]

• 작은 테이블이 크면 → 메모리 터짐 or GC 성능 저하 발생
• 클러스터 리소스와 상황을 고려해 조인 설정!

---

 

 

1. Spark 지연 연산
   - 스파크는 액션(show/count 등) 호출 전까지 계산 안 함, 액션마다 DAG 재실행됨

2. cache(): 같은 연산 여러 번 쓰면 "한 번 계산 → 메모리 재사용"
3. persist(): cache() 확장판, 스토리지레이어를 사용하여 메모리/디스크/직렬화 등 저정 전략 유연하게 선택
4. checkpoint(): 긴 DAG 끊고 디스크 저장 → 반복 연산 최적화 / 안전성용
5. Broadcast Join: 작은 테이블 복사해 셔플 제거 → 조인 성능 핵심( 디스크 + 네트워크 병목 제거 )

---