# 通知設計書 41-SparkListenerDriverAccumUpdates

## 概要

本ドキュメントは、Apache SparkのSQL実行エンジンにおけるドライバ側アキュムレータ更新イベント `SparkListenerDriverAccumUpdates` の通知設計を記載する。このイベントはドライバ側で更新されるSQLメトリクス（サブクエリ実行時間、ファイル数など）をSpark UIおよびイベントリスナーに伝達するために使用される。

### 本通知の処理概要

`SparkListenerDriverAccumUpdates` は、ドライバ側で計測・更新されたSQLメトリクスの値をLiveListenerBus経由でリスナーに非同期配信するイベント通知である。Executor側のメトリクスはタスク完了時に自動的に収集されるが、ドライバ側で計算されるメトリクス（例：サブクエリの実行時間、スキャン対象ファイル数、ブロードキャスト変数のサイズなど）は自動伝搬されないため、このイベントを通じて明示的にリスナーへ通知する必要がある。

**業務上の目的・背景**：SparkのSQL UIでは、各物理プラン演算子のメトリクスを表示してクエリのパフォーマンス解析を支援する。ドライバ側で計算されるメトリクス（サブクエリ実行時間、ファイル数、ブロードキャスト変数サイズなど）はExecutorのタスクメトリクスとは異なり自動収集されない。そのため、明示的にこのイベントを発火してUI上に表示させることで、ユーザーがクエリの全体的なパフォーマンスを正確に把握できるようにする。

**通知の送信タイミング**：`SQLMetrics.postDriverMetricUpdates()` または `SQLMetrics.postDriverMetricsUpdatedByValue()` メソッドが呼び出された時に発火する。具体的には、物理プラン演算子（SparkPlan）の `doExecute()` 実行後にドライバ側メトリクスを更新する際、SparkContextのListenerBusを通じてポストされる。

**通知の受信者**：`SQLAppStatusListener` が主要な受信者であり、受信したアキュムレータ更新をライブ実行データ（`LiveExecutionData`）の `driverAccumUpdates` フィールドに蓄積する。その他、`SparkListener` インタフェースを実装した任意のリスナー（History Server、カスタムリスナー等）も `onOtherEvent` メソッド経由で受信可能。

**通知内容の概要**：SQL実行ID（`executionId`）とアキュムレータIDからメトリクス値へのマッピング（`accumUpdates: Seq[(Long, Long)]`）を含む。各タプルの第1要素がアキュムレータID、第2要素がメトリクス値（64ビット整数）である。

**期待されるアクション**：受信したリスナーはアキュムレータ更新を既存のメトリクスデータに統合する。`SQLAppStatusListener` はこのデータを `LiveExecutionData.driverAccumUpdates` に追加し、メトリクス集計時にタスクメトリクスと合算してSpark UIに表示する。ドライバ値がタスク値より大きい場合はmax値を上書きし、UI上に"driver"と表示する。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 中（通常のイベントキュー） |
| リトライ | 無（イベントバスのキューに投入されるため、キュー溢れ時はドロップ） |

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

SparkContextに登録されたすべての `SparkListener` に対してブロードキャスト配信される。`onOtherEvent` メソッドでイベント型をパターンマッチして処理するリスナーが受信する。主たる受信者は `SQLAppStatusListener` であり、`onOtherEvent` 内で `SparkListenerDriverAccumUpdates` にマッチして `onDriverAccumUpdates` プライベートメソッドを呼び出す。

## 通知テンプレート

### メール通知の場合

本通知はメール通知ではなく、Sparkアプリ内のイベントバスを通じたインプロセス通知であるため、メールテンプレートは該当しない。

### 本文テンプレート

```
SparkListenerDriverAccumUpdates(
  executionId: Long,
  accumUpdates: Seq[(Long, Long)]  // (accumulatorId, value)
)
```

### 添付ファイル

該当なし（インプロセスイベント通知のため）

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| executionId | SQL実行を一意に識別するID | SQLExecution.EXECUTION_ID_KEY | Yes |
| accumUpdates | アキュムレータIDとメトリクス値のペアのシーケンス | SQLMetric.id -> SQLMetric.value | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| API呼び出し | SQLMetrics.postDriverMetricUpdates() | executionIdがnullでないこと | ドライバ側でSQLMetricオブジェクトのリストからID-値ペアを抽出しポスト |
| API呼び出し | SQLMetrics.postDriverMetricsUpdatedByValue() | executionIdがnullでないこと | 直接ID-値ペアのシーケンスを指定してポスト |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| executionIdがnull | SQL実行コンテキスト外での呼び出し時はイベントを送信しない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[SparkPlan.doExecute完了] --> B[SQLMetrics.postDriverMetricUpdates呼び出し]
    B --> C{executionId != null?}
    C -->|Yes| D[SparkListenerDriverAccumUpdatesイベント生成]
    C -->|No| E[送信スキップ]
    D --> F[SparkContext.listenerBus.post]
    F --> G[LiveListenerBusがイベントをキューに投入]
    G --> H[非同期でリスナーに配信]
    H --> I[SQLAppStatusListener.onOtherEvent]
    I --> J[onDriverAccumUpdates処理]
    J --> K[LiveExecutionData.driverAccumUpdatesに追加]
    K --> L[kvstoreへの書き込み更新]
    E --> M[終了]
    L --> M
```

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

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| liveExecutions (ConcurrentHashMap) | executionIdに対応するLiveExecutionDataの取得 | インメモリデータ構造 |

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

#### liveExecutions

| 参照項目（カラム名） | 用途 | 取得条件 |
|-------------------|------|---------|
| LiveExecutionData | 実行中のSQL実行データ | executionIdをキーとして取得 |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| LiveExecutionData.driverAccumUpdates | UPDATE（追記） | ドライバ側アキュムレータ更新値の追加 |
| ElementTrackingStore (kvstore) | WRITE | LiveExecutionDataのシリアライズと永続化 |

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

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| UPDATE | driverAccumUpdates | 既存値 ++ 新規accumUpdates | Seq[(Long, Long)]に追記 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| イベントキュー溢れ | LiveListenerBusのキューが満杯 | イベントがドロップされる。メトリクスはUI上で不完全になる可能性がある |
| 実行データ不在 | liveExecutionsにexecutionIdが存在しない | Option(liveExecutions.get(executionId)).foreach で安全にスキップ |
| JSON デシリアライズエラー | イベントログ読込時に型変換に失敗 | LongLongTupleConverterにより(Int,Int)から(Long,Long)への変換を自動実行 |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（LiveListenerBusのキュー容量による暗黙的制限） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。SQL実行時に随時発火される。

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

本イベントはインプロセスで配信されるため、外部ネットワーク通信は発生しない。イベントログに記録される場合、メトリクスデータ（数値のみ）が含まれるが、個人情報やセキュリティセンシティブな情報は含まれない。ただし、`executionId` からSQL文の内容が間接的に特定可能であるため、イベントログファイルのアクセス制御は適切に設定する必要がある。`@DeveloperApi` アノテーションが付与されており、将来のバージョンでAPIが変更される可能性がある。

## 備考

- `SparkListenerDriverAccumUpdates` は `@DeveloperApi` アノテーションが付与されている
- Jackson デシリアライズ時に `LongLongTupleConverter` が使用され、`(Int, Int)` から `(Long, Long)` への自動変換が行われる（SPARK-18462対応）
- `accumUpdates` フィールドには `@JsonDeserialize(contentConverter = classOf[LongLongTupleConverter])` が付与されている
- ドライバメトリクス値がタスクメトリクス値より大きい場合、`aggregateMetrics` 内で `maxMetricsFromAllStages` からエントリが削除され、UI上に"driver"と表示される

---

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

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

### 推奨読解順序

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

まず、イベントクラスの定義とデータ構造を理解することが重要である。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SQLListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala` | `SparkListenerDriverAccumUpdates` ケースクラスの定義（106-110行目）。executionIdとaccumUpdatesの構造を理解する |
| 1-2 | SQLListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala` | `LongLongTupleConverter` クラス（120-139行目）。Jacksonデシリアライズ時の型変換ロジック |

**読解のコツ**: `@DeveloperApi` アノテーションは開発者向けAPIであることを示す。`@JsonDeserialize` はJacksonによるJSONデシリアライズ時のカスタムコンバータ指定。Scalaのケースクラスは不変データオブジェクトでパターンマッチに利用可能。

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

イベント発火の起点である `SQLMetrics.postDriverMetricUpdates` を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SQLMetrics.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala` | `postDriverMetricUpdates` メソッド（215-223行目）。executionIdのnullチェック後、listenerBus経由でイベントをポストする |
| 2-2 | SQLMetrics.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala` | `postDriverMetricsUpdatedByValue` メソッド（201-209行目）。値を直接指定するバリアント |

**主要処理フロー**:
1. **215行目**: `postDriverMetricUpdates` メソッド定義。SparkContext, executionId, メトリクスリストを受け取る
2. **219行目**: `executionId != null` のガード条件チェック
3. **220-221行目**: `sc.listenerBus.post(SparkListenerDriverAccumUpdates(...))` でイベントを生成・配信

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

イベントを受信する `SQLAppStatusListener` の処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | `onOtherEvent` メソッド（438-445行目）。パターンマッチで `SparkListenerDriverAccumUpdates` を識別 |
| 3-2 | SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | `onDriverAccumUpdates` メソッド（430-436行目）。driverAccumUpdatesへの追記とupdate呼び出し |

**主要処理フロー**:
- **438行目**: `onOtherEvent` でイベント型のパターンマッチ
- **443行目**: `SparkListenerDriverAccumUpdates` にマッチした場合、`onDriverAccumUpdates` を呼び出し
- **431行目**: イベントからexecutionIdとaccumUpdatesを分解
- **432行目**: `liveExecutions.get(executionId)` で対応する実行データを取得
- **433行目**: `exec.driverAccumUpdates = exec.driverAccumUpdates ++ accumUpdates` で値を追記
- **434行目**: `update(exec)` でkvstoreへの書き込みをトリガー

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

ドライバメトリクスがタスクメトリクスと統合される `aggregateMetrics` の処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | `aggregateMetrics` メソッド（208-312行目）。driverAccumUpdatesの統合処理（278-296行目）に注目 |

**主要処理フロー**:
- **278行目**: `exec.driverAccumUpdates.foreach` でドライバ側メトリクスをイテレート
- **285行目**: ドライバ値がmaxタスク値より大きい場合、maxMetricsFromAllStagesからエントリを削除
- **288-289行目**: 既存タスクメトリクス配列にドライバ値を追加

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

```
SQLMetrics.postDriverMetricUpdates() [SQLMetrics.scala:215]
    |
    +-- SparkContext.listenerBus.post()
            |
            +-- LiveListenerBus (非同期キュー)
                    |
                    +-- SQLAppStatusListener.onOtherEvent() [SQLAppStatusListener.scala:438]
                            |
                            +-- onDriverAccumUpdates() [SQLAppStatusListener.scala:430]
                                    |
                                    +-- LiveExecutionData.driverAccumUpdates 更新
                                    |
                                    +-- update() [SQLAppStatusListener.scala:452]
                                            |
                                            +-- LiveExecutionData.write(kvstore, now)
```

### データフロー図

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

SQLMetric.id + value ------> postDriverMetricUpdates() -------> SparkListenerDriverAccumUpdates
                                                                       |
                                                                       v
                                                              LiveListenerBus (非同期)
                                                                       |
                                                                       v
                                                              SQLAppStatusListener
                                                                       |
                                                                       v
                                                      LiveExecutionData.driverAccumUpdates
                                                                       |
                                                                       v
                                                              aggregateMetrics()
                                                                       |
                                                                       v
                                                              ElementTrackingStore (kvstore)
                                                                       |
                                                                       v
                                                              Spark UI メトリクス表示
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SQLListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala` | ソース | SparkListenerDriverAccumUpdatesイベントクラスの定義 |
| SQLMetrics.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/metric/SQLMetrics.scala` | ソース | イベント発火のエントリーポイント（postDriverMetricUpdates） |
| SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | ソース | イベント受信・処理のリスナー実装 |
| SQLAppStatusListenerSuite.scala | `sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala` | テスト | リスナーのテストケース |
| SQLEventFilterBuilder.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/history/SQLEventFilterBuilder.scala` | ソース | History Serverでのイベントフィルタリング |
