# 通知設計書 6-SparkListenerTaskEnd

## 概要

本ドキュメントは、Apache Sparkのコアイベント通知である`SparkListenerTaskEnd`の設計について記述する。このイベントは、タスクの実行が終了した際に発火するイベント通知であり、タスクメトリクスおよびExecutorメトリクスを含む。

### 本通知の処理概要

SparkListenerTaskEndは、Sparkジョブ内の個別タスクが実行を完了（成功、失敗、キル含む）したことをリスナーに通知する最も情報量の多いイベントの一つである。

**業務上の目的・背景**：タスクの完了はステージ・ジョブの進捗に直結する最も重要なイベントである。タスクの成功・失敗の追跡、実行メトリクス（処理時間、シャッフルデータ量、GC時間等）の収集、Spark UIのタスク詳細表示、動的リソース割り当てのタスク完了カウント、イベントログへの記録など、多くの用途に利用される。失敗タスクの分析やパフォーマンスチューニングにおいて不可欠な情報を提供する。

**通知の送信タイミング**：DAGSchedulerの`handleTaskCompletion`メソッド内でタスク完了処理が行われる際に発火する（DAGScheduler.scala L1871-1873）。タスクの成功・失敗・キル等、すべての終了パターンで発火する。

**通知の受信者**：SparkListenerInterfaceを実装し、LiveListenerBusに登録されたすべてのリスナーが受信する。

**通知内容の概要**：stageId、stageAttemptId、taskType（タスクの種別名）、reason（TaskEndReason: 成功/失敗理由）、taskInfo（TaskInfo）、taskExecutorMetrics（ExecutorMetrics: ピークメモリ等）、taskMetrics（TaskMetrics: 処理メトリクス。失敗時はnullの可能性あり）を含む。

**期待されるアクション**：受信したリスナーは、タスクの完了状態とメトリクスをUIに反映する、イベントログに記録する、アキュムレータを更新する、動的リソース割り当ての判断材料とする。

## 通知種別

アプリ内通知（Sparkイベントバス経由の非同期インプロセス通知）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 中 |
| リトライ | 無し |

### 送信先決定ロジック

LiveListenerBusに登録された全リスナーに対してブロードキャスト配信される。

## 通知テンプレート

### メール通知の場合

本イベントはメール通知ではなく、Sparkイベントバス経由のインプロセス通知である。

### 本文テンプレート

```
イベントクラス: SparkListenerTaskEnd
シリアライズ形式: JSON（Jackson, @JsonTypeInfo）

{
  "Event": "org.apache.spark.scheduler.SparkListenerTaskEnd",
  "Stage ID": <int>,
  "Stage Attempt ID": <int>,
  "Task Type": "<string>",
  "Task End Reason": { ... },
  "Task Info": { ... },
  "Task Executor Metrics": { ... },
  "Task Metrics": { ... }  // nullの場合あり
}
```

### 添付ファイル

| ファイル名 | 形式 | 条件 | 説明 |
|----------|------|------|------|
| 該当なし | - | - | インプロセス通知のため添付ファイルなし |

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| stageId | ステージID | event.task.stageId | Yes |
| stageAttemptId | ステージアテンプト番号 | event.task.stageAttemptId | Yes |
| taskType | タスク種別（ShuffleMapTask/ResultTaskなど） | Utils.getFormattedClassName(event.task) | Yes |
| reason | タスク終了理由 | event.reason (TaskEndReason) | Yes |
| taskInfo | タスク詳細情報 | event.taskInfo | Yes |
| taskExecutorMetrics | Executorメトリクス（ピーク値） | new ExecutorMetrics(event.metricPeaks) | Yes |
| taskMetrics | タスク実行メトリクス | event.resultから復元、失敗時null | No（@Nullable） |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| スケジューラ内部処理 | DAGScheduler.handleTaskCompletion内 | タスク完了時（成功・失敗・キル問わず） | DAGScheduler.scala L1871-1873 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| LiveListenerBus停止後 | イベントは配信されない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[タスク実行完了] --> B[DAGScheduler.handleTaskCompletion]
    B --> C{taskMetrics復元可能?}
    C -->|Yes| D[taskMetricsを復元]
    C -->|No/例外| E[taskMetrics = null]
    D --> F[SparkListenerTaskEnd生成]
    E --> F
    F --> G[listenerBus.post]
    G --> H[各リスナーのonTaskEnd呼び出し]
    H --> I[終了]
```

## データベース参照・更新仕様

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| KVStore | AppStatusListenerがタスク完了情報を格納 | UI表示用 |

### テーブル別参照項目詳細

#### KVStore

| 参照項目（カラム名） | 用途 | 取得条件 |
|-------------------|------|---------|
| TaskDataWrapper | タスクの完了状態とメトリクス | taskIdで検索 |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| KVStore | UPDATE | タスク状態をSUCCESS/FAILEDに更新、メトリクスを記録 |
| イベントログファイル | APPEND | イベントをJSON記録 |

#### 送信ログテーブル

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| UPDATE | TaskDataWrapper | 完了状態、メトリクス | AppStatusListenerによる更新 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| taskMetrics復元失敗 | タスク失敗時にメトリクスが欠損 | taskMetrics=nullでイベントを生成（DAGScheduler.scala L1860-1868） |
| キューオーバーフロー | イベント生成速度超過 | イベントドロップ |
| リスナー例外 | onTaskEnd内で例外 | 例外キャッチ、ログ出力 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 0 |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| キュー容量 | デフォルト10,000イベント |

### 配信時間帯

時間帯制限なし。

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

- TaskInfoにはexecutorId、host等のインフラ情報が含まれる
- TaskMetricsにはメモリ使用量、GC時間等のシステムメトリクスが含まれる
- TaskEndReasonの失敗情報にはスタックトレースが含まれる可能性がある

## 備考

- SparkListenerTaskEndは7つのフィールドを持つ最も複雑なSparkListenerEventの一つである（SparkListener.scala L67-76）
- taskMetricsフィールドには`@Nullable`アノテーションが付与されており、タスク失敗時にnullとなる可能性がある（L74-75: コメント「may be null if the task has failed」）
- taskTypeはタスクオブジェクトのクラス名から`Utils.getFormattedClassName`で生成される（L1872）
- taskExecutorMetricsは`new ExecutorMetrics(event.metricPeaks)`で新規生成される（L1873）
- 大規模ジョブでは大量のTaskEndイベントが発火するため、リスナー処理のパフォーマンスが重要

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | L67-76: SparkListenerTaskEndケースクラス。7フィールド（stageId, stageAttemptId, taskType, reason, taskInfo, taskExecutorMetrics, taskMetrics）。L74-75: `@Nullable taskMetrics: TaskMetrics`のコメント |

**読解のコツ**: TaskEndReason（Success, FetchFailed, ExceptionFailure等）の種別はorg.apache.spark.TaskEndReasonで定義されている。TaskMetricsはShuffleReadMetrics、ShuffleWriteMetrics等の詳細なサブメトリクスを含む。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | L1871-1873: handleTaskCompletion内でSparkListenerTaskEndを生成しpost。L1860-1868: taskMetricsの復元とnullフォールバック |

**主要処理フロー**:
1. **L1860-1866**: taskMetricsの復元を試みる。NonFatal例外発生時はnullにフォールバック
2. **L1867-1868**: メトリクス復元不可能な場合はnull
3. **L1871-1873**: `SparkListenerTaskEnd(event.task.stageId, event.task.stageAttemptId, Utils.getFormattedClassName(event.task), event.reason, event.taskInfo, new ExecutorMetrics(event.metricPeaks), taskMetrics)`を生成しpost

#### Step 3: イベントディスパッチを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | L44-45: SparkListenerTaskEndのパターンマッチ → listener.onTaskEnd()呼び出し |

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

```
TaskResultGetter / TaskSetManager (タスク完了通知)
    |
    +-- DAGScheduler.handleTaskCompletion() [DAGScheduler.scala L1871]
            |
            +-- listenerBus.post(SparkListenerTaskEnd(...))
                    |
                    +-- LiveListenerBus.post()
                            |
                            +-- AsyncEventQueue.post() [各キュー]
                                    |
                                    +-- listener.onTaskEnd()
                                            |
                                            +-- AppStatusListener (タスク完了記録)
                                            +-- EventLoggingListener (イベントログ)
                                            +-- ExecutorAllocationManager (タスク数追跡)
```

### データフロー図

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

CompletionEvent             LiveListenerBus                  各リスナー
(task, reason,             (AsyncEventQueue x N)
 taskInfo, metricPeaks,
 taskMetrics)
                                                              AppStatusListener
stageId, stageAttemptId                                       (KVStore更新)
taskType, reason ---------> SparkListenerTaskEnd -----------> EventLoggingListener
taskInfo                                                      (イベントログ)
taskExecutorMetrics                                           ExecutorAllocationManager
taskMetrics(@Nullable)                                        (Executor管理)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | SparkListenerTaskEndケースクラス定義（L67-76） |
| DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | ソース | handleTaskCompletion内のイベント発火（L1871-1873） |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベントディスパッチ（L44-45） |
| LiveListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala` | ソース | 非同期配信制御 |
| AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | ソース | UI用タスク完了記録 |
