# 機能設計書 4-タスクスケジューラ

## 概要

本ドキュメントは、Apache SparkのTaskScheduler機能の設計を記述する。TaskSchedulerはDAGSchedulerが生成したTaskSetをクラスタ上のExecutorに割り当て、データローカリティを考慮した実行スケジューリングを行う低レベルスケジューリング層である。

### 本機能の処理概要

TaskSchedulerImplは、DAGSchedulerから受け取ったタスクセットをクラスタ上のExecutorリソースに割り当て、データローカリティ最適化、遅延スケジューリング、投機的実行、FIFO/FAIRスケジューリングポリシーを実装する。

**業務上の目的・背景**：分散クラスタ上で大量のタスクを効率的に実行するには、リソースの有効活用とデータローカリティの最適化が不可欠である。TaskSchedulerはExecutorからのリソースオファーに対して最適なタスク割り当てを行い、クラスタ全体のスループットを最大化する。

**機能の利用シーン**：DAGSchedulerからタスクセットが送信されるたびにTaskSchedulerが呼び出される。全てのSparkジョブ実行において、タスクレベルのスケジューリングを担当する。

**主要な処理内容**：
1. タスクセットの受信とTaskSetManagerの生成
2. リソースオファーに基づくタスクのExecutor割り当て（resourceOffers）
3. データローカリティ（PROCESS_LOCAL→NODE_LOCAL→RACK_LOCAL→ANY）に基づく配置最適化
4. 遅延スケジューリングによるデータローカリティの改善
5. FIFOまたはFAIRスケジューリングポリシーの適用
6. タスク完了・失敗のステータス管理
7. 投機的タスク（speculative task）の起動
8. ヘルスチェック・ブロックリスト管理

**関連システム・外部連携**：SchedulerBackend（Standalone/YARN/K8s/Local）、DAGScheduler、Executor

**権限による制御**：TaskSchedulerは内部コンポーネントであり、直接的な権限制御は持たない。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 3 | All Stages（全ステージ一覧） | 参照画面 | 各ステージのタスク割り当て状況・進捗情報を取得して表示 |
| 4 | Stage Detail（ステージ詳細） | 参照画面 | タスクのローカリティ別サマリー・実行時間内訳を取得して表示 |
| 5 | Pool Detail（プール詳細） | 参照画面 | FAIRスケジューラのプール設定と進捗を取得・表示 |

## 機能種別

タスクスケジューリング / リソース割り当て / データローカリティ最適化

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| sc | SparkContext | Yes | SparkContext参照 | null不可 |
| maxTaskFailures | Int | Yes | タスク最大失敗回数 | 1以上 |
| isLocal | Boolean | No | ローカルモードフラグ | デフォルトfalse |
| taskSet | TaskSet | Yes | DAGSchedulerから受信するタスクセット | null不可 |

### 入力データソース

- DAGScheduler.submitMissingTasks()からのTaskSet送信
- SchedulerBackendからのリソースオファー（WorkerOffer）
- Executorからのタスクステータスアップデート

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| TaskDescription | Seq[Seq[TaskDescription]] | Executorごとに割り当てられたタスク記述 |
| TaskResult | DirectTaskResult / IndirectTaskResult | タスク実行結果 |
| StatusUpdate | TaskState | タスク状態の更新通知 |

### 出力先

SchedulerBackend（タスク起動指示）、DAGScheduler（タスク完了通知）

## 処理フロー

### 処理シーケンス

```
1. タスクセット受信（submitTasks）
   └─ TaskSetManagerの生成とスケジューリングプールへの登録
2. リソースオファー受信（resourceOffers）
   └─ SchedulerBackendからのExecutorリソースオファー処理
3. タスク割り当て
   └─ ローカリティレベル順（PROCESS_LOCAL→NODE_LOCAL→RACK_LOCAL→ANY）でタスクを割り当て
4. 遅延スケジューリング判定
   └─ より良いローカリティを得るためにタスク起動を遅延させるか判定
5. タスク起動
   └─ TaskDescriptionを生成しSchedulerBackendへ送信
6. タスク完了処理（statusUpdate）
   └─ 成功: DAGSchedulerにTaskSucceeded通知
   └─ 失敗: リトライまたはDAGSchedulerにTaskFailed通知
7. 投機的実行チェック
   └─ 定期的に実行が遅いタスクを検出し、投機的タスクを起動
```

### フローチャート

```mermaid
flowchart TD
    A[submitTasks: TaskSet受信] --> B[TaskSetManager生成]
    B --> C[スケジューリングプールに登録]
    C --> D[resourceOffers: リソースオファー受信]
    D --> E[ローカリティレベル判定]
    E --> F{PROCESS_LOCAL割り当て可能?}
    F -->|Yes| G[タスク割り当て]
    F -->|No| H{NODE_LOCAL割り当て可能?}
    H -->|Yes| G
    H -->|No| I{遅延スケジューリング?}
    I -->|Yes| J[割り当て保留]
    I -->|No| K[ANY割り当て]
    G --> L[TaskDescription生成・起動]
    K --> L
    L --> M[statusUpdate受信]
    M --> N{成功?}
    N -->|Yes| O[DAGScheduler通知]
    N -->|No| P{リトライ可能?}
    P -->|Yes| D
    P -->|No| Q[ステージ失敗]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | データローカリティ優先 | PROCESS_LOCAL→NODE_LOCAL→RACK_LOCAL→ANYの順で配置を最適化 | 常時 |
| BR-02 | 遅延スケジューリング | spark.locality.wait設定に基づきデータローカリティ改善のために起動を遅延 | 常時 |
| BR-03 | タスクリトライ | spark.task.maxFailures回（デフォルト4）まで失敗タスクをリトライ | タスク失敗時 |
| BR-04 | 投機的実行 | spark.speculation=true時、遅いタスクの投機的コピーを起動 | 設定有効時 |
| BR-05 | FIFO/FAIRスケジューリング | spark.scheduler.modeでFIFOまたはFAIRポリシーを選択 | 常時 |

### 計算ロジック

遅延スケジューリング: TaskSetManagerが最後にタスクを起動してからの経過時間がspark.locality.wait（デフォルト3秒）を超えた場合、次のローカリティレベルにフォールバックする。

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

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TaskKilledException | キャンセル | タスクのキャンセル要求 | タスク中断、リソース解放 |
| ExecutorLostFailure | Executor障害 | Executorプロセスの消失 | 実行中タスクの再スケジューリング |
| TaskResultLost | 結果消失 | タスク結果の取得失敗 | タスクの再実行 |
| SparkCoreErrors | 内部エラー | スケジューリング論理エラー | ステージアボート |

### リトライ仕様

タスクの失敗は`spark.task.maxFailures`（デフォルト4）回までリトライされる。同一Executorでの連続失敗はブロックリスト（HealthTracker）で管理され、一定回数以上失敗したExecutorやノードはタスク割り当て対象から除外される。

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

TaskSchedulerImplの主要メソッド（submitTasks, resourceOffers, statusUpdate等）はsynchronizedブロックで保護され、マルチスレッドからの同時アクセスに対する安全性が確保される。

## パフォーマンス要件

- resourceOffers処理: ミリ秒オーダーで完了すべき
- 遅延スケジューリングの待機時間: spark.locality.wait（デフォルト3秒）
- 投機的実行チェック間隔: spark.speculation.interval（デフォルト100ms）

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

- タスク記述にはシリアライズされたタスクバイナリが含まれ、spark.authenticate有効時は暗号化通信で転送される
- タスク実行に必要なJAR/ファイルのフェッチにはセキュリティマネージャの認証が適用される

## 備考

- TaskSchedulerImplはTaskSchedulerトレイトの標準実装であり、全てのクラスタマネージャで共通
- SchedulerBackendがクラスタマネージャ固有の実装を提供（StandaloneSchedulerBackend、YarnSchedulerBackend等）
- FAIRスケジューラはfairscheduler.xml設定ファイルでプール定義を行う

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | TaskSet.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSet.scala` | タスクセットの定義（tasks, stageId, stageAttemptId） |
| 1-2 | Task.scala | `core/src/main/scala/org/apache/spark/scheduler/Task.scala` | タスクの基底クラス |
| 1-3 | TaskDescription.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala` | Executorに送信されるタスク記述 |
| 1-4 | WorkerOffer.scala | `core/src/main/scala/org/apache/spark/scheduler/WorkerOffer.scala` | Executorリソースオファー |
| 1-5 | SchedulingMode.scala | `core/src/main/scala/org/apache/spark/scheduler/SchedulingMode.scala` | FIFO/FAIR/NONEの定義 |

**読解のコツ**: TaskSchedulerImplのScaladoc（46-82行目）にスレッディングモデルと遅延スケジューリングの詳細が記載されている。複数スレッドからのアクセスパターンを理解することが重要。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | TaskSchedulerImpl.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala` | submitTasks(), resourceOffers(), statusUpdate() |

**主要処理フロー**:
- **83-88行目**: クラス定義（SparkContext, maxTaskFailures, isLocal, clock）
- submitTasks(): TaskSetManagerの生成、スケジューラブルビルダーへの追加
- resourceOffers(): Executorオファーに対するタスク割り当て、ローカリティ最適化
- statusUpdate(): タスク完了/失敗の処理

#### Step 3: タスク割り当てロジックを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TaskSetManager.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala` | タスクセット単位の管理（ローカリティ判定、投機的実行） |
| 3-2 | TaskLocality.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskLocality.scala` | ローカリティレベル定義 |

#### Step 4: スケジューリングポリシーを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SchedulableBuilder.scala | `core/src/main/scala/org/apache/spark/scheduler/SchedulableBuilder.scala` | FIFO/FAIRスケジューラビルダー |
| 4-2 | Pool.scala | `core/src/main/scala/org/apache/spark/scheduler/Pool.scala` | スケジューリングプール |
| 4-3 | SchedulingAlgorithm.scala | `core/src/main/scala/org/apache/spark/scheduler/SchedulingAlgorithm.scala` | FIFO/FAIRアルゴリズム |

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

```
DAGScheduler.submitMissingTasks()
    |
    +-- TaskSchedulerImpl.submitTasks(TaskSet)
            |
            +-- TaskSetManager(taskSet)
            +-- SchedulableBuilder.addTaskSetManager()
            +-- SchedulerBackend.reviveOffers()
                    |
                    +-- TaskSchedulerImpl.resourceOffers(WorkerOffer[])
                            |
                            +-- Pool.getSortedTaskSetQueue()
                            +-- TaskSetManager.resourceOffer()
                                    |
                                    +-- TaskLocality判定
                                    +-- 遅延スケジューリング判定
                            |
                            +-- TaskDescription生成
                            +-- SchedulerBackend.launchTasks()
```

### データフロー図

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

TaskSet            ───▶  submitTasks()            ───▶  TaskSetManager登録
(DAGSchedulerから)

WorkerOffer[]      ───▶  resourceOffers()         ───▶  TaskDescription[]
(SchedulerBackendから)       |                           (Executorへ送信)
                             +-- ローカリティ最適化
                             +-- 遅延スケジューリング

StatusUpdate       ───▶  statusUpdate()           ───▶  DAGScheduler通知
(Executorから)               |
                             +-- 成功/失敗判定
                             +-- リトライ管理
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| TaskSchedulerImpl.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala` | ソース | TaskSchedulerの標準実装 |
| TaskScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskScheduler.scala` | ソース | TaskSchedulerトレイト |
| TaskSetManager.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala` | ソース | タスクセット管理 |
| TaskSet.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSet.scala` | ソース | タスクセット定義 |
| Task.scala | `core/src/main/scala/org/apache/spark/scheduler/Task.scala` | ソース | タスク基底クラス |
| TaskDescription.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala` | ソース | タスク記述 |
| TaskLocality.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskLocality.scala` | ソース | ローカリティレベル |
| Pool.scala | `core/src/main/scala/org/apache/spark/scheduler/Pool.scala` | ソース | スケジューリングプール |
| SchedulableBuilder.scala | `core/src/main/scala/org/apache/spark/scheduler/SchedulableBuilder.scala` | ソース | スケジューラビルダー |
| SchedulingAlgorithm.scala | `core/src/main/scala/org/apache/spark/scheduler/SchedulingAlgorithm.scala` | ソース | スケジューリングアルゴリズム |
| HealthTracker.scala | `core/src/main/scala/org/apache/spark/scheduler/HealthTracker.scala` | ソース | ヘルスチェック・ブロックリスト |
| SchedulerBackend.scala | `core/src/main/scala/org/apache/spark/scheduler/SchedulerBackend.scala` | ソース | バックエンドインターフェース |
