Spark

spark-submit 파라미터가 실제로 만드는 것

--num-executors N, --executor-cores C, --executor-memory M을 주면 클러스터 매니저(YARN/K8s/Standalone)는 N개의 executor 프로세스를 요청하고, 각 프로세스는 C개의 태스크를 동시에 실행할 수 있는 스레드 슬롯과 M만큼의 JVM 힙을 가진 채로 뜬다. 즉 '코어 수'는 물리 CPU 코어가 아니라 그 executor가 동시에 처리할 수 있는 태스크 슬롯 개수를 의미한다(하나의 물리 코어에 여러 슬롯을 배정하는 것도 클러스터 설정에 따라 가능하다).

--driver-memory와 --driver-cores는 별개로 Driver 프로세스에 적용된다. client 모드에서는 spark-submit을 실행한 바로 그 머신에서 Driver가 뜨고(로컬 환경 자원을 그대로 씀), cluster 모드에서는 클러스터 안의 한 노드에서 Driver가 뜬다(YARN/K8s가 배치를 결정). 운영 환경에서는 보통 cluster 모드를 쓰는데, client 모드는 제출한 머신이 종료되면 Driver도 함께 죽기 때문이다.

코어 수 사이징 — 왜 3~5개 안팎이 흔한 경험칙인가

executor 하나에 코어를 아주 많이(예: 15~19개) 몰아주면 두 가지 문제가 생긴다. 첫째, HDFS 클라이언트는 executor 하나당 하나의 커넥션 풀을 쓰는데, 같은 executor 안에서 너무 많은 태스크가 동시에 그 풀을 두드리면 HDFS I/O 처리량 자체가 병목이 된다(이른바 'HDFS client 처리량 한계'). 둘째, 힙이 커질수록 JVM 가비지 컬렉션 일시정지(GC pause)가 길어져 태스크 지연으로 이어진다.

반대로 코어를 executor당 1~2개로 너무 잘게 쪼개면(이른바 'tiny executor'), executor 프로세스 자체의 고정 오버헤드(JVM 기동, 브로드캐스트 변수 복제본, 캐시 관리 구조 등)가 상대적으로 커지고 executor 개수가 지나치게 많아져 클러스터 매니저의 스케줄링 오버헤드도 늘어난다. 그래서 실무에서는 executor당 3~5개 코어, 8~16GB 메모리 안팎을 기본 출발점으로 삼고, 실제 워크로드(셔플이 많은지, GC 압박이 있는지)를 Spark UI로 관찰하며 조정하는 방식을 권장한다 — 이 숫자 자체를 절대 공식으로 암기하기보다 'fat executor는 HDFS 처리량과 GC에서, tiny executor는 오버헤드 비율에서 손해를 본다'는 트레이드오프 구조를 이해하는 것이 중요하다.

executor.memoryOverhead — 컨테이너가 조용히 죽는 이유

클러스터 매니저(YARN/K8s)가 executor에게 실제로 배정하는 컨테이너 총 메모리는 --executor-memory 하나만이 아니다: JVM 힙(--executor-memory) + memoryOverhead(오프힙 JVM 자체 오버헤드, 네이티브 라이브러리, 스레드 스택 등) + (켰다면) off-heap 메모리 + (PySpark라면) Python 워커 프로세스 메모리를 모두 합친 크기가 실제로 요청된다.

memoryOverhead의 기본값은 max(executor-memory의 10%, 384MB)이며, Spark 3.3부터는 spark.executor.memoryOverheadFactor로 이 비율 자체를 조정할 수 있다(YARN 기본 0.1, Kubernetes 기본 0.4 — K8s는 컨테이너 격리 오버헤드가 더 크다고 보고 더 넉넉히 잡는다).

함정은 이렇다: executor-memory 8192MB + 기본 overhead 819MB(10%) + off-heap 1024MB를 켰다면 필요한 총합은 10,035MB인데, 클러스터 매니저에 실제 요청한 컨테이너 크기가 이보다 작으면(예: 9011MB) executor는 Spark 자체의 OOM 에러 없이 컨테이너 매니저(YARN NodeManager, K8s kubelet)에 의해 조용히 강제 종료된다 — RSS(실제 물리 메모리 사용량) 기준으로 감시하기 때문에 JVM 힙 안에서는 멀쩡해 보여도 죽는다. PySpark로 pandas UDF/벡터화 UDF를 많이 쓴다면 Python 워커 프로세스가 별도로 1~3GB를 추가로 먹을 수 있어, 10% 공식만 믿지 말고 memoryOverhead를 최소 2GB 이상으로 명시적으로 늘리는 것이 실무 권장사항이다.

Dynamic Allocation — 필요한 만큼만 executor를 쓴다

정적 할당(--num-executors 고정)은 잡 시작부터 끝까지 같은 개수의 executor를 붙들고 있어, 잡 초반(작은 입력을 읽는 단계)이나 후반(결과를 모으는 단계)처럼 병렬성이 필요 없는 구간에서도 자원을 낭비한다. spark.dynamicAllocation.enabled=true로 켜면 Spark가 대기 중인 태스크 큐(backlog)를 보고 executor를 늘리거나(spark.dynamicAllocation.schedulerBacklogTimeout 동안 밀린 태스크가 있으면 증설), 유휴 상태가 일정 시간(spark.dynamicAllocation.executorIdleTimeout) 지속되면 줄인다. minExecutors/maxExecutors/initialExecutors로 그 범위를 제어한다.

**중요한 제약**: executor를 줄이려면 그 executor가 들고 있던 셔플 파일(다른 executor가 나중에 읽어갈 데이터)을 누군가는 계속 서빙해줘야 한다. 전통적으로는 External Shuffle Service(각 워커 노드에 상시 떠 있는 별도 프로세스)가 이 역할을 맡아, executor가 죽어도 셔플 데이터를 계속 내려줄 수 있었다. 그런데 Kubernetes는 이 External Shuffle Service를 (이 글 작성 시점 기준) 공식 지원하지 않으므로, K8s에서 Dynamic Allocation을 쓰려면 Spark 3.0부터 도입된 spark.dynamicAllocation.shuffleTracking.enabled=true를 대신 켜야 한다 — 이는 executor가 아직 서빙 중인 셔플 데이터를 추적해서, 그 데이터가 필요 없어지거나 타임아웃이 지날 때까지는 해당 executor를 회수하지 않도록 보류하는 방식이다. shuffle tracking만으로는 External Shuffle Service처럼 완전히 자유롭게 조기 회수를 못 하므로(살아있는 잡이 그 셔플 데이터를 참조하는 동안은 대기), 셔플이 아주 큰 잡에서는 K8s 환경의 Dynamic Allocation이 YARN+ESS 조합보다 축소가 더 보수적이라는 점을 실무에서 감안해야 한다.