# 通知設計書 1-SparkListenerStageSubmitted

## 概要

本ドキュメントは、Apache Sparkのコアイベント通知である`SparkListenerStageSubmitted`の設計について記述する。このイベントは、ステージがDAGSchedulerによってスケジューラに送信された際に発火する内部イベント通知である。

### 本通知の処理概要

SparkListenerStageSubmittedは、Sparkジョブ実行の中核的なイベント通知であり、ステージの送信タイミングをリスナーに伝達する。

**業務上の目的・背景**：Sparkアプリケーションの実行進捗をモニタリングし、ジョブパイプラインの各ステージがいつスケジューラに投入されたかを追跡するために必要である。Spark UIのステージタブ、イベントログ、カスタム監視ツールなどがこのイベントを利用してステージの実行状態を可視化する。

**通知の送信タイミング**：DAGScheduler内の`submitMissingTasks`メソッドでステージの実行準備が完了し、タスクセットがTaskSchedulerに投入される直前に発火する。また、タスク生成時に例外が発生した場合（`getPreferredLocs`の失敗時など）にも、ステージ送信済みとして発火してから中断処理が行われる。空のパーティション（0タスク）のジョブでは`submitJob`内で即座にJobStartとJobEndと共に発火する。

**通知の受信者**：SparkListenerInterfaceを実装し、LiveListenerBusに登録されたすべてのリスナーが受信する。主な受信者は、AppStatusListener（Spark UI更新）、EventLoggingListener（イベントログ記録）、ExecutorAllocationManager（動的リソース割り当て）である。

**通知内容の概要**：StageInfo（ステージID、名前、タスク数、RDD情報、親ステージ情報など）およびProperties（ジョブプロパティ、フェアスケジューラプール名など）を含む。

**期待されるアクション**：受信したリスナーは、ステージの状態をUIに反映する、イベントログに記録する、動的リソース割り当ての判断材料とするなどの処理を行う。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由、AsyncEventQueueで別スレッド配信） |
| 優先度 | 中（通常のイベントキューで処理） |
| リトライ | 無し（キュー溢れ時はイベントドロップ） |

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

LiveListenerBusに登録された全リスナーに対してブロードキャスト配信される。リスナーは以下のキューに分類される：
- `shared`：ユーザー登録リスナー
- `appStatus`：AppStatusListener（UI用）
- `executorManagement`：ExecutorAllocationManager
- `eventLog`：EventLoggingListener

各キューは独立したスレッドで配信処理を行う。

## 通知テンプレート

### メール通知の場合

本イベントはメール通知ではなく、Sparkイベントバス経由のインプロセス通知である。テンプレートは適用されない。

### 本文テンプレート

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

{
  "Event": "org.apache.spark.scheduler.SparkListenerStageSubmitted",
  "Stage Info": { ... },
  "Properties": { ... }
}
```

### 添付ファイル

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

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| stageInfo | ステージの詳細情報 | DAGScheduler.stage.latestInfo | Yes |
| properties | ジョブプロパティ | DAGScheduler.properties（クローン） | No（nullの場合あり） |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| スケジューラ内部処理 | DAGScheduler.submitMissingTasks | ステージの実行タスクが確定した時 | 通常のステージ送信パス（DAGScheduler.scala L1688） |
| スケジューラ内部処理 | DAGScheduler.submitMissingTasks（例外パス） | タスク生成中に例外が発生した時 | getPreferredLocs失敗時等（DAGScheduler.scala L1673） |
| スケジューラ内部処理 | DAGScheduler.submitJob（空パーティション） | パーティションが空のジョブ送信時 | 0タスクのジョブで即座にJobStart/JobEndと共に発火 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| LiveListenerBus停止後 | `stopped`フラグがtrueの場合、post()は即座にreturnしイベントは配信されない |
| キュー容量超過 | AsyncEventQueueの容量を超えた場合、イベントはドロップされる |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[DAGScheduler.submitMissingTasks呼び出し] --> B[ステージのタスク情報を確定]
    B --> C[stage.makeNewStageAttempt]
    C --> D[SparkListenerStageSubmitted生成]
    D --> E[listenerBus.post]
    E --> F{LiveListenerBus停止済み?}
    F -->|Yes| G[イベント破棄]
    F -->|No| H{バス開始済み?}
    H -->|No| I[queuedEventsにバッファリング]
    H -->|Yes| J[全AsyncEventQueueにpost]
    J --> K[各キューのLinkedBlockingQueueに投入]
    K --> L{キュー容量内?}
    L -->|Yes| M[配信スレッドがリスナーに配信]
    L -->|No| N[イベントドロップ・カウンタ増加]
    M --> O[listener.onStageSubmitted呼び出し]
    O --> P[終了]
    G --> P
    I --> P
    N --> P
```

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

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| KVStore（InMemoryStore/LevelDB） | AppStatusListenerがステージ情報を格納・参照 | Spark UIのバックエンドストア |

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

#### KVStore

| 参照項目（カラム名） | 用途 | 取得条件 |
|-------------------|------|---------|
| StageDataWrapper | ステージの状態情報 | stageId, stageAttemptIdで検索 |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| KVStore | INSERT/UPDATE | AppStatusListenerがStageDataWrapperを書き込み |
| イベントログファイル | APPEND | EventLoggingListenerがJSON形式でイベントを追記 |

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

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| WRITE | StageDataWrapper | ステージ状態=ACTIVE | AppStatusListenerによるKVStore更新 |
| APPEND | イベントログJSON | シリアライズされたイベント | EventLoggingListenerによるログファイル追記 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| キューオーバーフロー | イベント生成速度がリスナー処理速度を上回った場合 | イベントがドロップされ、droppedEventsCounterが増加。ログに警告出力 |
| リスナー例外 | リスナーのonStageSubmittedが例外をスローした場合 | ListenerBus.doPostEventで例外をキャッチしログに記録。他リスナーへの配信は継続 |
| シリアライズ失敗 | イベントログ書き込み時にJSON変換が失敗した場合 | EventLoggingListenerで例外処理 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 0（リトライなし） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A（ドロップされたイベントは復元されない） |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| キュー容量 | デフォルト10,000イベント（`spark.scheduler.listenerbus.eventqueue.capacity`で設定可能） |
| キュー別容量 | `spark.scheduler.listenerbus.eventqueue.${name}.capacity`で個別設定可能 |

### 配信時間帯

時間帯による制限はない。SparkContext起動中は常時配信可能。

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

- StageInfoにはユーザーコードの呼び出し元情報（callSite）が含まれるため、イベントログファイルのアクセス制御が必要
- Propertiesにはフェアスケジューラプール名などのジョブ設定が含まれる
- EventLoggingListenerはSparkListenerEnvironmentUpdateのプロパティを秘匿化（redact）するが、StageSubmittedのProperties自体は秘匿化されない
- イベントログファイルはHDFS/S3等に保存される場合、ファイルシステムレベルのアクセス制御に依存する

## 備考

- SparkListenerEventトレイトを継承し、`@JsonTypeInfo`アノテーションによりJacksonでの多態的シリアライズ/デシリアライズに対応している
- `logEvent`プロパティはデフォルトtrueであり、イベントログに記録される
- propertiesパラメータはデフォルト値nullが設定されているため、引数省略時はnullとなる
- DAGSchedulerからの投入時には`Utils.cloneProperties`でプロパティがクローンされ、スレッドセーフ性が確保されている

---

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

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

### 推奨読解順序

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

まず、SparkListenerStageSubmittedイベントのデータ構造と、その基底トレイトであるSparkListenerEventを把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListenerEvent.scala | `common/utils/src/main/scala/org/apache/spark/scheduler/SparkListenerEvent.scala` | SparkListenerEventトレイトの定義。L25: `@JsonTypeInfo`アノテーションによるポリモーフィックJSON対応、L27: `logEvent`プロパティ |
| 1-2 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | L33-34: SparkListenerStageSubmittedケースクラスの定義。stageInfo（StageInfo型）とproperties（Properties型、デフォルトnull）の2フィールド |

**読解のコツ**: Scalaのcase classは自動的にequals/hashCode/toStringが生成される。`extends SparkListenerEvent`によりJacksonシリアライズ対応となる。

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

イベントが発火される起点であるDAGSchedulerのsubmitMissingTasksメソッドを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | L1688: 通常パスでのSparkListenerStageSubmittedのpost。L1673: 例外パスでのpost |

**主要処理フロー**:
1. **L1660-1668**: タスクのPreferred Locationsを計算
2. **L1670-1677**: 例外発生時にStageSubmittedを発火してからabortStage
3. **L1680**: `stage.makeNewStageAttempt`でステージアテンプト情報を更新
4. **L1685-1686**: パーティションが空でなければsubmissionTimeを設定
5. **L1688-1689**: `SparkListenerStageSubmitted`を生成し`listenerBus.post`で配信

#### Step 3: イベントバスの配信機構を理解する

LiveListenerBusによる非同期配信メカニズムを把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | LiveListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala` | L46: LiveListenerBusクラスの定義。L126-153: post()メソッドの非同期配信ロジック。L155-160: postToQueues()による全キューへの配信 |
| 3-2 | AsyncEventQueue.scala | `core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala` | L38-44: AsyncEventQueueの定義。L61: LinkedBlockingQueueによるイベントバッファリング。L53-59: キュー容量の設定 |

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

SparkListenerBusでのパターンマッチによるリスナーメソッド呼び出しを把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | L28-102: doPostEvent()メソッド。L32-33: SparkListenerStageSubmittedのパターンマッチ → listener.onStageSubmitted()の呼び出し |

#### Step 5: リスナーインターフェースを理解する

SparkListenerInterfaceの定義とデフォルト実装を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | L304: SparkListenerInterfaceトレイト。L313-314: onStageSubmitted()メソッド定義。L506-509: SparkListenerクラスのデフォルト（no-op）実装 |

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

```
DAGScheduler.submitMissingTasks() [DAGScheduler.scala L1688]
    |
    +-- listenerBus.post(SparkListenerStageSubmitted(...))
            |
            +-- LiveListenerBus.post() [LiveListenerBus.scala L126]
                    |
                    +-- postToQueues() [LiveListenerBus.scala L155]
                            |
                            +-- AsyncEventQueue.post() [各キューに投入]
                                    |
                                    +-- LinkedBlockingQueue.put()
                                            |
                                            +-- [配信スレッド] doPostEvent() [SparkListenerBus.scala L28]
                                                    |
                                                    +-- listener.onStageSubmitted() [各リスナー実装]
                                                            |
                                                            +-- AppStatusListener.onStageSubmitted()
                                                            +-- EventLoggingListener (onOtherEvent/logEvent)
                                                            +-- ExecutorAllocationManager
```

### データフロー図

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

DAGScheduler              LiveListenerBus                    各リスナー
(StageInfo,            (AsyncEventQueue x N)
 Properties)
                                                              AppStatusListener
stage.latestInfo -----> SparkListenerStageSubmitted -------> (KVStore更新 -> Spark UI)
Utils.cloneProperties      |
                           +--[shared queue]----------------> ユーザー登録リスナー
                           +--[appStatus queue]-------------> AppStatusListener
                           +--[executorManagement queue]----> ExecutorAllocationManager
                           +--[eventLog queue]--------------> EventLoggingListener
                                                              (イベントログファイル)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListenerEvent.scala | `common/utils/src/main/scala/org/apache/spark/scheduler/SparkListenerEvent.scala` | ソース | イベント基底トレイト定義 |
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | SparkListenerStageSubmittedケースクラス定義、SparkListenerInterface/SparkListenerクラス |
| DAGScheduler.scala | `core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala` | ソース | イベント発火元（submitMissingTasks） |
| LiveListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala` | ソース | イベントバスの非同期配信制御 |
| AsyncEventQueue.scala | `core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.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` | ソース | Spark UI用のステータス更新リスナー |
