# 通知設計書 25-SparkListenerStageExecutorMetrics

## 概要

本ドキュメントは、Apache SparkにおけるSparkListenerStageExecutorMetricsイベント通知の設計仕様を定義する。このイベントは、ステージ完了時にExecutorのピークメトリクス値を記録する目的で発火する。ライブアプリケーションでは呼ばれず、履歴ログからの読み込み時にのみ処理される。

### 本通知の処理概要

本通知は、EventLoggingListenerがステージ完了時にExecutorレベルのピークメトリクスをイベントログに書き込む際に生成されるイベントである。履歴サーバでのログ読み込み時にAppStatusListenerにて処理される。

**業務上の目的・背景**：Sparkの履歴サーバ（History Server）では、過去に実行されたアプリケーションのメトリクスを正確に再現する必要がある。ライブアプリケーション中はSparkListenerExecutorMetricsUpdateイベントでリアルタイムにメトリクスが追跡されるが、履歴ログではステージ完了時点のピーク値のみが記録される。本イベントはその記録と再生のために使用される。

**通知の送信タイミング**：EventLoggingListenerのonStageCompletedメソッド内で、ステージに関連するすべてのExecutorのピークメトリクスをログに書き込む際に生成される。spark.eventLog.logStageExecutorMetrics設定が有効（デフォルトは設定依存）な場合にのみ記録される。

**通知の受信者**：ライブアプリケーションでは実質的な受信者はEventLoggingListenerのみ（ログ書き込み）。履歴サーバでのログ再生時には、AppStatusListenerがデッドExecutorを含むExecutorメモリメトリクスのピーク値を更新する。BasicEventFilterBuilderもフィルタリング対象として処理する。

**通知内容の概要**：execId（Executor ID）、stageId（ステージID）、stageAttemptId（ステージ試行ID）、executorMetrics（ExecutorMetricsオブジェクト）の4フィールドを含む。

**期待されるアクション**：履歴サーバでログを読み込む際に、AppStatusListenerがExecutorのメモリメトリクスピーク値を正確に復元し、Spark UIの履歴ビューに反映する。

## 通知種別

アプリ内イベント通知（Sparkイベントリスナーバス経由の非同期配信 / イベントログ経由の永続化・再生）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由）。ただしライブ時はEventLoggingListenerが直接ログに書き込み |
| 優先度 | 低 |
| リトライ | 無 |

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

LiveListenerBusに登録されたすべてのSparkListenerInterface実装リスナーに対してブロードキャスト配信される。SparkListenerBus.doPostEventメソッドにおいて、SparkListenerStageExecutorMetricsイベントとパターンマッチし、listener.onStageExecutorMetrics()が呼び出される。

## 通知テンプレート

### メール通知の場合

該当なし。本通知はSparkの内部イベントバスを介したプログラマティック通知である。

### 本文テンプレート

```
イベント: SparkListenerStageExecutorMetrics
Executor ID: {execId}
Stage ID: {stageId}
Stage Attempt ID: {stageAttemptId}
Executor Metrics: {executorMetrics}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| execId | Executor ID | EventLoggingListener内部のpeakExecutorMetricsマップ | Yes |
| stageId | ステージID | StageInfo | Yes |
| stageAttemptId | ステージ試行ID | StageInfo | Yes |
| executorMetrics | Executorレベルのピークメトリクス値 | liveStageExecutorMetricsマップ | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| ステージ完了 | SparkListenerStageCompleted | spark.eventLog.logStageExecutorMetrics=true | EventLoggingListener.onStageCompletedの中でステージに関連する全Executorのピーク値を書き出す |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| spark.eventLog.logStageExecutorMetrics=false | ステージExecutorメトリクスの記録が無効な場合、イベントは生成されない |
| spark.eventLog.enabled=false | イベントログ自体が無効な場合 |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[SparkListenerStageCompleted発火] --> B[EventLoggingListener.onStageCompleted]
    B --> C{shouldLogStageExecutorMetrics?}
    C -->|true| D[liveStageExecutorMetricsから該当ステージのピーク値取得]
    C -->|false| E[スキップ]
    D --> F[各ExecutorごとにSparkListenerStageExecutorMetrics生成]
    F --> G[logEvent でイベントログに書き込み]
    G --> H[履歴サーバでログ再生時]
    H --> I[AppStatusListener.onStageExecutorMetrics]
    I --> J[Executorメモリメトリクスピーク値更新]
```

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

### 参照テーブル一覧

該当なし。

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| AppStatusStore (KVStore) | UPDATE | 履歴サーバ再生時にAppStatusListenerがExecutorメトリクスピーク値を更新 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | リスナーのonStageExecutorMetrics実装で例外発生 | ListenerBusが例外をキャッチしログ出力 |
| ログ書き込み失敗 | イベントログファイルへの書き込みエラー | EventLoggingListenerが例外をログ出力 |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（ステージ完了頻度に依存） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。ステージ完了時に発火する。

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

本通知にはExecutor ID、ステージID、メトリクス数値のみが含まれ、個人情報や機密データは含まれない。

## 備考

- ライブアプリケーションでは直接呼ばれない。EventLoggingListenerが内部的にログに書き込み、履歴サーバのログ再生時にのみリスナーメソッドが呼ばれる。
- SparkListenerInterface行379-382のコメント：「Note that this is only present when reading from the event log (as in the history server), and is never called in a live application.」
- EventLoggingListenerの行73-75でliveStageExecutorMetricsがステージ単位のピーク値を追跡する内部マップとして定義されている。
- 以前のステージ試行のメトリクスは、次の試行の完了時にクリアされる（EventLoggingListener行138-141）。

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 行256-270: SparkListenerStageExecutorMetricsケースクラスの定義。execId、stageId、stageAttemptId、executorMetricsの4フィールド |

**読解のコツ**: コメント（行257-263）に「Peak metric values for the executor for the stage, written to the history log at stage completion」と明記されている。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | 行136-158: onStageCompletedメソッド。ステージ完了時にliveStageExecutorMetricsからピーク値を取得し、SparkListenerStageExecutorMetricsを生成してlogEvent |

**主要処理フロー**:
1. **行137**: shouldLogStageExecutorMetricsフラグのチェック
2. **行139-141**: 以前のステージ試行のメトリクスをクリア
3. **行146-147**: 該当ステージのExecutorメトリクスマップを取得
4. **行149-151**: 各Executorのピーク値についてSparkListenerStageExecutorMetricsを生成しlogEvent
5. **行157**: ステージ完了イベント自体をログに記録

#### Step 3: リスナー処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | 行982-989: onStageExecutorMetricsの実装。ログ読み込み時のみ処理。liveExecutorsとdeadExecutorsの両方をチェック |
| 3-2 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | 行60-61: doPostEventでのパターンマッチ |

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

```
EventLoggingListener.onStageCompleted (ライブ時: ログ書き込み)
    |
    +-- liveStageExecutorMetrics.remove(stageKey)
    |      |
    |      +-- 各Executorについて:
    |             |
    |             +-- logEvent(new SparkListenerStageExecutorMetrics(...))
    |
    +-- logEvent(stageCompletedEvent)

FsHistoryProvider (履歴サーバ: ログ再生時)
    |
    +-- ReplayListenerBus.replay
           |
           +-- SparkListenerBus.doPostEvent
                  |
                  +-- AppStatusListener.onStageExecutorMetrics
                         |
                         +-- Executorメモリメトリクスピーク値更新
```

### データフロー図

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

ライブ時:
Executorハートビートメトリクス --> EventLoggingListener -----------> イベントログ (JSON)
  (ステージ単位ピーク値蓄積)       .onStageCompleted

履歴サーバ再生時:
イベントログ (JSON) -----------> FsHistoryProvider.replay ----------> AppStatusStore更新
                                  AppStatusListener                   (Executorメトリクス復元)
                                  .onStageExecutorMetrics
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベントケースクラス定義 |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベント配信ディスパッチ |
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | ソース | イベント生成元（ステージ完了時にピーク値を書き出し） |
| AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | ソース | 履歴サーバでのメトリクス復元リスナー |
| BasicEventFilterBuilder.scala | `core/src/main/scala/org/apache/spark/deploy/history/BasicEventFilterBuilder.scala` | ソース | 履歴サーバ用イベントフィルタ |
| JsonProtocol.scala | `core/src/main/scala/org/apache/spark/util/JsonProtocol.scala` | ソース | JSON直列化・逆直列化 |
