# 機能設計書 5-Executorプロセス管理

## 概要

本ドキュメントは、Apache SparkのExecutorプロセス管理機能の設計を記述する。Executorは各ワーカーノード上でタスクを実行するプロセスであり、タスクの実行管理、メトリクスの収集、ドライバーへの報告を担当する。

### 本機能の処理概要

Executorクラスは、ドライバーから送信されたタスクをスレッドプール上で実行し、結果の返却とメトリクスの収集・報告を行うワーカーノード側の中核コンポーネントである。

**業務上の目的・背景**：分散処理においてタスクを実際に実行するのはExecutorプロセスである。タスクの効率的な実行、メモリ管理、ハートビートによる死活監視、メトリクスの収集により、クラスタ全体の安定稼働と性能監視を実現する。

**機能の利用シーン**：Sparkアプリケーション実行中、クラスタの各ワーカーノード上でExecutorプロセスが起動され、ドライバーから割り当てられたタスクを継続的に実行する。

**主要な処理内容**：
1. タスクの受信とスレッドプール上での実行（launchTask）
2. タスク結果のシリアライズとドライバーへの返却
3. ハートビートの送信によるドライバーとの死活監視
4. タスクメトリクス（実行時間、GC時間、メモリ使用量等）の収集
5. セッション分離（IsolatedSessionState）によるマルチテナント対応
6. クラスローダー管理とJAR/ファイルの動的更新
7. スレッドダンプ・ヒープヒストグラムの取得（デバッグ支援）

**関連システム・外部連携**：ドライバー（RPC通信）、BlockManager（データストレージ）、MetricsSystem

**権限による制御**：Executor自体は権限制御を持たないが、セキュリティマネージャにより通信の認証・暗号化が行われる。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 1 | All Jobs（全ジョブ一覧） | 補助機能 | Executorのタイムライン表示 |
| 2 | Job Detail（ジョブ詳細） | 補助機能 | Executorタイムライン表示、ステージのKill操作時のExecutor連携 |
| 6 | Task Thread Dump | 主機能 | SparkContext.getTaskThreadDumpを呼び出しタスクスレッド情報取得 |
| 8 | RDD Detail（RDD詳細） | 補助機能 | パーティションのExecutor配置情報表示 |
| 10 | Executors（エグゼキュータ一覧） | 主機能 | 各Executorの統計情報取得・表示 |
| 11 | Executor Thread Dump | 主機能 | SparkContext.getExecutorThreadDumpで全スレッドダンプ取得 |
| 12 | Executor Heap Histogram | 主機能 | SparkContext.getExecutorHeapHistogramでヒープ情報取得 |
| 25 | Application Detail | 補助機能 | ExecutorのState・リソース使用状況表示 |
| 28 | Worker Overview | 補助機能 | ワーカー上のExecutorRunner情報表示 |

## 機能種別

タスク実行管理 / メトリクス収集 / 死活監視

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| executorId | String | Yes | Executor一意識別子 | 空文字列不可 |
| executorHostname | String | Yes | Executorホスト名 | 有効なホスト名 |
| env | SparkEnv | Yes | Spark実行環境 | null不可 |
| isLocal | Boolean | Yes | ローカルモードフラグ | - |
| taskData | ByteBuffer | Yes | シリアライズされたタスクデータ | - |

### 入力データソース

ドライバーからRPC経由で送信されるTaskDescription（シリアライズされたタスクバイナリ）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| DirectTaskResult | DirectTaskResult | 小さなタスク結果（ドライバーに直接送信） |
| IndirectTaskResult | IndirectTaskResult | 大きなタスク結果（BlockManagerに保存、参照のみ送信） |
| Heartbeat | Heartbeat | ハートビートメッセージ（アキュムレータ更新を含む） |
| ExecutorMetrics | ExecutorMetrics | Executorレベルのメトリクス |

### 出力先

ドライバー（RPC経由）、BlockManager（大きなタスク結果の保存）

## 処理フロー

### 処理シーケンス

```
1. Executor初期化
   └─ スレッドプール生成、ハートビートスレッド開始
2. タスク受信（launchTask）
   └─ TaskRunnerの生成とスレッドプールへの送信
3. タスク実行（TaskRunner.run）
   └─ タスクのデシリアライズ→実行→結果のシリアライズ
4. 結果送信
   └─ 結果サイズに応じてDirect/Indirectで送信
5. メトリクス収集
   └─ タスクメトリクス（実行時間、GC時間等）の収集
6. ハートビート送信
   └─ 定期的にドライバーへハートビートとメトリクスを送信
7. タスク完了/失敗処理
   └─ ドライバーへステータスアップデート送信
```

### フローチャート

```mermaid
flowchart TD
    A[launchTask: タスク受信] --> B[TaskRunner生成]
    B --> C[スレッドプールで実行]
    C --> D[タスクデシリアライズ]
    D --> E[task.run実行]
    E --> F{成功?}
    F -->|Yes| G[結果シリアライズ]
    G --> H{結果サイズ > maxDirectResultSize?}
    H -->|Yes| I[BlockManagerに保存、IndirectTaskResult送信]
    H -->|No| J[DirectTaskResult送信]
    F -->|No| K[例外をシリアライズ]
    K --> L[StatusUpdate: FAILED送信]
    I --> M[StatusUpdate: FINISHED送信]
    J --> M
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | 結果サイズ判定 | maxDirectResultSize（デフォルト1MB）超過時はBlockManager経由で返却 | タスク結果送信時 |
| BR-02 | maxResultSize制限 | spark.driver.maxResultSize（デフォルト1GB）を超える結果は破棄 | タスク結果送信時 |
| BR-03 | ハートビート必須 | spark.executor.heartbeatInterval間隔でハートビートを送信 | 常時 |
| BR-04 | セッション分離 | IsolatedSessionStateによりセッションごとのクラスローダーを管理 | マルチテナント時 |

### 計算ロジック

タスクメトリクス収集: executorDeserializeTime, executorRunTime, executorCpuTime, jvmGCTime, resultSerializationTime, memoryBytesSpilled, diskBytesSpilled, peakExecutionMemory等のメトリクスを収集。

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

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| OutOfMemoryError | メモリエラー | JVMヒープメモリ不足 | Executorプロセスの再起動 |
| FetchFailedException | シャッフルエラー | シャッフルデータ取得失敗 | DAGSchedulerへ通知、親ステージ再実行 |
| TaskKilledException | キャンセル | タスクのキャンセル要求 | タスク中断、リソース解放 |
| CommitDeniedException | コミット拒否 | OutputCommitCoordinatorがコミットを拒否 | タスク結果破棄 |

### リトライ仕様

Executor自体のリトライはクラスタマネージャが管理する。タスクレベルのリトライはTaskSchedulerが管理する。

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

各タスクは独立して実行され、タスク間のトランザクション管理は行わない。タスク結果の送信はアトミックに行われる。

## パフォーマンス要件

- タスクデシリアライズ時間: executorDeserializeTimeメトリクスで追跡
- ハートビート間隔: spark.executor.heartbeatInterval（デフォルト10秒）
- スレッドプールサイズ: spark.executor.coresで制御

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

- ドライバーとの通信はspark.authenticate有効時に認証・暗号化される
- タスクデータはシリアライズされた状態で転送される
- IsolatedSessionStateによりセッション間の分離が確保される

## 備考

- Executorは1つ以上のコアを持ち、各コアで1タスクを並列実行する
- CoarseGrainedExecutorBackendがExecutorのRPC通信を担当する
- ローカルモードではドライバーと同一JVMでExecutorが実行される

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | TaskMetrics.scala | `core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala` | タスクメトリクスの定義 |
| 1-2 | ExecutorMetrics.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorMetrics.scala` | Executorメトリクスの定義 |
| 1-3 | IsolatedSessionState | `core/src/main/scala/org/apache/spark/executor/Executor.scala` 61-100行目 | セッション分離状態 |

**読解のコツ**: Executor.scalaの冒頭にあるIsolatedSessionStateの状態遷移図（91-100行目）がセッション管理の理解に重要。Created→Active→Idle→Evictedの遷移を理解すること。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | Executor.scala | `core/src/main/scala/org/apache/spark/executor/Executor.scala` | launchTask()メソッド |
| 2-2 | CoarseGrainedExecutorBackend.scala | `core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala` | RPC通信バックエンド |

#### Step 3: タスク実行ロジックを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Executor.scala（TaskRunner内部クラス） | `core/src/main/scala/org/apache/spark/executor/Executor.scala` | TaskRunner.run()のタスク実行フロー |

#### Step 4: メトリクス収集を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | ExecutorMetricsPoller.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorMetricsPoller.scala` | メトリクスポーリング |
| 4-2 | ExecutorSource.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala` | メトリクスソース |

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

```
CoarseGrainedExecutorBackend (RPC受信)
    |
    +-- Executor.launchTask(taskDescription)
            |
            +-- TaskRunner(taskDescription)
                    |
                    +-- run()
                            |
                            +-- タスクデシリアライズ
                            +-- task.run(taskAttemptId, attemptNumber, metricsSystem)
                            |       +-- RDD.compute()
                            |       +-- メトリクス収集
                            |
                            +-- 結果シリアライズ
                            +-- execBackend.statusUpdate(FINISHED/FAILED)
    |
    +-- Executor.reportHeartBeat()
            +-- ドライバーへハートビート送信
```

### データフロー図

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

TaskDescription     ───▶  launchTask()            ───▶  TaskRunner生成
(ドライバーから)             |
                            +-- TaskRunner.run()
                                    |
                                    +-- task.run()        ───▶  DirectTaskResult
                                    +-- メトリクス収集           / IndirectTaskResult
                                                               (ドライバーへ)
Heartbeat Timer     ───▶  reportHeartBeat()       ───▶  Heartbeat
                                                         (ドライバーへ)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Executor.scala | `core/src/main/scala/org/apache/spark/executor/Executor.scala` | ソース | Executorメインクラス |
| CoarseGrainedExecutorBackend.scala | `core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala` | ソース | RPC通信バックエンド |
| TaskMetrics.scala | `core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala` | ソース | タスクメトリクス |
| ExecutorMetrics.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorMetrics.scala` | ソース | Executorメトリクス |
| ExecutorMetricsPoller.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorMetricsPoller.scala` | ソース | メトリクスポーリング |
| ExecutorSource.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala` | ソース | メトリクスソース |
| InputMetrics.scala | `core/src/main/scala/org/apache/spark/executor/InputMetrics.scala` | ソース | 入力メトリクス |
| ShuffleReadMetrics.scala | `core/src/main/scala/org/apache/spark/executor/ShuffleReadMetrics.scala` | ソース | シャッフル読み取りメトリクス |
| ShuffleWriteMetrics.scala | `core/src/main/scala/org/apache/spark/executor/ShuffleWriteMetrics.scala` | ソース | シャッフル書き込みメトリクス |
| ExecutorBackend.scala | `core/src/main/scala/org/apache/spark/executor/ExecutorBackend.scala` | ソース | バックエンドインターフェース |
