# 通知設計書 24-SparkListenerExecutorMetricsUpdate

## 概要

本ドキュメントは、Apache SparkにおけるSparkListenerExecutorMetricsUpdateイベント通知の設計仕様を定義する。このイベントは、ExecutorからのハートビートでタスクメトリクスとExecutorレベルのメトリクスを受信した際に発火し、タスクの進捗状況とリソース使用状況のリアルタイム監視を可能にする。

### 本通知の処理概要

本通知は、Executorが定期的なハートビートを通じてドライバに送信するタスクメトリクスおよびExecutorレベルのメトリクス更新情報を、リスナーに配信するイベントである。

**業務上の目的・背景**：Sparkの分散実行環境では、Executorが定期的にハートビートを送信してタスクの進捗状況とリソース使用状況を報告する。このイベントにより、Spark UIでのリアルタイムなタスクメトリクス表示、イベントログへのExecutorメトリクスの記録（ステージ単位のピーク値追跡）、および外部監視システムとの連携が可能となる。ドライバ自身もheartbeatを通じて自身のメトリクスを報告する。

**通知の送信タイミング**：ExecutorからのハートビートがDAGSchedulerのexecutorHeartbeatReceivedメソッドで処理された際に発火する。また、ドライバ自身もSparkContext.reportDriverHeartBeatMetrics内でドライバメトリクスを報告する際に発火する。

**通知の受信者**：LiveListenerBusに登録されたすべてのSparkListenerInterface実装クラスが受信対象。主要な受信者はAppStatusListener（タスクメトリクスのUI更新、Executorメモリメトリクスのピーク値追跡）、EventLoggingListener（ドライバメトリクスのログ記録、ステージ単位のピーク値追跡）である。

**通知内容の概要**：execId（Executor ID）、accumUpdates（タスクID・ステージID・ステージ試行ID・アキュムレータ更新のタプルシーケンス）、executorUpdates（ステージキー別のExecutorMetrics）の3フィールドを含む。

**期待されるアクション**：AppStatusListenerがタスクの進捗情報を更新し、Spark UIに反映する。EventLoggingListenerがステージ単位のExecutorメトリクスのピーク値を追跡する。

## 通知種別

アプリ内イベント通知（Sparkイベントリスナーバス経由の非同期配信）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 低（高頻度で発生するため） |
| リトライ | 無（イベントバスのキュー管理に依存） |

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

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

## 通知テンプレート

### メール通知の場合

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

### 本文テンプレート

```
イベント: SparkListenerExecutorMetricsUpdate
Executor ID: {execId}
メトリクス更新数: {accumUpdates.size}
Executor更新キー数: {executorUpdates.size}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| execId | Executor ID（ドライバの場合は"driver"） | ハートビートメッセージ | Yes |
| accumUpdates | (taskId, stageId, stageAttemptId, accumUpdates)のシーケンス | ハートビートメッセージ | Yes |
| executorUpdates | (stageId, stageAttemptId) -> ExecutorMetricsのマップ | ハートビートメッセージ | No（デフォルトMap.empty） |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| ハートビート | Executorからのハートビート受信 | 常時（Executorがアクティブである限り） | DAGScheduler.executorHeartbeatReceivedで処理 |
| ハートビート | ドライバのハートビートメトリクス報告 | 常時 | SparkContext.reportDriverHeartBeatMetricsで処理 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 特になし | ハートビートが到着するたびにイベントが発火する |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A1[Executor: ハートビート送信] --> B[DAGScheduler.executorHeartbeatReceived]
    A2[Driver: reportDriverHeartBeatMetrics] --> C[listenerBus.post SparkListenerExecutorMetricsUpdate]
    B --> C
    C --> D[SparkListenerBus.doPostEvent]
    D --> E[listener.onExecutorMetricsUpdate]
    E --> F[AppStatusListener: タスクメトリクス更新]
    E --> G[EventLoggingListener: ピーク値追跡]
```

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

### 参照テーブル一覧

該当なし。

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| AppStatusStore (KVStore) | UPDATE | AppStatusListenerがタスクメトリクスを更新 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | リスナーのonExecutorMetricsUpdate実装で例外発生 | ListenerBusが例外をキャッチしログ出力 |
| イベントキュー溢れ | LiveListenerBusのキューが満杯 | イベントがドロップされる |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（ハートビート間隔に依存。デフォルト10秒間隔） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。Executorがアクティブである限り、ハートビート間隔で継続的に発火する。

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

本通知にはExecutor ID、タスクID、ステージID、メトリクス数値が含まれるが、アプリケーションのデータ自体は含まれない。個人情報や機密データへの直接的なアクセスリスクはない。

## 備考

- @since 3.1.0のアノテーションが付与されている。
- ドライバからの更新時のexecIdは"driver"固定であり、ステージキーはダミーの(-1, -1)が使用される（EventLoggingListener.DRIVER_STAGE_KEY）。
- EventLoggingListenerではspark.eventLog.logStageExecutorMetrics設定に基づき、ドライバメトリクスのログ記録とステージ単位のピーク値追跡を行う。
- AppStatusListenerではSparkListenerExecutorMetricsUpdateイベントはライブUIでのみ処理される（行959のコメント参照）。

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 行241-254: SparkListenerExecutorMetricsUpdateケースクラスの定義。execId、accumUpdates、executorUpdatesの3フィールド |

**読解のコツ**: accumUpdatesは(taskId: Long, stageId: Int, stageAttemptId: Int, accumUpdates: Seq[AccumulableInfo])のタプルシーケンス。executorUpdatesは(stageId, stageAttemptId) -> ExecutorMetricsのマップ。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | 行381-391: executorHeartbeatReceivedメソッド。ExecutorハートビートからイベントをPost |
| 2-2 | SparkContext.scala | `core/src/main/scala/org/apache/spark/SparkContext.scala` | 行2976-2981: reportDriverHeartBeatMetrics。ドライバ自身のメトリクスをPost |

**主要処理フロー**:
1. **DAGScheduler行386**: listenerBus.post(SparkListenerExecutorMetricsUpdate(execId, accumUpdates, executorUpdates))
2. **SparkContext行2980**: listenerBus.post(SparkListenerExecutorMetricsUpdate("driver", accumUpdates, driverUpdates))

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | 行938-980: onExecutorMetricsUpdateの実装。タスクメトリクス更新とExecutorピーク値追跡 |
| 3-2 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | 行250-268: onExecutorMetricsUpdateの実装。ドライバメトリクスのログとピーク値追跡 |

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

```
Executor (ハートビート)
    |
    +-- HeartbeatReceiver.receiveAndReply
           |
           +-- DAGScheduler.executorHeartbeatReceived
                  |
                  +-- listenerBus.post(SparkListenerExecutorMetricsUpdate)

SparkContext (ドライバハートビート)
    |
    +-- reportDriverHeartBeatMetrics
           |
           +-- listenerBus.post(SparkListenerExecutorMetricsUpdate("driver", ...))

           LiveListenerBus
                  |
                  +-- SparkListenerBus.doPostEvent
                         |
                         +-- AppStatusListener.onExecutorMetricsUpdate
                         +-- EventLoggingListener.onExecutorMetricsUpdate
```

### データフロー図

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

Executorハートビート --------> DAGScheduler ----> SparkListenerExecutorMetricsUpdate
(execId, accumUpdates,              |                     |
 executorUpdates)                   v                     v
                             LiveListenerBus        AppStatusStore更新
                                                    EventLog (ピーク値追跡)

ドライバハートビート --------> SparkContext ----> SparkListenerExecutorMetricsUpdate("driver")
(currentMetrics)                                       |
                                                       v
                                                 EventLog (ドライバメトリクス)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベントケースクラス定義 |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベント配信ディスパッチ |
| DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | ソース | Executorハートビート経由のイベント発火元 |
| SparkContext.scala | `core/src/main/scala/org/apache/spark/SparkContext.scala` | ソース | ドライバハートビートメトリクスのイベント発火元 |
| AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | ソース | Spark UIメトリクス更新リスナー |
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.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直列化・逆直列化 |
