이번에 EMR 사용하는 파이프라인의 에러 디버깅 및 개선 작업을 하며 공부한것들 정리..
EMR / Spark 컴포넌트
EMR과 YARN
EMR 클러스터 노드 타입
•
Master:
◦
YARN ResourceManager, HDFS NameNode, Spark Driver 위치
◦
클러스터 메타 관리, 오케스트레이션 담당
•
Core:
◦
YARN NodeManager 실행
◦
Spark Executor 컨테이너가 실제 뜨는 노드
◦
HDFS 데이터 저장도 수행
•
Task
◦
실행만 하는 임시 노드
◦
HDFS 저장 없음 → Spot/Auto Scaling 용도로 활용
YARN(클러스터 리소스 매니저)
•
ResourceManager(RM): 클러스터 전체 자원(메모리/vCore) 스케줄링
•
NodeManager(NM): 각 노드에서 컨테이너 띄우고 상태 보고
•
Container: YARN이 할당하는 실행 단위로 Spark Driver/Executor가 컨테이너로 뜬다
•
ApplicationMaster(AM): YARN 앱의 리더컨테이너
◦
Spark 클러스터 모드에서는 Driver가 AM 컨테이너로 실행
◦
Task 스케줄링·재시도·상태 관리 담당
Spark 컴포넌트
Driver
•
코드 파싱 → 논리 계획(DAG) 생성 → 물리 계획 → 스케줄링
•
YARN 모드에서는 ApplicationMaster 컨테이너로 실행
•
코어 노드 중 하나에 배치되며, 메모리/코어를 점유
Executor
•
실제 연산을 수행하는 워커 프로세스
•
여러 개의 코어(=슬롯)를 갖고 Task를 병렬 실행
•
주요 파라미터:
◦
spark.executor.instances: Executor 컨테이너 개수
◦
spark.executor.cores: Executor 1개가 동시에 돌릴 Task 스레드 수
◦
spark.executor.memory: Executor JVM 힙 크기
•
GC, Task 경합 등을 고려해 보통 코어당 1~3 Task가 적절
Task / Partition
•
한 파티션 = 한 Task
•
Stage 시작 시 파티션 수만큼 Task가 만들어져 Executor 슬롯에 배치
Job / Stage
•
Job: count(), write.parquet() 같은 Action 1회마다 생성되는 실행 단위(여러 스테이지 포함)
•
Stage: 셔플 경계(wide transformation)에서 분리되는 실행 단위
◦
Stage 안에서는 Task가 병렬 처리되고
◦
Stage가 끝나야 다음 Stage 실행 가능
Spark 실행 흐름
1.
Driver 시작 – 코드 파싱 및 실행 계획 생성
2.
DAG(논리 계획) – Transformation 의존 그래프 구성
3.
Job 생성 – Action 만나면 Job이 만들어져 실행 시작
4.
Stage 분할 – narrow 연산은 묶이고, wide 연산(셔플)에서 Stage 분리
5.
Task 생성 – Stage별 파티션 수만큼 Task 생성
6.
자원 요청 & 배치 – Driver가 YARN에 Executor N개 × C코어 요청
7.
실행 – TaskScheduler가 Task를 Executor 슬롯에 배포, Stage 단위로 진행
리소스 설정 예시
슬롯(Task 동시 실행 수)
•
Executor 코어 1개 = 슬롯 1개
•
총 슬롯 = executor.instances × executor.cores
•
슬롯 수의 3~6배 Task가 있을 때 가장 안정적(슬롯 네개 일때 task 12~24개)
YARN 자원 배분 예시
•
m4.large: vCPU 2, 메모리 8 GiB
•
YARN이 실제로 쓰는 메모리: OS/데몬을 빼고 대략 6–6.5 GiB
•
컨테이너(Executor) 메모리 계산
◦
executorMemory + memoryOverhead
◦
기본 Overhead = max(384MB, executorMemory의 10%)
◦
예: executor.memory=5G → Overhead≈512MB → 컨테이너≈5.5GiB
◦
→ 노드당 1 executor가 현실적
•
예시 클러스터 - 1 Master, 2 core 노드
◦
executor=2 코어 노드수와 매칭
◦
cores=2 CPU 2개와 매칭
•
EMR(YARN cluster mode)에서는 Driver가 ApplicationMaster 컨테이너로 뜨는데, 이 컨테이너는 마스터 노드에만 고정되는 게 아니라, 코어 노드 중 하나에 스케줄링
•
처음엔 단순히 코어 노드 개수 = executor 인스턴스 개수로 맞췄는데 드라이버 때문에 Executor 수는 Core 노드 수 - 1 로 설정해야함
파티션·셔플·출력 파일
입력 파티션
•
S3 기반 소스:
◦
Parquet/ORC → Row Group 단위 split
◦
CSV/JSON → 파일 크기에 따라 split
•
작은 파일이 많으면 작은 파티션이 쌓이고, 큰 파일은 적은 파티션으로 처리
셔플 파티션
•
spark.sql.shuffle.partitions(기본 200)
•
wide 연산(Join, GroupBy)마다 생성
출력 파티션
•
dafault: 입력 파티션 수 = 출력 파티션 수
•
권장: 256~512MB
•
필요 시 repartition() / coalesce()로 조정
•
Parquet Writer 기본 RowGroup = 128MB