# 通知設計書 32-SparkListenerExecutorBlacklistedForStage

## 概要

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

### 本通知の処理概要

SparkListenerExecutorBlacklistedForStageは、特定ステージの実行中にExecutorのタスク失敗数がステージレベルの閾値を超えた際に発火するイベント通知である。

**業務上の目的・背景**：ステージレベルでの障害分離は、アプリケーション全体の除外よりも細かい粒度で問題のあるExecutorを排除するために必要である。特定のステージで特定のExecutorが問題を起こしている場合、そのステージに限定して除外することで、他のステージでの処理には影響を与えない。TaskSetExcludelistクラスがステージ単位の除外管理を担当する。

**通知の送信タイミング**：TaskSetExcludelist.updateExcludedForFailedTask内で、あるExecutor上でのユニークタスク失敗数が`spark.excludeOnFailure.stage.maxFailedTasksPerExecutor`の閾値に達した時点で発火する。

**通知の受信者**：LiveListenerBusに登録された全てのSparkListenerInterface実装。主にAppStatusListener、EventLoggingListenerなど。

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

**期待されるアクション**：Spark UIでステージレベルの除外状態が表示される。運用者はこの情報から特定ステージでの問題を迅速に特定できる。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 中 |
| リトライ | なし（イベントバスのキューイングにより配信保証） |

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

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

## 通知テンプレート

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

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

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

```scala
@deprecated("use SparkListenerExecutorExcludedForStage instead", "3.1.0")
case class SparkListenerExecutorBlacklistedForStage(
    time: Long,
    executorId: String,
    taskFailures: Int,
    stageId: Int,
    stageAttemptId: Int)
  extends SparkListenerEvent
```

### 添付ファイル

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

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| time | イベント発生時刻（ミリ秒） | `clock.getTimeMillis()` | Yes |
| executorId | 除外対象のExecutor ID | TaskSetExcludelist内部管理 | Yes |
| taskFailures | ユニークタスク失敗数 | ExecutorFailuresInTaskSet.numUniqueTasksWithFailures | Yes |
| stageId | 対象ステージID | TaskSetExcludelist.stageId | Yes |
| stageAttemptId | 対象ステージ試行ID | TaskSetExcludelist.stageAttemptId | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | タスク失敗発生 | Executor上のユニークタスク失敗数 >= MAX_FAILURES_PER_EXEC_STAGE | TaskSetExcludelist.updateExcludedForFailedTask内で発火 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 既にステージ除外済み | excludedExecsに既に追加済みのExecutorは再度イベントが発行されない |
| isDryRun=true | ドライラン時はタスク/ステージレベルの除外が実行されない（情報収集のみ） |
| excludeOnFailure無効 | タスク/ステージレベルの除外機能が無効の場合はTaskSetExcludelistが生成されない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[タスク失敗発生] --> B[TaskSetExcludelist.updateExcludedForFailedTask]
    B --> C[Executor失敗カウント更新]
    C --> D{ユニーク失敗数 >= MAX_FAILURES_PER_EXEC_STAGE?}
    D -->|Yes| E{既にステージ除外済み?}
    D -->|No| Z[処理終了]
    E -->|No| F[excludedExecsにExecutor追加]
    E -->|Yes| Z
    F --> G[SparkListenerExecutorBlacklistedForStage発行]
    F --> H[SparkListenerExecutorExcludedForStage発行]
    G --> I[LiveListenerBusでリスナーに配信]
    H --> I
    I --> J{ノード除外閾値到達?}
    J -->|Yes| K[SparkListenerNodeBlacklistedForStage発行]
    J -->|No| Z
    K --> Z
```

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

### 参照テーブル一覧

該当なし（RDBMSは使用しない。インメモリのデータ構造を参照）

### インメモリデータ構造

| データ構造 | 用途 | 備考 |
|-----------|------|------|
| execToFailures | Executor毎のタスク失敗情報 | HashMap[String, ExecutorFailuresInTaskSet] |
| excludedExecs | ステージレベルで除外済みのExecutor集合 | HashSet[String] |
| excludedNodes | ステージレベルで除外済みのノード集合 | HashSet[String] |

### 更新テーブル一覧

該当なし（インメモリ管理）

## エラー処理

### エラーケース一覧

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

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし（イベント発生時に即座に配信）

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

- 本イベントはクラスタ内部の情報のみを含み、個人情報は含まない
- LiveListenerBusはJVMプロセス内での配信であり、ネットワーク経由の送信は行わない

## 備考

- 本イベントは`@deprecated`アノテーション付きであり、Spark 3.1.0以降はSparkListenerExecutorExcludedForStageを使用することが推奨される
- 後方互換性のため、TaskSetExcludelistでは両方のイベントが同時にpostされる（149-152行目）
- ステージレベルの除外はアプリケーションレベルの除外（HealthTracker管理）とは独立して動作する

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 142-151行目: SparkListenerExecutorBlacklistedForStageのcase class定義。stageId, stageAttemptIdを追加で持つ |
| 1-2 | SparkListener.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala` | 153-161行目: 後継のSparkListenerExecutorExcludedForStageの定義 |

**読解のコツ**: アプリケーションレベル（No.31のSparkListenerExecutorBlacklisted）との違いは、stageIdとstageAttemptIdの有無である。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | TaskSetExcludeList.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSetExcludeList.scala` | 41-47行目: TaskSetExcludelistクラスの定義。listenerBus, stageId, stageAttemptIdを保持 |
| 2-2 | TaskSetExcludeList.scala | `core/src/main/scala/org/apache/spark/scheduler/TaskSetExcludeList.scala` | 110-168行目: updateExcludedForFailedTaskメソッド。メイン処理ロジック |

**主要処理フロー**:
1. **138行目**: `numFailures >= MAX_FAILURES_PER_EXEC_STAGE`で閾値判定
2. **149-150行目**: `SparkListenerExecutorBlacklistedForStage`をpost
3. **151-152行目**: `SparkListenerExecutorExcludedForStage`をpost
4. **154行目**: ノード除外閾値の判定に進む

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

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

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

```
TaskSetManager (タスク失敗処理)
    |
    +-- TaskSetExcludelist.updateExcludedForFailedTask()
            |
            +-- ExecutorFailuresInTaskSet.updateWithFailure()
            +-- numUniqueTasksWithFailures >= MAX_FAILURES_PER_EXEC_STAGE
            +-- listenerBus.post(SparkListenerExecutorBlacklistedForStage)
            +-- listenerBus.post(SparkListenerExecutorExcludedForStage)
            |
            +-- (ノード除外判定)
                +-- listenerBus.post(SparkListenerNodeBlacklistedForStage)
                +-- listenerBus.post(SparkListenerNodeExcludedForStage)
```

### データフロー図

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

タスク失敗情報            TaskSetExcludelist                     SparkListenerExecutor
(host, exec,       --->  .updateExcludedForFailedTask()  --->   BlacklistedForStage
 taskIndex,              失敗カウント・閾値判定                      イベント
 failureReason)                                                    |
                                                                   v
                                                            LiveListenerBus
                                                                   |
                                                            +------+------+
                                                            |             |
                                                      AppStatus     EventLogging
                                                      Listener      Listener
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| 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` | ソース | ステージレベル除外ロジック・イベント発行元 |
| HealthTracker.scala | `core/src/main/scala/org/apache/spark/scheduler/HealthTracker.scala` | ソース | アプリレベル除外（ステージ完了後にTaskSetExcludelistの情報を引き継ぐ） |
| 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` | ソース | イベントログへの永続化 |
