# 通知設計書 13-SparkListenerExecutorAdded

## 概要

本ドキュメントは、Apache SparkのSparkListenerExecutorAddedイベント通知の設計仕様を記載する。このイベントはドライバが新しいExecutorを登録した際に発火し、LiveListenerBus経由で非同期に配信される。

### 本通知の処理概要

SparkListenerExecutorAddedは、クラスタに新しいExecutorが追加され、ドライバに登録された際にリスナーバスへ通知を行うイベントである。

**業務上の目的・背景**：Sparkクラスタでは、Executorはジョブの実行を担う計算リソースの基本単位である。動的リソース割り当て（Dynamic Allocation）やクラスタマネージャ（YARN、Kubernetes、Standalone等）を通じてExecutorが追加される。この通知により、Spark UIへのExecutor情報の表示、イベントログへの記録、HeartbeatReceiverへのExecutor登録、動的リソース管理でのExecutor追跡が可能となる。

**通知の送信タイミング**：CoarseGrainedSchedulerBackendのRegisterExecutorメッセージ処理時に`listenerBus.post(SparkListenerExecutorAdded(...))`が呼び出される（313行目）。またLocalSchedulerBackendのstart()メソッドでも呼び出される（138行目）。

**通知の受信者**：LiveListenerBusに登録された全てのSparkListenerInterface実装クラス。主要な受信者はAppStatusListener（UI更新）、EventLoggingListener（イベントログ）、HeartbeatReceiver（ハートビート管理）、ExecutorMonitor（動的リソース管理）である。

**通知内容の概要**：Executor追加の時刻、ExecutorのID、ExecutorInfo（ホスト、コア数、ログURL、属性情報、リソース情報、リソースプロファイルID）が含まれる。

**期待されるアクション**：受信者はExecutorの情報をUIに表示し、HeartbeatReceiverは当該Executorのハートビート監視を開始し、ExecutorMonitorはアイドル時間の追跡を開始する。

## 通知種別

アプリ内イベント通知（SparkListenerEvent）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由、AsyncEventQueue使用） |
| 優先度 | 中（通常のイベントキュー処理） |
| リトライ | なし |

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

LiveListenerBusに登録された全てのAsyncEventQueueに対してブロードキャスト配信される。SparkListenerBus.doPostEventメソッド（62行目）でイベントタイプに基づくパターンマッチングが行われ、`listener.onExecutorAdded(executorAdded)`が呼び出される。

## 通知テンプレート

### メール通知の場合

該当なし。本通知はSpark内部のイベントバスを通じたプログラム間通知である。

### 本文テンプレート

```
イベント種別: SparkListenerExecutorAdded
JSON形式:
{
  "Event": "SparkListenerExecutorAdded",
  "Timestamp": <timestamp>,
  "Executor ID": "<executorId>",
  "Executor Info": {
    "Host": "<host>",
    "Total Cores": <totalCores>,
    "Log Urls": { ... },
    "Attributes": { ... },
    "Resources": { ... },
    "Resource Profile Id": <rpId>,
    "Registration Time": <registrationTs>,
    "Request Time": <requestTs>
  }
}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| time | Executor追加時刻（ミリ秒） | System.currentTimeMillis() | Yes |
| executorId | ExecutorのID文字列 | RegisterExecutorメッセージ | Yes |
| executorInfo | Executor詳細情報 | ExecutorInfoオブジェクト | Yes |
| executorInfo.executorHost | ホスト名 | ExecutorInfo.executorHost | Yes |
| executorInfo.totalCores | 利用可能コア数 | ExecutorInfo.totalCores | Yes |
| executorInfo.logUrlMap | ログURL | ExecutorInfo.logUrlMap | No |
| executorInfo.attributes | Executor属性 | ExecutorInfo.attributes | No |
| executorInfo.resourcesInfo | リソース情報 | ExecutorInfo.resourcesInfo | No |
| executorInfo.resourceProfileId | リソースプロファイルID | ExecutorInfo.resourceProfileId | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | RegisterExecutorメッセージ受信 | Executorからの登録要求が成功した場合 | CoarseGrainedSchedulerBackend 313行目 |
| 内部処理 | LocalSchedulerBackend起動 | ローカルモードでExecutorが作成された場合 | LocalSchedulerBackend.start() 138行目 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| LiveListenerBus停止済み | stop()後はイベントがドロップされる |
| ExecutorID重複 | 既に同じIDのExecutorが登録済みの場合、RegisterExecutor処理自体が拒否される |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[Executor起動・登録要求] --> B[CoarseGrainedSchedulerBackend.DriverEndpoint.receiveAndReply]
    B --> C[RegisterExecutorメッセージ処理]
    C --> D[executorDataMapにExecutorData追加]
    D --> E[listenerBus.post SparkListenerExecutorAdded]
    E --> F[LiveListenerBus.post]
    F --> G[各AsyncEventQueueにイベント配信]
    G --> H[SparkListenerBus.doPostEvent]
    H --> I{受信リスナー}
    I -->|AppStatusListener| J[UI Executor情報追加]
    I -->|EventLoggingListener| K[イベントログ書き込み]
    I -->|HeartbeatReceiver| L[ハートビート監視開始]
    I -->|ExecutorMonitor| M[Executor追跡開始]
```

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

### 参照テーブル一覧

| テーブル名 | 用途 | 備考 |
|-----------|------|------|
| executorDataMap（インメモリMap） | 既存Executor情報の確認 | CoarseGrainedSchedulerBackend内部のHashMap |

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

#### executorDataMap

| 参照項目（カラム名） | 用途 | 取得条件 |
|-------------------|------|---------|
| executorId | 重複登録チェック | RegisterExecutorメッセージのexecutorIdで検索 |

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| executorDataMap（インメモリMap） | INSERT | 新規ExecutorDataの追加 |
| KVStore（AppStatusListener経由） | INSERT/UPDATE | Executor情報の登録・更新 |

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

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| INSERT | Event | SparkListenerExecutorAdded | EventLoggingListenerによりイベントログに書き込み |
| INSERT | Timestamp | System.currentTimeMillis() | Executor追加時刻 |
| INSERT | Executor ID | executorId | ExecutorのID |
| INSERT | Executor Info | executorInfo | Executorの詳細情報 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| キュー満杯 | AsyncEventQueueのキャパシティ超過 | イベントがドロップされ、ドロップカウンターが増加 |
| リスナー例外 | リスナーのonExecutorAddedで例外発生 | ListenerBusが例外をキャッチしログに記録 |
| 重複登録 | 同一ExecutorIDの再登録 | 登録処理自体が拒否され、イベントは発火されない |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。Executorの追加が発生した任意のタイミングで配信される。

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

ExecutorInfoにはホスト名、ポート番号、ログURLなどのクラスタ内部情報が含まれる。Spark UIのACLやイベントログファイルのパーミッションにより適切なアクセス制限が必要。ログURLにはExecutorのstdout/stderrへのリンクが含まれる場合があり、ログ内容の機密性に応じた対策が必要。

## 備考

- SparkListenerExecutorAddedはSparkListenerExecutorRemovedと対になるイベントであり、Executorのライフサイクルを追跡するために使用される。
- 動的リソース割り当てが有効な場合、ExecutorMonitorがこのイベントを受信してExecutorのアイドル時間追跡を開始する。
- CoarseGrainedSchedulerBackendでは、ExecutorDataにregistrationTsやrequestTsも記録されるが、イベント自体にはExecutorInfoとして公開される情報のみが含まれる。

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 119行目: case class SparkListenerExecutorAdded(time, executorId, executorInfo)の定義 |
| 1-2 | ExecutorInfo.scala | `core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorInfo.scala` | ExecutorInfoのデータ構造（host, totalCores, logUrlMap, attributes, resourcesInfo, resourceProfileId） |

**読解のコツ**: ExecutorInfoはExecutorの静的情報（ホスト、コア数等）を保持するクラスで、動的なタスク実行情報は含まない。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | CoarseGrainedSchedulerBackend.scala | `core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala` | 313行目: RegisterExecutor処理内でlistenerBus.postが呼ばれる |
| 2-2 | LocalSchedulerBackend.scala | `core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala` | 138行目: ローカルモードでのExecutor追加 |

**主要処理フロー**:
1. **300行目**: ExecutorDataオブジェクトの生成
2. **307行目**: executorDataMapへの追加
3. **312-313行目**: `listenerBus.post(SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data))` -- イベント発火

#### Step 3: イベントバスの配信メカニズムを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | LiveListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala` | 126行目: postメソッド |
| 3-2 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | 62行目: doPostEventでlistener.onExecutorAddedを呼び出す |

#### Step 4: 受信リスナーの処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | 215行目: onExecutorAddedでExecutor情報をKVStoreに登録 |
| 4-2 | HeartbeatReceiver.scala | `core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala` | 182行目: onExecutorAddedでハートビート管理に登録 |
| 4-3 | ExecutorMonitor.scala | `core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala` | 342行目: onExecutorAddedでExecutor追跡を開始 |
| 4-4 | EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | 185行目: イベントログに記録 |

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

```
Executor起動 --> CoarseGrainedSchedulerBackend.DriverEndpoint
    |
    +-- receiveAndReply(RegisterExecutor)
            |
            +-- ExecutorData作成・executorDataMapに追加
            +-- listenerBus.post(SparkListenerExecutorAdded)
                    |
                    +-- LiveListenerBus.post(event)
                            |
                            +-- postToQueues(event)
                                    |
                                    +-- SparkListenerBus.doPostEvent(listener, event)
                                            |
                                            +-- AppStatusListener.onExecutorAdded()
                                            +-- HeartbeatReceiver.onExecutorAdded()
                                            +-- ExecutorMonitor.onExecutorAdded()
                                            +-- EventLoggingListener.onExecutorAdded()
```

### データフロー図

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

RegisterExecutor ---------> CoarseGrainedSchedulerBackend ------> SparkListenerExecutorAdded
メッセージ                    .DriverEndpoint                        イベント
(executorId, host,               |
 cores, logUrls, ...)            v
                           LiveListenerBus.post() --------------> AsyncEventQueue
                                                                       |
                                                                       v
                                                                 AppStatusListener --> KVStore
                                                                 HeartbeatReceiver --> タイマー登録
                                                                 ExecutorMonitor --> 追跡開始
                                                                 EventLoggingListener --> ログ
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベントcase classの定義 |
| CoarseGrainedSchedulerBackend.scala | `core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala` | ソース | 主要な発火元（分散モード） |
| LocalSchedulerBackend.scala | `core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala` | ソース | ローカルモードでの発火元 |
| LiveListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala` | ソース | 非同期イベントバス |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベントディスパッチ |
| AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | ソース | UI用データの更新 |
| HeartbeatReceiver.scala | `core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala` | ソース | ハートビート管理 |
| ExecutorMonitor.scala | `core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala` | ソース | 動的リソース管理 |
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | ソース | イベントログ記録 |
| JsonProtocol.scala | `core/src/main/scala/org/apache/spark/util/JsonProtocol.scala` | ソース | JSON変換 |
