# 機能設計書 15-動的リソース割り当て

## 概要

本ドキュメントは、Apache Sparkの動的リソース割り当て機能について、ワークロードに応じたExecutor数の自動調整の設計仕様を記述した機能設計書である。

### 本機能の処理概要

ワークロードの変動に応じてExecutorの数を動的に増減させるリソースオートスケーリング機能を提供する。ペンディングタスクの蓄積に応じてExecutorを追加し、アイドル状態のExecutorを自動的に削除する。

**業務上の目的・背景**：固定数のExecutorでは、ワークロードが少ない時にリソースを無駄に消費し、ワークロードが多い時にリソース不足で処理遅延が発生する。動的リソース割り当てにより、コスト効率とパフォーマンスの最適なバランスを自動的に維持する。

**機能の利用シーン**：(1) 可変ワークロードのバッチ処理、(2) 対話型クエリセッション（アイドル時のリソース解放）、(3) マルチテナント環境でのリソース共有最適化、(4) クラウド環境でのコスト最適化。

**主要な処理内容**：
1. ExecutorAllocationManagerによる100msごとの定期的なスケジューリングループ
2. ペンディングタスク数に基づくExecutor追加（指数的バックオフ方式）
3. アイドルExecutorのタイムアウト検出と削除
4. ResourceProfile単位でのExecutor数管理
5. ExecutorAllocationListenerによるタスクイベント監視
6. ExecutorMonitorによるアイドル状態の追跡

**関連システム・外部連携**：クラスタマネージャ（YARN、Kubernetes、Standalone）、External Shuffle Service

**権限による制御**：特になし。設定値によるリソース制限（minExecutors, maxExecutors）で制御。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 10 | Executors（エグゼキュータ一覧） | 補助機能 | Executorの動的追加・削除イベント情報を表示 |
| 24 | Master Overview（マスター概要） | 補助機能 | クラスタ全体のCPUコア・メモリの使用率・残量を集計して表示 |

## 機能種別

リソース管理 / オートスケーリング

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| spark.dynamicAllocation.enabled | Boolean | No | 動的割り当ての有効化 | デフォルト: false |
| spark.dynamicAllocation.minExecutors | Int | No | Executor下限 | 0以上、maxExecutors以下 |
| spark.dynamicAllocation.maxExecutors | Int | No | Executor上限 | 0より大きい |
| spark.dynamicAllocation.initialExecutors | Int | No | 初期Executor数 | min-max範囲内 |
| spark.dynamicAllocation.executorAllocationRatio | Double | No | タスク数に対する割り当て比率 | 0 < ratio <= 1.0 |
| spark.dynamicAllocation.schedulerBacklogTimeout | Long | No | バックログタイムアウト(秒) | 0より大きい |
| spark.dynamicAllocation.sustainedSchedulerBacklogTimeout | Long | No | 持続バックログタイムアウト(秒) | 0より大きい |
| spark.dynamicAllocation.executorIdleTimeout | Long | No | アイドルタイムアウト(秒) | - |
| spark.dynamicAllocation.cachedExecutorIdleTimeout | Long | No | キャッシュ保持Executorのアイドルタイムアウト(秒) | - |

### 入力データソース

SparkListenerBusから配信されるタスク・ステージイベント（SparkListenerStageSubmitted, SparkListenerTaskStart, SparkListenerTaskEnd等）。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Executor追加要求 | Int | クラスタマネージャへのExecutor追加数 |
| Executor削除要求 | Seq[String] | 削除対象のExecutor IDリスト |
| メトリクス | Gauge/Counter | ExecutorAllocationManagerSourceのメトリクス |

### 出力先

クラスタマネージャ（ExecutorAllocationClient経由）。MetricsSystemへのメトリクス報告。

## 処理フロー

### 処理シーケンス

```
1. ExecutorAllocationManagerの初期化
   └─ 設定値のバリデーション、初期Executor数の要求
2. リスナー登録
   └─ ExecutorAllocationListenerとExecutorMonitorをListenerBusに登録
3. 定期スケジューリング（100ms間隔）
   └─ schedule()でExecutor数の調整とアイドルExecutorの削除を実行
4. Executor追加判定
   └─ バックログタイムアウト超過時、指数的にExecutor追加数を増加
5. Executor削除判定
   └─ ExecutorMonitorがアイドルタイムアウト超過のExecutorを報告
6. クラスタマネージャへの要求
   └─ requestTotalExecutors()で目標Executor数を同期
```

### フローチャート

```mermaid
flowchart TD
    A[schedule 100ms間隔] --> B[timedOutExecutors取得]
    B --> C{アイドルExecutorあり?}
    C -->|Yes| D[initializing = false]
    C -->|No| E[次の判定へ]
    D --> E
    E --> F[updateAndSyncNumExecutorsTarget]
    F --> G{initializing?}
    G -->|Yes| H[変更なし]
    G -->|No| I{maxNeeded < target?}
    I -->|Yes| J[decrementExecutors]
    I -->|No| K{addTime超過?}
    K -->|Yes| L[addExecutors 指数増加]
    K -->|No| M[変更なし]
    J --> N[requestTotalExecutors]
    L --> N
    N --> O{アイドルExecutorあり?}
    O -->|Yes| P[removeExecutors]
    O -->|No| Q[完了]
    P --> Q
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-15-01 | 指数的増加 | Executor追加数は1, 2, 4, 8...と指数的に増加 | バックログ持続時 |
| BR-15-02 | 上限制約 | maxExecutorsを超えるExecutor追加は行わない | 常時 |
| BR-15-03 | 下限制約 | minExecutors未満へのExecutor削除は行わない | 常時 |
| BR-15-04 | ResourceProfile単位管理 | Executor数はResourceProfile単位で独立管理 | 全操作 |
| BR-15-05 | 初期化期間 | 最初のステージ提出またはアイドルタイムアウトまでExecutor追加を抑制 | 起動直後 |
| BR-15-06 | Decommission対応 | spark.decommission.enabledがtrueの場合、kill代わりにdecommissionを実行 | Executor削除時 |
| BR-15-07 | シャッフルサービス要件 | External Shuffle Serviceまたはシャッフルトラッキングが必要 | 有効化条件 |

### 計算ロジック

maxNumExecutorsNeeded = ceil(numRunningOrPendingTasks * executorAllocationRatio / tasksPerExecutor)

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

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

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

### テーブル別操作詳細

該当なし

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | SparkException | minExecutors < 0 または maxExecutors < 0 | 起動時に例外スロー |
| - | SparkException | maxExecutors == 0 | 起動時に例外スロー |
| - | SparkException | minExecutors > maxExecutors | 起動時に例外スロー |
| - | SparkException | シャッフルサービス未有効 | 起動時に例外スロー（テストモード除外） |
| - | NonFatal | クラスタマネージャ到達不可 | logInfoで記録、ターゲット数をロールバック |

### リトライ仕様

クラスタマネージャへの要求が失敗した場合、次のスケジューリングサイクル（100ms後）で再試行される。明示的なリトライロジックは持たない。

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

該当なし。スケジューリングループはsynchronizedブロックで排他制御される。

## パフォーマンス要件

スケジューリングループは100ms間隔で実行。Executorの追加・削除はクラスタマネージャの応答速度に依存する。

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

特になし。クラスタマネージャのセキュリティ設定に依存。

## 備考

- Streaming用のExecutorAllocationManager（streaming/scheduler/ExecutorAllocationManager.scala）も別途存在する
- ExecutorMonitor（scheduler/dynalloc/ExecutorMonitor.scala）がアイドル検出の中核
- YARN AMの再起動時にはreset()で状態をリセット

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ExecutorAllocationManager.scala | `core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala` | numExecutorsTargetPerResourceProfileId, numExecutorsToAddPerResourceProfileIdのHashMap構造 |

**読解のコツ**: ResourceProfile IDをキーとしたHashMapで各ResourceProfileのExecutor数を独立管理する点に注目。`synchronized`ブロックによる排他制御も重要。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ExecutorAllocationManager.scala | `core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala` | start(), schedule(), updateAndSyncNumExecutorsTarget()の流れ |

**主要処理フロー**:
1. **102-109行目**: コンストラクタでExecutorAllocationClient等を受け取る
2. **117-119行目**: minNumExecutors, maxNumExecutors, initialNumExecutorsの読み込み
3. **189-239行目**: validateSettings()で設定値を検証
4. **245-266行目**: start()でリスナー登録とスケジューリングタスク開始
5. **341-352行目**: schedule()でtimedOutExecutors取得→updateAndSync→removeExecutors
6. **367-394行目**: updateAndSyncNumExecutorsTarget()でResourceProfileごとにターゲット更新
7. **506-535行目**: addExecutors()で指数的増加ロジック

#### Step 3: タスクイベント監視層を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ExecutorAllocationListener (inner class) | `core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala` | onStageSubmitted, onTaskStart, onTaskEndの実装 |

**主要処理フロー**:
- **687-732行目**: onStageSubmitted()でバックログ通知、ResourceProfile管理
- **762-784行目**: onTaskStart()でランニングタスク数更新
- **786-827行目**: onTaskEnd()でタスク完了・失敗処理

#### Step 4: アイドル監視層を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | ExecutorMonitor.scala | `core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala` | timedOutExecutors()、executorCountWithResourceProfile()の実装 |

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

```
SparkContext
    │
    ├─ ExecutorAllocationManager
    │      ├─ start()
    │      │    ├─ listenerBus.addToManagementQueue(listener)
    │      │    ├─ listenerBus.addToManagementQueue(executorMonitor)
    │      │    └─ executor.scheduleWithFixedDelay(schedule, 0, 100ms)
    │      │
    │      ├─ schedule() ─ 100ms間隔
    │      │    ├─ executorMonitor.timedOutExecutors()
    │      │    ├─ updateAndSyncNumExecutorsTarget()
    │      │    │    ├─ maxNumExecutorsNeededPerResourceProfile()
    │      │    │    ├─ addExecutors() / decrementExecutors()
    │      │    │    └─ client.requestTotalExecutors()
    │      │    └─ removeExecutors()
    │      │         └─ client.killExecutors() / client.decommissionExecutors()
    │      │
    │      ├─ ExecutorAllocationListener
    │      │    ├─ onStageSubmitted() → onSchedulerBacklogged()
    │      │    ├─ onTaskStart() → onSchedulerQueueEmpty() (if no pending)
    │      │    └─ onTaskEnd() → onSchedulerBacklogged() (if failed)
    │      │
    │      └─ ExecutorAllocationManagerSource (メトリクス)
    │
    └─ ExecutorMonitor
         └─ timedOutExecutors() ─ アイドルExecutor検出
```

### データフロー図

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

SparkListenerEvent    ───▶ ExecutorAllocationListener  ───▶ backlog状態
(タスク・ステージ)          └─ pending/runningタスク数算出

backlog状態 + 時間   ───▶ schedule()                  ───▶ Executor追加/削除要求
                          ├─ 指数的増加ロジック              → ClusterManager
                          └─ アイドル検出

ExecutorMonitor      ───▶ timedOutExecutors()         ───▶ 削除対象リスト
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ExecutorAllocationManager.scala | `core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala` | ソース | 動的割り当てのメインクラス（Listener含む） |
| ExecutorMonitor.scala | `core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala` | ソース | Executorアイドル監視 |
| ExecutorAllocationClient.scala | `core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala` | ソース | クラスタマネージャへの要求インターフェース |
| ResourceProfileManager.scala | `core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala` | ソース | ResourceProfile管理 |
