7. Inside the Scheduler

• 4장에서 Action → Job → Stage → Task로 이어지는 흐름과 DAGScheduler·TaskScheduler의 역할 분담을 봤음
• 이번 장은 그 두 Scheduler 안을 열어 봄
  • Stage 경계를 실제로 어떻게 찾는가
  • “이미 계산된 Stage"를 어떻게 알아보고 건너뛰는가
  • Locality를 위해 얼마나, 어떤 규칙으로 기다리는가
  • 느린 Task와 실패한 Task를 각각 누가, 어떻게 처리하는가
• 근거는 Apache Spark 4.2.0 소스 (core/src/main/scala/org/apache/spark/scheduler/)

flowchart TD
  action["rdd.collect()"] --> runJob["SparkContext.runJob"]
  runJob --> submit["DAGScheduler.submitJob<br/>JobWaiter 반환"]
  submit --> loop["Event Loop<br/>JobSubmitted"]
  loop --> createStage["createResultStage<br/>Shuffle 경계 탐색"]
  createStage --> submitStage["submitStage<br/>부모 Stage 먼저"]
  submitStage --> missing["submitMissingTasks<br/>TaskSet 생성"]
  missing --> tsi["TaskSchedulerImpl.submitTasks"]
  tsi --> tsm["TaskSetManager<br/>Locality · Speculation · 재시도"]
  tsm --> backend["SchedulerBackend.reviveOffers"]
  backend --> executors["Executors"]

7.1 The Single-Threaded Event Loop

DAGScheduler의 상태 변경은 전부 단일 스레드 이벤트 루프 안에서만 일어남.

RDD.collect()
  → SparkContext.runJob
    → DAGScheduler.runJob
      → submitJob            (JobWaiter 반환, 호출 스레드는 blocking await)
        → eventProcessLoop.post(JobSubmitted)
          → handleJobSubmitted   ← 여기부터 이벤트 루프 스레드

• Job 제출, Task 완료, Executor 유실, Fetch 실패 같은 사건이 모두 이벤트로 큐에 들어감
• 이벤트 루프가 하나씩 꺼내 처리하므로, 락 대신 직렬화로 동시성 문제를 피함
• 대신 루프를 오래 붙잡는 작업은 금물

그래서 느릴 수 있는 작업은 루프에 넣기 전에 미리 끝냄.

// SPARK-23626: `RDD.getPartitions()` can be slow, so we eagerly compute
// `.partitions` on every RDD in the DAG to ensure that `getPartitions()`
// is evaluated outside of the DAGScheduler's single-threaded event loop
eagerlyComputePartitionsForRddAndAncestors(rdd)

val jobId = nextJobId.getAndIncrement()
val waiter = new JobWaiter[U](this, jobId, partitions.size, resultHandler)
eventProcessLoop.post(JobSubmitted(...))
waiter

• 예: 파일 기반 RDD의 getPartitions()는 파일 목록 조회가 필요해 오래 걸릴 수 있음
• 이걸 호출 스레드에서 먼저 계산해 둬서, 루프 안에서는 캐시된 값만 읽게 함

참고로 SparkContext.runJob이 하는 일:

• 사용자 함수를 ClosureCleaner로 정리함 (4.9에서 본 Task Closure를 직렬화하기 전에 불필요한 참조를 없앰)
• 호출 위치(call site)를 기록함. 이 값이 Spark UI의 Stage 이름이 됨
• Job이 끝난 뒤에 rdd.doCheckpoint()를 실행함

7.2 Finding Shuffle Boundaries

4.3·4.5에서 “DAGScheduler가 최종 RDD부터 부모 방향으로 추적하다가 Shuffle에서 Stage를 자른다"고 했음. 실제 구현은 명시적 스택을 쓰는 DFS임.

toVisit.dependencies.foreach {
  case shuffleDep: ShuffleDependency[_, _, _] =>
    parents += shuffleDep        // 경계: 더 올라가지 않음
  case dependency =>
    enqueue(dependency.rdd)      // Narrow면 계속 위로 = 같은 Stage
}

• Narrow Dependency를 만나면 부모 RDD를 스택에 넣고 계속 올라감
• ShuffleDependency를 만나면 거기서 멈추고 “부모 Shuffle"로 기록
• 즉 한 Stage는 Shuffle 경계까지 Narrow Dependency로 닿는 RDD 전부
• 4.x에서 재귀 호출을 명시적 스택으로 바꿨음. Lineage가 매우 길어도 StackOverflowError가 나지 않게 하기 위함

왜 Narrow는 이어 붙일 수 있나

abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] {
  def getParents(partitionId: Int): Seq[Int]
}

class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {
  override def getParents(partitionId: Int): List[Int] = List(partitionId)
}

• getParents(partitionId)는 즉시 계산할 수 있는 함수
  • OneToOneDependency: map·filter 계열. 부모 Partition 번호가 자기 번호와 같음
  • RangeDependency: union. 번호를 일정 범위만큼 옮김
• 자식 Partition에 필요한 부모 Partition을 실행하지 않고도 알 수 있으므로, 같은 Task 안에서 Pipelining할 수 있음

ShuffleDependency에는 getParents가 없음.

• 자식 Partition 하나가 모든 부모 Partition 출력의 일부를 필요로 하기 때문
• 그래서 부모 Stage 전체가 끝나야 하는 Barrier가 생김
• ShuffleDependency가 생성되는 순간 shuffleManager.registerShuffle()도 호출되고, 이 시점에 Shuffle Writer 종류가 정해짐 (8.1)

toDebugString으로 경계 보기

print(result.toDebugString().decode())

출력 모양 (단순화):

(2) ShuffledRDD[4] at reduceByKey
 +-(4) MapPartitionsRDD[3] at map
    |  MapPartitionsRDD[2] at filter
    |  MapPartitionsRDD[1] at map
    |  orders.csv HadoopRDD[0] at textFile

• 들여쓰기가 바뀌는 +- 지점이 ShuffleDependency, 곧 Stage 경계
• 괄호 안 숫자는 Partition 수
• spark.logLineage=true면 Job을 제출할 때마다 자동으로 출력됨

7.3 Stage Reuse and Skipped Stages

부모 Stage는 getOrCreateShuffleMapStage로 만듦. 이름 그대로 이미 있으면 재사용함.

shuffleIdToMapStage: shuffleId → ShuffleMapStage

• 같은 Shuffle을 여러 Job이 쓰면 Stage 객체도 공유됨
• 그 Shuffle Output이 이미 다 있으면 다시 실행할 필요가 없음
• Spark UI에서 회색 skipped stage로 보이는 것이 이것

pairs = orders.map(lambda x: (x.region, x.amount)).reduceByKey(add)

pairs.count()     # Job 0: ShuffleMapStage 실행 → ResultStage 실행
pairs.collect()   # Job 1: ShuffleMapStage는 skipped → ResultStage만 실행

“무엇이 빠졌나"를 판단하는 기준

두 Stage는 “아직 계산하지 않은 Partition"을 서로 다른 곳에서 확인함.

// ShuffleMapStage
def isAvailable: Boolean = numAvailableOutputs == numPartitions
override def findMissingPartitions(): Seq[Int] =
  mapOutputTrackerMaster.findMissingPartitions(shuffleDep.shuffleId)
    .getOrElse(0 until numPartitions)

// ResultStage
override def findMissingPartitions(): Seq[Int] = {
  val job = activeJob.get
  (0 until job.numPartitions).filter(id => !job.finished(id))
}

• Executor가 죽어 Map Output 일부가 사라지면 MapOutputTracker에서 그 항목이 지워짐
• 그러면 findMissingPartitions()가 사라진 Partition만 다시 돌려줌
• 이것이 Stage 전체가 아니라 빠진 조각만 재계산하는 구조의 핵심

캐시도 경계를 바꿈

getMissingParentStages는 캐시 상태를 반영함.

• 부모 RDD의 모든 Partition이 캐시돼 있으면 그 위로 올라가지 않음
• 캐시가 Lineage를 “잘라 주는” 효과가 Stage를 만드는 단계에서 이렇게 드러남

7.4 From Stage to TaskSet

Preferred Location 결정 순서

submitMissingTasks는 각 Partition의 선호 위치를 다음 순서로 찾음.

① 캐시 위치        → 그 Partition이 캐시된 Executor
② RDD 자체 선호    → 예: HDFS 블록이 있는 호스트 (3.3의 getPreferredLocations)
③ Narrow 부모 재귀 → 부모 Partition의 선호 위치를 물려받음
④ 없음             → NO_PREF

• ③ 때문에 textFile().map().filter()처럼 Narrow로 이어진 RDD도 원본 블록 위치를 그대로 선호함
• Shuffle 이후 RDD는 ③에서 멈추므로 보통 선호 위치가 없음

Task Binary 브로드캐스트

ShuffleMapStage → (rdd, shuffleDep) 직렬화
ResultStage     → (rdd, func) 직렬화
          ↓
Broadcast로 Executor에 한 번만 배포

Executor 쪽 실행은 단 한 줄씩임.

// ShuffleMapTask.runTask → MapStatus 반환
dep.shuffleWriterProcessor.write(rdd.iterator(partition, context), dep, mapId, partitionId, context)

// ResultTask.runTask → 결과 U 반환
func(context, rdd.iterator(partition, context))

• rdd.iterator()를 한 번 호출하면 Stage 안의 모든 Narrow 변환이 Iterator 체인으로 흐름 (4.10)

TaskSchedulerImpl.submitTasks

• TaskSet마다 TaskSetManager(TSM)를 하나 만듦
• 같은 Stage가 재시도되면 이전 TSM은 zombie로 표시되므로, Stage당 활성 TSM은 하나뿐
• TSM을 Scheduling Pool에 넣고(7.9) backend.reviveOffers()로 자원 제안을 요청

7.5 Locality Levels in Detail

4.9의 Locality 단계를 코드로 보면:

object TaskLocality extends Enumeration {
  val PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY = Value
  def isAllowed(constraint: TaskLocality, condition: TaskLocality): Boolean =
    condition <= constraint
}

• Enumeration 순서가 곧 서수라서 <= 비교로 “더 로컬한가"를 판정함
• NO_PREF가 NODE_LOCAL과 RACK_LOCAL 사이에 있음
  • 선호 위치가 없는 Task는 노드 단위까지만 기다린 뒤 바로 실행할 수 있음

Task는 대기 큐 여러 개에 중복 등록됨.

TaskSetManager pending queues
├─ forExecutor  ← 캐시 블록이 있는 Executor
├─ forHost
├─ forRack
├─ noPrefs
└─ all

• forExecutor에 들어가는 건 캐시 블록 위치(ExecutorCacheTaskLocation)뿐
• 즉 PROCESS_LOCAL은 사실상 캐시된 RDD 전용
• 그런데 noPrefs 큐에서 꺼낸 Task도 PROCESS_LOCAL로 보고됨. UI에 PROCESS_LOCAL이 생각보다 많이 보이는 이유가 이것

유효한 레벨은 실제로 살아 있는 Executor, 호스트, 랙이 있는 레벨만 계산하고, ANY는 항상 포함됨.

7.6 Delay Scheduling

“데이터 가까운 Executor를 얼마나 기다릴 것인가"를 정하는 알고리즘.

while (currentLocalityIndex < myLocalityLevels.length - 1) {
  val moreTasks = ... // 현재 레벨 큐에 실행할 Task가 남았는가
  if (!moreTasks) {
    // 이 레벨에 Task가 없으면 기다리지 않고 바로 다음 레벨로 (SPARK-4939)
    lastLocalityWaitResetTime = curTime
    currentLocalityIndex += 1
  } else if (curTime - lastLocalityWaitResetTime >= localityWaits(currentLocalityIndex)) {
    // 대기 시간 초과 → 한 단계 강등
    lastLocalityWaitResetTime += localityWaits(currentLocalityIndex)
    currentLocalityIndex += 1
  } else {
    return myLocalityLevels(currentLocalityIndex)   // 아직 기다림 → 이 offer는 거절
  }
}

flowchart LR
  p["PROCESS_LOCAL"] -- "3s 동안 못 잡음" --> n["NODE_LOCAL"]
  n -- "3s" --> np["NO_PREF<br/>대기 0"]
  np --> r["RACK_LOCAL"]
  r -- "3s" --> a["ANY<br/>대기 0"]

• 기본 대기 시간은 spark.locality.wait = 3s이고, 레벨별로 .process·.node·.rack으로 따로 지정할 수 있음
• 실제로 기다리는 레벨은 PROCESS, NODE, RACK 셋뿐 (NO_PREF, ANY는 0)
• 강등할 때 기준 시각을 현재 시각으로 덮어쓰지 않고 대기 시간만큼 더함
  • 덮어쓰면 다음 레벨 타이머가 처음부터 다시 시작
  • 더하면 이미 오래 기다린 경우 다음 레벨도 곧바로 만료될 수 있음
• 바깥 루프는 가장 로컬한 레벨부터 자원 offer를 소진한 뒤 다음 레벨로 넘어감

실무 감각

locality.wait을 줄이면
→ 데이터가 멀어도 빨리 시작 (네트워크 읽기 증가)

locality.wait을 늘리면
→ 로컬 실행은 늘지만 Executor가 노는 시간 증가

• 오브젝트 스토리지(S3 등)에서 읽는 경우 원본 데이터의 “노드 로컬” 개념이 거의 없음
• 이럴 때는 캐시된 RDD가 아니면 기다릴 이유가 적음

7.7 Speculative Execution

4.11의 재시도는 “실패한 Task"를 다룸. 느리기만 한 Task(straggler)는 Speculation으로 다룸.

if (numSuccessfulTasks >= minFinishedForSpeculation) {
  val medianDuration = successfulTaskDurations.percentile()
  val threshold = max(speculationMultiplier * medianDuration, minTimeToSpeculation.toDouble)
  foundTasks = checkAndSubmitSpeculatableTasks(timeMs, threshold)
}

• 시작 조건: 전체 Task의 quantile 비율 이상이 성공했을 때 (기본 90%)
• 대상: 실행 시간이 max(multiplier × 중앙값, 최소 시간)을 넘은 Task (기본 3 × median, 100ms)
• 이미 복제본이 하나 돌고 있으면 더 띄우지 않으므로 Task당 최대 2개
• 먼저 끝난 쪽 결과를 쓰고, 나머지 attempt는 kill함
• 일반 Task가 항상 Speculative Task보다 먼저 배정됨

• 3.4부터는 처리율(processed rows/sec)이 평균보다 낮을 때만 복제하는 efficiency 모드도 있음 (spark.speculation.efficiency.enabled)
• 주의: Skew 때문에 느린 Task는 복제해도 같은 데이터를 다시 처리하므로 소용없음 (5.4)
• 외부 시스템에 쓰는 Task는 두 attempt가 동시에 쓸 수 있으므로 출력 커밋이 멱등해야 함

7.8 Two Layers of Failure Handling

4.11에서 본 구분을 코드 기준으로 정리하면:

Stage 안에서 Task가 그냥 실패
→ TaskScheduler(TaskSetManager)가 재시도

Shuffle 데이터가 사라짐 (FetchFailed, Executor 유실)
→ DAGScheduler가 부모 Stage를 다시 제출

Task 재시도

if (!isZombie && reason.countTowardsTaskFailures) {
  numFailures(index) += 1
  if (numFailures(index) >= maxTaskFailures) {
    abort("Task %d in stage %s failed %d times...")
    return
  }
}

• 실패 횟수는 Task index별로 세고, 성공하면 0으로 리셋함
• spark.task.maxFailures 기본값은 4. 같은 Task가 4번 실패하면 Stage를 abort하고 Job이 실패함
• 모든 실패가 횟수에 들어가지는 않음
  • 애플리케이션과 무관한 Executor 유실(노드 회수, 스팟 인스턴스 종료 등)은 countTowardsTaskFailures = false
  • 인프라 사정으로 죽은 건 재시도 예산을 쓰지 않음
• Barrier Stage는 Task 하나만 실패해도 Stage 전체를 다시 시작

Stage 재시도: FetchFailed

case FetchFailed(bmAddress, shuffleId, _, mapIndex, reduceId, failureMessage) =>
  // 잃어버린 Map Output 등록 해제
  mapOutputTracker.unregisterMapOutput(shuffleId, mapIndex, bmAddress)
  // 실패한 Stage와 부모 Map Stage를 함께 재제출 대기열에
  failedStages += failedStage
  failedStages += mapStage
  if (noResubmitEnqueued) scheduleResubmit()

• unregisterMapOutput 덕분에 부모 Stage는 7.3의 findMissingPartitions()로 잃은 조각만 재계산함
• 재시도 한도는 두 겹
  • spark.stage.maxConsecutiveAttempts (기본 4): 연속 실패 횟수, 성공하면 초기화
  • spark.stage.maxAttempts (기본 Int.MaxValue): 누적 절대 상한

Executor 유실과 External Shuffle Service

val fileLost = !sc.shuffleDriverComponents.supportsReliableStorage() &&
  (workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled)

• External Shuffle Service(ESS)가 켜져 있으면 Shuffle 파일은 Executor 프로세스가 아니라 노드의 별도 서비스가 서빙함
• 그래서 Executor가 죽어도 파일이 남아 있으면 재계산이 필요 없음
• Dynamic Allocation이 ESS(또는 Shuffle Tracking)를 요구하는 이유: Executor를 회수해도 Shuffle 파일이 살아 있어야 하기 때문
• 이미 성공한 ShuffleMapTask라도 그 출력이 있던 Executor가 죽으면 성공을 취소하고 다시 큐에 넣음

flowchart TD
  fail["Task 실패"] --> kind{"원인"}
  kind -- "일반 예외" --> tsm["TaskSetManager<br/>index별 실패 횟수 +1"]
  tsm --> limit{"maxFailures 도달?"}
  limit -- "아니오" --> retry["다른 Executor에서 재시도"]
  limit -- "예" --> abort["Stage abort → Job 실패"]
  kind -- "앱 무관 Executor 유실" --> retryFree["횟수 증가 없이 재시도"]
  kind -- "FetchFailed" --> dag["DAGScheduler<br/>Map Output 등록 해제"]
  dag --> resubmit["부모 Map Stage + 실패 Stage 재제출"]
  resubmit --> partial["findMissingPartitions<br/>잃은 조각만 재계산"]

7.9 FIFO and FAIR Scheduling

한 Application 안에서 여러 Job이 동시에 제출될 때 TaskSet 사이의 순서를 정하는 정책.

• spark.scheduler.mode 기본 FIFO
  • 먼저 제출된 Job의 Stage가 자원을 먼저 차지
  • 앞 Job이 크면 뒤의 작은 Job이 오래 기다림
• FAIR
  • Pool 단위로 자원을 나눔
  • 노트북·서버처럼 여러 사용자가 한 SparkContext를 공유할 때 유용

// FairSchedulingAlgorithm.comparator 요약
if (s1Needy && !s2Needy) return true                     // ① minShare 미달 Pool 우선
else if (s1Needy && s2Needy) compare = minShareRatio1 vs minShareRatio2   // ② 미달 비율 비교
else compare = taskToWeightRatio1 vs taskToWeightRatio2  // ③ weight 대비 실행 Task 수 비교

• Pool 속성: schedulingMode, weight(기본 1), minShare(기본 0)
• Pool은 스레드 로컬 속성으로 지정함

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

• 이후 같은 스레드에서 제출하는 Job은 pool1에 들어감

7.10 Summary

flowchart TD
  dag["DAGScheduler<br/>무엇을 어떤 순서로"] --> boundary["Shuffle 경계 탐색<br/>스택 DFS"]
  dag --> reuse["Stage 재사용<br/>shuffleIdToMapStage"]
  dag --> missing["findMissingPartitions<br/>MapOutputTracker"]
  dag --> fetch["FetchFailed 처리<br/>Stage 재제출"]
  ts["TaskScheduler / TaskSetManager<br/>어디서 실행할지"] --> locality["Delay Scheduling"]
  ts --> spec["Speculation"]
  ts --> retry["Task 재시도"]
  ts --> pool["FIFO / FAIR Pool"]
  dag -- "TaskSet" --> ts

• 한 줄 요약
  • Task가 그냥 실패하면 TaskScheduler가 다시 돌림
  • 데이터가 사라지면 DAGScheduler가 Stage를 다시 돌림
  • 재실행 범위를 최소화하는 장치는 MapOutputTracker와 findMissingPartitions()