# 機能設計書 3-DAGスケジューラ

## 概要

本ドキュメントは、Apache SparkのDAGスケジューラ機能の設計を記述する。DAGSchedulerはジョブをステージに分割し、RDD依存関係に基づく有向非巡回グラフ（DAG）を構築して、タスク実行順序を最適化する高レベルスケジューリング層である。

### 本機能の処理概要

DAGSchedulerは、ユーザーが発行したジョブ（RDDアクション）をステージ群に分割し、依存関係に基づくDAGを構築して、TaskSchedulerにタスクセットを送信する。

**業務上の目的・背景**：分散データ処理において、複数のRDD変換操作を効率的に実行するには、シャッフル境界でのステージ分割と、キャッシュの再利用、最適な実行順序の決定が不可欠である。DAGSchedulerはこれらを自動的に行い、ユーザーが個々のタスクの実行順序を意識することなく最適な分散処理を実現する。

**機能の利用シーン**：RDDアクション（count, collect等）の実行時にSparkContext.runJobから呼び出される。全てのSparkジョブ実行で利用される中核コンポーネントである。

**主要な処理内容**：
1. ジョブのDAGステージ分割（シャッフル境界での分割）
2. ResultStage（最終ステージ）とShuffleMapStage（中間ステージ）の構築
3. ステージの依存関係解析と実行順序の決定
4. キャッシュ済みRDDパーティションの追跡と再利用
5. シャッフルマップ出力の追跡と再利用
6. タスクの推奨実行場所（データローカリティ）の計算
7. シャッフル出力ファイル消失時のステージ再実行
8. ジョブのキャンセルとステージのアボート処理

**関連システム・外部連携**：TaskScheduler（下位スケジューラ）、BlockManager（キャッシュ管理）、MapOutputTracker（シャッフル出力追跡）

**権限による制御**：DAGSchedulerは内部コンポーネントであり、直接的な権限制御は持たない。ジョブの実行制御はSparkContext経由で行われる。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 1 | All Jobs（全ジョブ一覧） | 参照画面 | ジョブのステージ分割情報を参照し、各ジョブのステージ数を表示 |
| 2 | Job Detail（ジョブ詳細） | 参照画面 | ジョブのDAG可視化（operationGraphForJob）を表示 |
| 3 | All Stages（全ステージ一覧） | 参照画面 | FAIRスケジューラ使用時のプール情報を取得して表示 |
| 4 | Stage Detail（ステージ詳細） | 参照画面 | ステージのDAG可視化（operationGraphForStage）を表示 |
| 5 | Pool Detail（プール詳細） | 参照画面 | プールに属するステージのDAG情報を参照 |
| 15 | SQL Execution Detail（SQL実行詳細） | 遷移先 | 関連ジョブへのリンクを提供 |
| 19 | Batch Detail（バッチ詳細） | 遷移先 | バッチに関連するジョブ・ステージ詳細へのリンク |

## 機能種別

ジョブスケジューリング / DAG構築・最適化 / タスク配置決定

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| rdd | RDD[_] | Yes | 最終結果RDD | null不可 |
| func | (TaskContext, Iterator[_]) => _ | Yes | パーティション処理関数 | - |
| partitions | Seq[Int] | Yes | 処理対象パーティションのリスト | 有効なパーティションインデックス |
| callSite | CallSite | Yes | 呼び出し元情報 | - |
| resultHandler | (Int, _) => Unit | Yes | 結果処理関数 | - |
| properties | Properties | No | ジョブプロパティ（スケジューラプール等） | - |

### 入力データソース

SparkContext.runJob()からの呼び出し。内部的にはDAGSchedulerEventProcessLoopのイベントキューで処理される。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| JobSubmitted | イベント | ジョブ送信イベント |
| TaskSet | TaskSet | ステージ内のタスクセット（TaskSchedulerへ送信） |
| MapStatus | Array[MapStatus] | シャッフルマップ出力の場所情報 |

### 出力先

TaskScheduler（タスクセット送信）、MapOutputTracker（シャッフル出力登録）、SparkListenerBus（イベント通知）

## 処理フロー

### 処理シーケンス

```
1. ジョブ受信（submitJob/submitMapStage）
   └─ ActiveJobの生成とJobSubmittedイベントの投稿
2. 最終ステージの取得・生成
   └─ ResultStageまたはShuffleMapStageの取得
3. 親ステージの再帰的構築
   └─ シャッフル依存関係に基づきShuffleMapStageを生成
4. 実行可能ステージの特定
   └─ 全ての親ステージが完了済みのステージを特定
5. タスクセットの生成と送信
   └─ ステージ内の未完了パーティションからTaskSetを生成しTaskSchedulerへ送信
6. タスク完了の処理
   └─ ShuffleMapTask: MapStatus登録、ResultTask: 結果ハンドラ呼び出し
7. ステージ完了の判定
   └─ 全パーティション完了時にステージ完了、子ステージの実行を開始
8. ジョブ完了の処理
   └─ 最終ステージ完了時にJobSucceededイベント通知
```

### フローチャート

```mermaid
flowchart TD
    A[submitJob] --> B[ResultStage生成]
    B --> C[親ステージ再帰構築]
    C --> D[実行可能ステージ特定]
    D --> E[TaskSet生成・送信]
    E --> F{タスク完了?}
    F -->|ShuffleMap| G[MapStatus登録]
    F -->|Result| H[結果ハンドラ呼び出し]
    G --> I{ステージ完了?}
    H --> I
    I -->|Yes| J{子ステージあり?}
    I -->|No| E
    J -->|Yes| D
    J -->|No| K[ジョブ完了]
    K --> L[JobSucceeded通知]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | シャッフル境界分割 | wide dependency（ShuffleDependency）の境界でステージを分割 | 常時 |
| BR-02 | ステージ共有 | 同じRDDを使用する複数ジョブはステージを共有可能 | 同一RDD参照時 |
| BR-03 | キャッシュ再利用 | キャッシュ済みパーティションは再計算をスキップ | persist済みRDD |
| BR-04 | シャッフル出力再利用 | 完了済みシャッフルマップの出力は再利用 | シャッフルマップ完了済み |
| BR-05 | フェッチ障害時再実行 | シャッフル出力ファイル消失時は親ステージを再実行 | FetchFailedException発生時 |

### 計算ロジック

データローカリティ計算: `getPreferredLocs(rdd, partition)`により、キャッシュ場所→チェックポイント場所→親RDDの推奨場所の順で計算される。

## データベース操作仕様

### 操作別データベース影響一覧

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | DAGSchedulerはデータベース操作を行わない |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| FetchFailed | シャッフル障害 | シャッフル出力ファイル読み取り失敗 | 親ステージの再実行 |
| TaskKilled | キャンセル | ジョブ/ステージのキャンセル | リソース解放 |
| ExceptionFailure | タスク失敗 | タスク実行中の例外 | TaskSchedulerによるリトライ |
| StageAborted | ステージ失敗 | 最大リトライ回数超過 | ジョブ失敗としてユーザーに通知 |

### リトライ仕様

ステージのリトライは`spark.stage.maxConsecutiveAttempts`（デフォルト4）まで許可される。シャッフル出力消失時のステージ再実行は自動的に行われる。

## トランザクション仕様

DAGSchedulerイベントはDAGSchedulerEventProcessLoopで単一スレッドにて順次処理される。これによりイベント処理の順序性が保証される。

## パフォーマンス要件

- イベントループの処理遅延: ミリ秒オーダー
- ステージ分割処理: RDDグラフの深さに比例
- タスク推奨場所計算: キャッシュ情報のルックアップ時間に依存

## セキュリティ考慮事項

DAGSchedulerは内部コンポーネントであり、外部からの直接アクセスはない。タスク記述のシリアライズ時に、ブロードキャスト変数のバイナリデータが暗号化通信（spark.authenticate有効時）で転送される。

## 備考

- DAGSchedulerのイベント処理は専用のイベントループスレッド（dag-scheduler-event-loop）で実行される
- ステージIDとジョブIDは各SparkContext内で一意のシーケンシャルな整数が割り当てられる
- RDD_CACHE_VISIBILITY_TRACKING_ENABLED設定によりキャッシュの可視性追跡が制御される

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

DAGSchedulerが管理するジョブ・ステージ・タスクの構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ActiveJob.scala | `core/src/main/scala/org/apache/spark/scheduler/ActiveJob.scala` | ジョブの表現（jobId, finalStage, resultHandler） |
| 1-2 | Stage.scala | `core/src/main/scala/org/apache/spark/scheduler/Stage.scala` | ステージの基底クラス |
| 1-3 | ResultStage.scala | `core/src/main/scala/org/apache/spark/scheduler/ResultStage.scala` | 最終ステージ（アクション実行） |
| 1-4 | ShuffleMapStage.scala | `core/src/main/scala/org/apache/spark/scheduler/ShuffleMapStage.scala` | シャッフルマップステージ |
| 1-5 | DAGSchedulerEvent.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala` | イベント定義 |

**読解のコツ**: DAGSchedulerはイベント駆動型アーキテクチャを採用している。各操作（ジョブ送信、タスク完了等）はイベントとしてキューに投稿され、単一のイベントループスレッドで順次処理される。

#### Step 2: エントリーポイントを理解する

DAGSchedulerのジョブ送信処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | submitJob()メソッド、handleJobSubmitted()メソッド |

**主要処理フロー**:
- **58-100行目**: DAGSchedulerのScaladoc（設計概要の理解）
- submitJob(): JobWaiterの生成とJobSubmittedイベントの投稿
- handleJobSubmitted(): ResultStage生成、親ステージ構築、submitMissingTasks()呼び出し

#### Step 3: ステージ構築ロジックを理解する

シャッフル境界でのステージ分割ロジックを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | getOrCreateShuffleMapStage(), getMissingParentStages() |

#### Step 4: タスク実行と完了処理を理解する

タスクセット送信とタスク完了ハンドリングを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | submitMissingTasks(), handleTaskCompletion() |
| 4-2 | MapOutputTracker.scala | `core/src/main/scala/org/apache/spark/MapOutputTracker.scala` | シャッフル出力の場所追跡 |

### プログラム呼び出し階層図

```
SparkContext.runJob()
    |
    +-- DAGScheduler.submitJob()
            |
            +-- eventProcessLoop.post(JobSubmitted)
                    |
                    +-- handleJobSubmitted()
                            |
                            +-- createResultStage()
                            |       +-- getOrCreateParentStages()
                            |               +-- getShuffleDependencies()
                            |               +-- getOrCreateShuffleMapStage()
                            |
                            +-- submitMissingTasks()
                                    |
                                    +-- getPreferredLocs()
                                    +-- TaskScheduler.submitTasks(TaskSet)
```

### データフロー図

```
[入力]                    [処理]                         [出力]

RDD + Action      ───▶  submitJob()             ───▶  ActiveJob
                         |
                         +-- DAG構築               ───▶  Stage依存関係グラフ
                         |
                         +-- submitMissingTasks()  ───▶  TaskSet → TaskScheduler
                         |
TaskCompletion    ───▶  handleTaskCompletion()   ───▶  MapStatus / Result
                         |
                         +-- ステージ完了判定       ───▶  次ステージ実行 / ジョブ完了
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | ソース | DAGスケジューリングのメインクラス |
| DAGSchedulerEvent.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala` | ソース | イベント型定義 |
| DAGSchedulerSource.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala` | ソース | メトリクスソース |
| Stage.scala | `core/src/main/scala/org/apache/spark/scheduler/Stage.scala` | ソース | ステージ基底クラス |
| ResultStage.scala | `core/src/main/scala/org/apache/spark/scheduler/ResultStage.scala` | ソース | 結果ステージ |
| ShuffleMapStage.scala | `core/src/main/scala/org/apache/spark/scheduler/ShuffleMapStage.scala` | ソース | シャッフルマップステージ |
| ActiveJob.scala | `core/src/main/scala/org/apache/spark/scheduler/ActiveJob.scala` | ソース | アクティブジョブ |
| MapOutputTracker.scala | `core/src/main/scala/org/apache/spark/MapOutputTracker.scala` | ソース | シャッフル出力追跡 |
| JobWaiter.scala | `core/src/main/scala/org/apache/spark/scheduler/JobWaiter.scala` | ソース | ジョブ完了待機 |
| OutputCommitCoordinator.scala | `core/src/main/scala/org/apache/spark/scheduler/OutputCommitCoordinator.scala` | ソース | 出力コミット調整 |
