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

## 概要

本ドキュメントは、Apache Flinkのスケジューラ機能（flink-runtime モジュール内SchedulerNG）の設計仕様を記述する。スケジューラ機能は、ジョブのスケジューリング戦略とタスク配置を担当する。

### 本機能の処理概要

SchedulerNGは、Flinkジョブの実行計画からタスクの実行順序と配置を決定するコンポーネント。複数のスケジューリング戦略（Default、Adaptive、AdaptiveBatch）を提供する。

**業務上の目的・背景**：分散処理環境において、リソースの効率的な利用とジョブの性能最大化を実現するために、適切なタスク配置とスケジューリング戦略が必要。ストリーミングとバッチで異なる最適化を提供。

**機能の利用シーン**：
- ジョブ開始時のタスクスケジューリング
- タスク障害時のリスケジュール
- リソース変動時の適応的なスケーリング（Adaptive）
- バッチジョブの段階的実行（AdaptiveBatch）

**主要な処理内容**：
1. ExecutionGraphの作成と管理
2. タスク実行順序の決定
3. スロット要求とタスク配置
4. 障害検出とフェイルオーバー
5. チェックポイント調整への連携

**関連システム・外部連携**：
- JobMaster（ジョブ管理）
- SlotPool（スロット取得）
- CheckpointCoordinator（チェックポイント）
- ExecutionGraph（実行グラフ）

**権限による制御**：JobMaster内でのみインスタンス化。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 6 | ジョブ概要（DAGグラフ） | 関連機能 | 実行計画表示 |

## 機能種別

スケジューリング / リソース最適化

## 入力仕様

### 入力パラメータ

#### スケジューラ作成

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| executionPlan | ExecutionPlan | Yes | 実行計画 | 有効なDAG |
| configuration | Configuration | Yes | 設定 | null不可 |
| slotPoolService | SlotPoolService | Yes | スロットプール | null不可 |

#### タスク状態更新

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| taskExecutionState | TaskExecutionStateTransition | Yes | タスク状態 | 有効な状態遷移 |

### 入力データソース

- JobMasterからの制御指示
- TaskExecutorからの状態更新
- CheckpointCoordinatorからのACK

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| ExecutionGraphInfo | ExecutionGraphInfo | 実行グラフ情報 |
| JobStatus | JobStatus | ジョブステータス |
| CheckpointStatsSnapshot | CheckpointStatsSnapshot | チェックポイント統計 |

### 出力先

- JobMaster（ジョブ状態）
- SlotPool（スロット要求）
- TaskExecutor（タスクデプロイ）

## 処理フロー

### 処理シーケンス

```
1. スケジューラ初期化
   └─ ExecutionGraph作成
   └─ フェイルオーバー戦略設定

2. スケジューリング開始
   └─ startScheduling() 呼び出し
   └─ スケジューリング戦略に従い実行

3. スロット要求
   └─ SlotPoolへスロット要求
   └─ スロット取得時にタスクデプロイ

4. タスク状態監視
   └─ updateTaskExecutionState() で状態受信
   └─ 必要に応じてリスケジュール

5. チェックポイント連携
   └─ triggerCheckpoint() でトリガー
   └─ acknowledgeCheckpoint() でACK処理

6. ジョブ完了/終了
   └─ 全タスク完了でジョブ完了
   └─ リソース解放
```

### フローチャート

```mermaid
flowchart TD
    A[ExecutionGraph作成] --> B[startScheduling]
    B --> C{スケジューラ種別}
    C -->|Default| D[全タスク同時スケジュール]
    C -->|Adaptive| E[リソースに応じてスケジュール]
    C -->|AdaptiveBatch| F[段階的スケジュール]

    D --> G[スロット要求]
    E --> G
    F --> G

    G --> H{スロット取得?}
    H -->|Yes| I[タスクデプロイ]
    H -->|No| J[待機]
    J --> H

    I --> K[タスク実行監視]
    K --> L{タスク状態}
    L -->|FINISHED| M{全タスク完了?}
    L -->|FAILED| N[フェイルオーバー]
    M -->|Yes| O[ジョブ完了]
    M -->|No| K
    N --> G
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | 依存関係順序 | DAGの依存関係に従ってスケジュール | 全スケジューラ |
| BR-002 | スロット共有 | 同一スロット共有グループ内はスロット共有 | 設定時 |
| BR-003 | ローカリティ優先 | 可能な限りデータローカリティを優先 | タスク配置時 |
| BR-004 | フェイルオーバー戦略 | 設定された戦略でリスケジュール | 障害時 |

### 計算ロジック

- パイプライン領域の計算
- フェイルオーバー領域の決定

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

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

本機能は直接データベースを操作しない。

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| FlinkException | システムエラー | スケジュール失敗 | ジョブ再サブミット |
| GlobalFailure | 重大エラー | グローバル障害 | ジョブ再起動 |
| TimeoutException | タイムアウト | スロット取得タイムアウト | リソース確認 |

### リトライ仕様

- タスク障害: フェイルオーバー戦略に従いリトライ
- スロット取得失敗: タイムアウトまで待機

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

スケジューリング決定はアトミックに実行。

## パフォーマンス要件

- スケジューリング決定: ミリ秒オーダー
- タスクデプロイ: 数百ミリ秒

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

- スケジューラへの直接アクセスは不可（JobMaster経由）

## 備考

- DefaultScheduler: 標準的なストリーミングジョブ向け
- AdaptiveScheduler: Reactive Mode対応、動的スケーリング
- AdaptiveBatchScheduler: バッチジョブの効率的実行

---

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

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

### 推奨読解順序

#### Step 1: インターフェースを理解する

SchedulerNGのインターフェースを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SchedulerNG.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java` | メインインターフェース |
| 1-2 | GlobalFailureHandler.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/GlobalFailureHandler.java` | グローバル障害ハンドラ |

**読解のコツ**:
- **74行目**: SchedulerNGはGlobalFailureHandlerを継承
- **76行目**: startScheduling()でスケジューリング開始
- **82-86行目**: updateTaskExecutionState()でタスク状態更新

#### Step 2: 実装を理解する

各スケジューラ実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SchedulerBase.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java` | 基底クラス |
| 2-2 | DefaultScheduler.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java` | デフォルト実装 |
| 2-3 | AdaptiveScheduler.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java` | アダプティブ実装 |

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

```
SchedulerNG (interface)
    │
    ├─ startScheduling()
    │      └─ スケジューリング戦略実行
    │
    ├─ updateTaskExecutionState()
    │      └─ ExecutionGraph更新
    │             └─ 必要に応じてリスケジュール
    │
    ├─ triggerCheckpoint()
    │      └─ CheckpointCoordinator.triggerCheckpoint()
    │
    ├─ acknowledgeCheckpoint()
    │      └─ CheckpointCoordinator.acknowledgeCheckpoint()
    │
    └─ requestJob()
           └─ ExecutionGraphInfo返却
```

### データフロー図

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

JobGraph ───────────────▶ SchedulerNG ──────────────▶ タスクデプロイ
                              │
                              ▼
                       ExecutionGraph
                              │
           ┌──────────────────┼──────────────────┐
           │                  │                  │
           ▼                  ▼                  ▼
    SchedulingStrategy   FailoverStrategy   CheckpointCoordinator
           │                  │                  │
           ▼                  ▼                  ▼
       スロット要求       リスケジュール      チェックポイント
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SchedulerNG.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java` | ソース | インターフェース |
| SchedulerBase.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java` | ソース | 基底クラス |
| DefaultScheduler.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java` | ソース | デフォルト実装 |
| AdaptiveScheduler.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java` | ソース | アダプティブ実装 |
| AdaptiveBatchScheduler.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java` | ソース | バッチ実装 |
| SchedulerNGFactory.java | `flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNGFactory.java` | ソース | ファクトリ |
