# 通知設計書 33-SparkListenerNodeBlacklistedForStage

## 概要

本ドキュメントは、Apache SparkにおけるSparkListenerNodeBlacklistedForStageイベント通知の設計について記述する。本イベントは特定ステージにおいてノード（ホスト）がブラックリスト（除外）された際に発火する非推奨イベントであり、Spark 3.1.0以降はSparkListenerNodeExcludedForStageに置き換えられている。

### 本通知の処理概要

SparkListenerNodeBlacklistedForStageは、特定ステージの実行中にノード上の除外Executor数がステージレベルの閾値を超えた際に発火するイベント通知である。

**業務上の目的・背景**：ステージレベルでのノード除外は、あるノード上の複数Executorが繰り返し問題を起こしている場合に、ノード全体をそのステージから排除するために必要である。これにより、ハードウェア障害やネットワーク問題を抱えるノードへのタスク割り当てを防ぎ、ステージの処理効率を向上させる。

**通知の送信タイミング**：TaskSetExcludelist.updateExcludedForFailedTask内で、あるExecutorがステージレベルで除外された後、そのノード上の除外Executor数が`spark.excludeOnFailure.stage.maxFailedExecutorsPerNode`の閾値に達した時点で発火する。

**通知の受信者**：LiveListenerBusに登録された全てのSparkListenerInterface実装。

**通知内容の概要**：除外されたノードのホストID、除外発生時刻、除外Executor数、対象のステージID、ステージ試行IDが含まれる。

**期待されるアクション**：Spark UIでステージレベルのノード除外状態が表示される。運用者はこの情報からノードレベルの問題を検知し、ハードウェア点検やノード入替の判断材料とする。

## 通知種別

アプリ内通知（Sparkイベントバスによるインプロセス非同期イベント配信）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 中 |
| リトライ | なし |

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

LiveListenerBusに登録された全てのSparkListenerInterface実装に対してブロードキャスト配信される。SparkListenerBus.doPostEventのパターンマッチにより`onNodeBlacklistedForStage`メソッドが呼び出される。

## 通知テンプレート

### イベントオブジェクト

| 項目 | 内容 |
|-----|------|
| イベントクラス | `SparkListenerNodeBlacklistedForStage` |
| パッケージ | `org.apache.spark.scheduler` |
| 親クラス | `SparkListenerEvent` |
| シリアライズ形式 | JSON（JsonProtocol経由） |

### イベントデータ構造

```scala
@deprecated("use SparkListenerNodeExcludedForStage instead", "3.1.0")
case class SparkListenerNodeBlacklistedForStage(
    time: Long,
    hostId: String,
    executorFailures: Int,
    stageId: Int,
    stageAttemptId: Int)
  extends SparkListenerEvent
```

### 添付ファイル

該当なし

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| time | イベント発生時刻（ミリ秒） | `clock.getTimeMillis()` | Yes |
| hostId | 除外対象のノードホスト名 | TaskSetExcludelist内部管理 | Yes |
| executorFailures | 除外Executor数 | excludedExecutorsOnNode.size | Yes |
| stageId | 対象ステージID | TaskSetExcludelist.stageId | Yes |
| stageAttemptId | 対象ステージ試行ID | TaskSetExcludelist.stageAttemptId | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | Executorのステージ除外発生 | ノード上の除外Executor数 >= MAX_FAILED_EXEC_PER_NODE_STAGE | TaskSetExcludelist.updateExcludedForFailedTask内で発火 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 既にノード除外済み | excludedNodesに既に追加済みのノードは再度イベントが発行されない |
| isDryRun=true | ドライラン時は除外が実行されない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[Executorがステージ除外される] --> B[ノード上の除外Executor数を確認]
    B --> C{除外Executor数 >= MAX_FAILED_EXEC_PER_NODE_STAGE?}
    C -->|Yes| D{既にノード除外済み?}
    C -->|No| Z[処理終了]
    D -->|No| E[excludedNodesにノード追加]
    D -->|Yes| Z
    E --> F[SparkListenerNodeBlacklistedForStage発行]
    E --> G[SparkListenerNodeExcludedForStage発行]
    F --> H[LiveListenerBusでリスナーに配信]
    G --> H
    H --> Z
```

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

該当なし（インメモリのデータ構造を使用）

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| イベント配信失敗 | ListenerBusキュー溢れ | LiveListenerBusのドロップポリシーに従う |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし

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

- 本イベントはクラスタ内部の情報（ホスト名、Executor失敗数）のみを含み、個人情報は含まない
- LiveListenerBusはJVMプロセス内での配信のみ

## 備考

- 本イベントは`@deprecated`であり、SparkListenerNodeExcludedForStageが推奨される
- 両イベントはTaskSetExcludelist内で同時発行される（160-163行目）
- ノード除外はExecutorのステージ除外を前提条件とする（Executorが先に除外され、その後ノード除外判定が行われる）

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 163-171行目: SparkListenerNodeBlacklistedForStageのcase class定義。hostId（ノード識別子）を使用 |

**読解のコツ**: Executor除外イベント（executorId）とノード除外イベント（hostId）でフィールド名が異なる点に注意。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | TaskSetExcludeList.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSetExcludeList.scala` | 110-168行目: updateExcludedForFailedTaskメソッド。Executor除外の後にノード除外判定が続く |

**主要処理フロー**:
1. **138行目**: Executorのステージ除外判定
2. **144-145行目**: ノード上の除外Executor集合を取得
3. **154行目**: `numFailExec >= MAX_FAILED_EXEC_PER_NODE_STAGE`でノード除外閾値判定
4. **160-161行目**: `SparkListenerNodeBlacklistedForStage`をpost
5. **162-163行目**: `SparkListenerNodeExcludedForStage`をpost

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | 68-69行目: doPostEvent内でSparkListenerNodeBlacklistedForStageをonNodeBlacklistedForStageにルーティング |

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

```
TaskSetManager (タスク失敗処理)
    |
    +-- TaskSetExcludelist.updateExcludedForFailedTask()
            |
            +-- (Executor除外判定・実行)
            |      +-- listenerBus.post(SparkListenerExecutorBlacklistedForStage)
            |      +-- listenerBus.post(SparkListenerExecutorExcludedForStage)
            |
            +-- (ノード除外判定・実行)
                   +-- excludedExecutorsOnNode.size >= MAX_FAILED_EXEC_PER_NODE_STAGE
                   +-- listenerBus.post(SparkListenerNodeBlacklistedForStage)
                   +-- listenerBus.post(SparkListenerNodeExcludedForStage)
```

### データフロー図

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

Executorステージ除外    TaskSetExcludelist                        SparkListenerNode
(host, exec,      ---> .updateExcludedForFailedTask()  --->     BlacklistedForStage
 numFailExec)          ノード除外閾値判定                            イベント
                                                                    |
                                                                    v
                                                             LiveListenerBus
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | ソース | イベントクラス定義 |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベントディスパッチ |
| TaskSetExcludeList.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSetExcludeList.scala` | ソース | ステージレベル除外ロジック・イベント発行元 |
| AppStatusListener.scala | `core/src/main/scala/org/apache/spark/status/AppStatusListener.scala` | ソース | UI向けイベント受信 |
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | ソース | イベントログへの永続化 |
