스파크 잡 스케줄링

스파크 컴퓨팅 간에 자원을 스케줄링하는 방법은 크게 두 층위로 나뉘어요. 하나는 서로 다른 애플리케이션(각 SparkContext 인스턴스)이 하나의 클러스터를 나눠 쓰는 '애플리케이션 간' 스케줄링이고, 다른 하나는 애플리케이션 안에서 여러 잡(액션)이 동시에 도는 '애플리케이션 내' 스케줄링이에요. 클러스터 매니저가 전자를 담당하고, 스파크의 fair 스케줄러가 후자를 담당하죠. 이 페이지에서 두 층위를 각각 뜯어볼게요.

출처: Job Scheduling - Apache Spark 공식 문서

본문

애플리케이션 간 스케줄링

클러스터에서 각 스파크 애플리케이션은 독립된 익스큐터 JVM 집합을 갖고, 그 애플리케이션의 태스크를 실행하고 데이터를 저장해요. 여러 사용자가 클러스터를 공유해야 한다면 클러스터 매니저에 따라 할당을 관리하는 옵션이 달라져요.

모든 클러스터 매니저에서 쓸 수 있는 가장 단순한 방법은 자원을 정적으로 분할하는 거예요. 이 방식에서는 각 애플리케이션에 최대 사용 자원이 주어지고, 애플리케이션 전체 수명 동안 그 자원을 계속 붙잡아요. 스탠드얼론, YARN, K8s 모드가 모두 이 방식을 써요. 클러스터 유형별 설정은 이렇게 달라져요.

스탠드얼론 모드: 기본적으로 제출된 애플리케이션은 FIFO(선입선출) 순서로 돌고, 각 애플리케이션은 사용 가능한 모든 노드를 쓰려 해요. spark.cores.max 설정으로 한 애플리케이션이 쓰는 노드 수를 제한할 수 있고, 이 설정이 없는 애플리케이션의 기본값은 spark.deploy.defaultCores로 바꿀 수 있어요. 코어 외에도 spark.executor.memory가 각 애플리케이션의 메모리 사용을 제어해요.
YARN: --num-executors 옵션(설정 속성 spark.executor.instances)이 클러스터에 할당할 익스큐터 수를, --executor-memory(spark.executor.memory)와 --executor-cores(spark.executor.cores)가 익스큐터당 자원을 제어해요.
K8s: YARN과 상황이 같아요. 게다가 Spark on K8s는 spark.kubernetes.executor.limit.cores와 spark.kubernetes.executor.request.cores의 더 높은 우선순위 버전을 spark.executor.cores보다 제공해요.

어느 모드도 현재 애플리케이션 간 메모리 공유를 제공하지 않아요. 데이터를 그렇게 공유하고 싶다면, 같은 RDD를 조회하며 여러 요청을 처리하는 단일 서버 애플리케이션을 돌리는 걸 권장해요.

동적 자원 할당

스파크는 워크로드에 따라 애플리케이션이 점유하는 자원을 동적으로 조절하는 메커니즘도 제공해요. 더 이상 안 쓰는 자원은 클러스터에 돌려주고, 수요가 생기면 다시 요청하는 식이에요. 여러 애플리케이션이 스파크 클러스터 자원을 공유할 때 특히 유용하죠. 이 기능은 기본 비활성이며, 모든 coarse-grained 클러스터 매니저(스탠드얼론, YARN, K8s)에서 사용 가능해요.

사용하려면 먼저 spark.dynamicAllocation.enabled를 true로 설정해야 해요. 그리고 아래 중 하나를 함께 만족해야 해요.

동일 클러스터의 각 워커 노드에 외부 셔플 서비스를 구축한 뒤 spark.shuffle.service.enabled를 true로 설정, 또는
spark.dynamicAllocation.shuffleTracking.enabled를 true로 설정, 또는
spark.decommission.enabled와 spark.storage.decommission.shuffleBlocks.enabled를 둘 다 true로 설정, 또는
spark.shuffle.sort.io.plugin.class에 신뢰할 수 있는 저장을 지원하는 ShuffleDriverComponents가 있는 커스텀 ShuffleDataIO를 설정.

외부 셔플 서비스나 셔플 트래킹이 필요한 이유는, 익스큐터가 제거돼도 그 익스큐터가 쓴 셔플 파일이 삭제되지 않게 하려는 거예요. 셔플 트래킹은 켜기 쉽지만 외부 셔플 서비스 구성은 매니저별로 달라요. 스탠드얼론은 워커를 spark.shuffle.service.enabled=true로 시작하면 되고, YARN은 별도 안내를 따라요. 나머지 관련 설정은 모두 선택이며 spark.dynamicAllocation.*와 spark.shuffle.service.* 네임스페이스 아래 있어요.

주의할 점이 두 가지 있어요. 스탠드얼론에서 spark.executor.cores를 명시하지 않으면 익스큐터가 워커의 모든 코어를 가져가는데, 이 상태에서 동적 할당을 켜면 예상보다 훨씬 많은 익스큐터를 확보할 수 있어요. 또 K8s는 아직 외부 셔플 서비스를 지원하지 않아서 spark.shuffle.service.enabled=true로는 이 기능을 쓸 수 없어요.

자원 할당 정책

높은 수준에서 보면, 스파크는 더 이상 안 쓰는 익스큐터를 놓아주고 필요한 익스큐터를 확보해요. 제거될 익스큐터가 곧 태스크를 돌릴지, 새로 추가될 익스큐터가 실제로 놀지는 알 수 없으니 제거·요청 시점을 판단하는 휴리스틱이 필요해요.

요청 정책: 동적 할당이 켜진 애플리케이션은 스케줄 대기 중인 pending 태스크가 있을 때 익스큐터를 추가로 요청해요. pending 태스크가 있다는 건 기존 익스큐터만으로는 제출됐지만 끝나지 않은 모든 태스크를 동시에 채우기 부족하다는 뜻이죠. 요청은 라운드 단위로 이뤄지는데, spark.dynamicAllocation.schedulerBacklogTimeout 초 동안 pending 태스크가 있으면 처음 요청이 발동하고, 이후에도 pending 대기열이 이어지면 spark.dynamicAllocation.sustainedSchedulerBacklogTimeout 초마다 다시 요청돼요. 게다가 라운드마다 요청하는 익스큐터 수는 이전 라운드에서 지수적으로 늘어나요 — 첫 라운드에 1개, 이후 2, 4, 8개 식으로요. 지수 증가 정책의 이유는 두 가지예요. 첫째, 사실 몇 개만 더 있으면 될 수도 있으니 처음엔 신중하게 요청하고(TCP slow start와 같은 논리), 둘째, 실제로 많은 익스큐터가 필요하다면 제때 자원 사용을 늘려야 하기 때문이에요.

제거 정책: 익스큐터 제거는 훨씬 단순해요. spark.dynamicAllocation.executorIdleTimeout 초보다 오래 유휴 상태면 익스큐터를 제거해요. 대부분의 상황에서 이 조건은 요청 조건과 상호 배타적이에요 — 아직 스케줄 대기 중인 태스크가 있으면 익스큐터가 유휴 상태일 수 없으니까요.

익스큐터의 우아한 퇴역(graceful decommission)

동적 할당이 없던 시절엔 애플리케이션이 끝날 때 익스큐터가 함께 나가서 그 상태를 버려도 안전했어요. 하지만 동적 할당에서는 애플리케이션이 계속 도는 중에 익스큐터가 명시적으로 제거돼요. 애플리케이션이 그 익스큐터에 저장·기록된 상태를 접근하려면 상태를 재계산해야 하죠. 그래서 스파크는 익스큐터를 제거하기 전에 그 상태를 보존하며 우아하게 퇴역시키는 메커니즘이 필요해요.

이 요구는 특히 셔플에서 중요해요. 셔플 중 스파크 익스큐터는 자기 map 출력을 로컬 디스크에 먼저 쓰고, 다른 익스큐터가 그 파일을 가져갈 때 서버 역할을 해요. 동료보다 훨씬 오래 도는 straggler 태스크가 생길 때, 셔플이 끝나기 전에 익스큐터가 제거되면 그 익스큐터가 쓴 셔플 파일을 불필요하게 재계산해야 해요.

셔플 파일 보존의 해법은 스파크 1.2에서 도입된 외부 셔플 서비스예요. 이 서비스는 스파크 애플리케이션과 그 익스큐터와 무관하게 클러스터 각 노드에서 도는 장기 실행 프로세스예요. 서비스가 켜져 있으면 익스큐터는 서로에게서가 아니라 서비스에서 셔플 파일을 가져와요. 즉 어떤 익스큐터가 쓴 셔플 상태가 익스큐터 수명을 넘어 계속 제공될 수 있는 거죠.

셔플 파일 외에도 익스큐터는 데이터를 디스크나 메모리에 캐시해요. 익스큐터가 제거되면 그 캐시 데이터는 더 이상 접근 불가능해져요. 기본적으로 캐시 데이터를 가진 익스큐터는 절대 제거되지 않는데, 이 동작을 spark.dynamicAllocation.cachedExecutorIdleTimeout으로 설정할 수 있어요. spark.shuffle.service.fetch.rdd.enabled를 true로 설정하면 ExternalShuffleService로 디스크에 영속된 RDD 블록을 가져올 수 있고, 이 기능이 켜진 상태에서는 디스크 영속 블록만 가진 익스큐터도 spark.dynamicAllocation.executorIdleTimeout 후 유휴로 간주돼 해제돼요.

애플리케이션 내 스케줄링

하나의 스파크 애플리케이션(SparkContext 인스턴스) 안에서도 서로 다른 스레드에서 제출된 여러 병렬 잡이 동시에 돌 수 있어요. 여기서 '잡'은 하나의 스파크 액션(예: save, collect)과 그 액션을 평가하는 데 필요한 태스크를 뜻해요. 스파크 스케줄러는 완전히 스레드 안전해서, 여러 사용자의 쿼리를 처리하는 애플리케이션 같은 다중 요청 시나리오를 지원해요.

기본적으로 스파크 스케줄러는 잡을 FIFO 방식으로 돌려요. 각 잡은 stage(map·reduce 단계)로 나뉘고, 첫 잡이 태스크를 실행할 수 있는 동안 모든 가용 자원을 우선 사용한 뒤 두 번째 잡이 우선권을 갖는 식이에요. 대기열 앞쪽 잡이 클러스터 전체를 안 쓴다면 뒤쪽 잡이 바로 시작할 수 있지만, 앞쪽 잡이 크다면 뒤쪽 잡은 크게 지연될 수 있어요.

스파크 0.8부터 잡 간 공평 공유(fair sharing)도 설정할 수 있어요. 공평 공유에서는 스파크가 잡 사이에 태스크를 '라운드 로빈' 방식으로 나눠 줘서 모든 잡이 클러스터 자원을 거의 동등하게 나눠 가져요. 긴 잡이 도는 중에 제출된 짧은 잡이 끝나길 기다리지 않고 바로 자원을 받아 좋은 응답 시간을 유지할 수 있어요. 멀티 사용자 환경에 가장 적합하죠. 기본 비활성이며 스탠드얼론, YARN, K8s 모두에서 쓸 수 있어요.

fair 스케줄러를 켜려면 SparkContext를 설정할 때 spark.scheduler.mode 속성을 FAIR로 지정하면 돼요.

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.set("spark.scheduler.mode", "FAIR")
val sc = new SparkContext(conf)

Fair 스케줄러 풀

fair 스케줄러는 잡을 풀(pool)로 묶고 풀마다 다른 스케줄링 옵션(예: weight)을 설정할 수 있어요. 더 중요한 잡을 위한 '우선순위 높은' 풀을 만들거나, 사용자별로 잡을 묶어 동시 잡 수와 무관하게 사용자에게 동등한 몫을 주는 식으로 쓸 수 있어요. 이 접근은 Hadoop Fair Scheduler를 본뜬 거예요.

별도 조작 없으면 새로 제출된 잡은 기본 풀로 들어가요. 잡의 풀은 제출하는 스레드의 SparkContext에 spark.scheduler.pool '로컬 속성'을 추가해 설정돼요.

// Assuming sc is your SparkContext variable
sc.setLocalProperty("spark.scheduler.pool", "pool1")

이 로컬 속성을 설정하면 그 스레드가 제출하는 모든 잡(RDD.save, count, collect 호출 등)이 이 풀 이름을 사용해요. 설정은 스레드 단위라 한 스레드가 같은 사용자를 대신해 여러 잡을 돌리기 쉬워요. 스레드가 연결된 풀을 지우려면 이렇게 해요.

sc.setLocalProperty("spark.scheduler.pool", null)

풀 기본 동작: 기본적으로 각 풀은 클러스터의 동등한 몫을 받지만(기본 풀의 각 잡과도 동등), 풀 안에서는 잡이 FIFO 순서로 돌아요. 예를 들어 사용자별로 풀을 하나씩 만들면 사용자마다 클러스터의 동등한 몫을 받고, 각 사용자의 쿼리는 나중 쿼리가 앞선 쿼리의 자원을 빼앗지 않고 순서대로 돌죠.

풀 속성 설정: 구체적인 풀 속성은 설정 파일로도 바꿀 수 있어요. 각 풀은 세 가지 속성을 지원해요.

schedulingMode: FIFO 또는 FAIR. 풀 안 잡이 서로 뒤에 대기열을 이루는지(기본), 풀 자원을 공평하게 나누는지 제어해요.
weight: 다른 풀에 상대적인 클러스터 몫을 제어해요. 기본 weight는 1이고, 특정 풀에 2를 주면 다른 활성 풀보다 2배 많은 자원을 받아요. weight 1000 같은 높은 값도 풀 간 우선순위를 구현하는 데 쓸 수 있어요 — 사실상 weight-1000 풀은 잡이 활성일 때마다 항상 태스크를 먼저 실행하게 돼요.
minShare: 전체 weight와 별개로 관리자가 원하는 최소 몫(CPU 코어 수)을 풀에 줄 수 있어요. fair 스케줄러는 weight에 따라 남는 자원을 재분배하기 전에 먼저 모든 활성 풀의 최소 몫을 맞추려 해요. 그래서 minShare는 클러스터 나머지에 높은 우선순위를 주지 않으면서도 특정 풀이 특정 자원 수(예: 10코어)까지는 빨리 확보하게 하는 또 다른 방법이에요. 기본 minShare는 0이에요.

풀 속성은 conf/fairscheduler.xml.template 같은 XML 파일을 만들어 설정해요. fairscheduler.xml 이름으로 클래스패스에 넣거나, SparkConf의 spark.scheduler.allocation.file 속성으로 지정하면 돼요. 파일 경로는 hadoop 설정을 따르며 로컬 경로나 HDFS 경로가 될 수 있어요.

// scheduler file at local
conf.set("spark.scheduler.allocation.file", "file:///path/to/file")
// scheduler file at hdfs
conf.set("spark.scheduler.allocation.file", "hdfs:///path/to/file")

XML 형식은 풀마다 <pool> 요소를 두고 그 안에 여러 설정 요소를 넣는 거예요.

<?xml version="1.0"?>
<allocations>
  <pool name="production">
    <schedulingMode>FAIR</schedulingMode>
    <weight>1</weight>
    <minShare>2</minShare>
  </pool>
  <pool name="test">
    <schedulingMode>FIFO</schedulingMode>
    <weight>2</weight>
    <minShare>3</minShare>
  </pool>
</allocations>

전체 예시는 conf/fairscheduler.xml.template에도 있어요. XML에 설정되지 않은 풀은 모든 설정을 기본값(스케줄링 모드 FIFO, weight 1, minShare 0)으로 받아요.

JDBC 커넥션 스케줄링: JDBC 클라이언트 세션에 Fair Scheduler 풀을 설정하려면 spark.sql.thriftserver.scheduler.pool 변수를 쓰면 돼요.

SET spark.sql.thriftserver.scheduler.pool=accounting;

PySpark의 병행 잡: PySpark는 기본적으로 PVM 스레드와 JVM 스레드를 동기화하지 않아서, 여러 PVM 스레드에서 여러 잡을 띄워도 각 잡이 각각 대응하는 JVM 스레드에서 실행된다는 보장이 없어요. 이 한계 때문에 별도 PVM 스레드에서 sc.setJobGroup으로 다른 잡 그룹을 설정하는 것도 불가능하고, 나중에 sc.cancelJobGroup으로 잡을 취소하는 것도 막혀요. PVM 스레드가 로컬 속성 같은 상속 가능 속성을 JVM 스레드에서 물려받으려면 pyspark.InheritableThread를 함께 쓰는 걸 권장해요.

스파크 잡 스케줄링