# 通知設計書 79-StreamingListenerBatchCompleted

## 概要

本ドキュメントは、Apache SparkのDStream（Discretized Stream）ベースのStreaming処理において、バッチジョブの処理が完了した際に発火されるイベント通知「StreamingListenerBatchCompleted」の設計を記述する。

### 本通知の処理概要

StreamingListenerBatchCompletedは、DStreamベースのSpark Streamingにおいて、バッチ内のすべてのジョブの処理が完了した際に発火されるイベント通知である。BatchInfoオブジェクトを内包し、バッチ時間、送信時刻、処理開始時刻、処理終了時刻、各出力操作の情報などの完全なバッチ処理メトリクスが含まれる。

**業務上の目的・背景**：バッチの処理完了は、DStream Streamingの運用監視における最も重要な計測ポイントである。StreamingListenerBatchCompletedにより、処理時間（processingDelay）、スケジューリング遅延（schedulingDelay）、総遅延（totalDelay）の実測値を取得し、SLA遵守の確認、パフォーマンストレンドの分析、ボトルネックの特定に利用できる。処理時間がバッチ間隔を超過する場合はバックプレッシャ状態を示唆し、対応が必要となる。

**通知の送信タイミング**：JobScheduler.handleJobCompletionメソッドにおいて、バッチ内のすべてのジョブが完了した際（jobSet.hasCompleted=true）に発火される。最後のジョブの完了処理後に送信されるため、processingEndTimeを含む完全なBatchInfoが配信される。

**通知の受信者**：StreamingListenerインタフェースを実装し、StreamingContext.addStreamingListener()で登録されたすべてのリスナーが受信する。onBatchCompletedメソッドにはデフォルト実装（空メソッド）がある。RateControllerもリスナーとして登録されており、スループットに基づくレート調整に利用する。

**通知内容の概要**：BatchInfoオブジェクトが含まれ、batchTime、streamIdToInputInfo、submissionTime、processingStartTime（設定済み）、processingEndTime（設定済み）、outputOperationInfosが格納される。すべてのフィールドが設定済みのため、schedulingDelay, processingDelay, totalDelayの各計算プロパティが利用可能。

**期待されるアクション**：リスナーは通知を受けて、バッチ処理メトリクスの記録、SLA確認、パフォーマンスダッシュボードの更新、レート制御の調整、処理遅延アラートの判定などを実行することが期待される。

## 通知種別

アプリ内通知（StreamingListenerBus経由。WrappedStreamingListenerEventとしてSpark LiveListenerBusにラップして非同期配信）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（StreamingListenerBus経由でSpark LiveListenerBusにラップして非同期配信） |
| 優先度 | 中 |
| リトライ | 無し |

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

StreamingListenerBusに登録されたすべてのStreamingListenerに対して配信される。

## 通知テンプレート

### メール通知の場合

該当なし。

### 本文テンプレート

```
イベント型: StreamingListenerBatchCompleted
バッチ時間: {batchTime}
送信時刻: {submissionTime}
処理開始時刻: {processingStartTime}
処理終了時刻: {processingEndTime}
スケジューリング遅延: {schedulingDelay}ms
処理時間: {processingDelay}ms
総遅延: {totalDelay}ms
入力レコード数: {numRecords}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| batchInfo | バッチ情報オブジェクト | JobSet.toBatchInfo | Yes |
| batchInfo.batchTime | バッチの時間 | JobSet.time | Yes |
| batchInfo.submissionTime | ジョブセット送信時刻 | JobSet.submissionTime | Yes |
| batchInfo.processingStartTime | 処理開始時刻 | JobSet.processingStartTime | Yes |
| batchInfo.processingEndTime | 処理終了時刻 | JobSet.processingEndTime | Yes |
| batchInfo.schedulingDelay | スケジューリング遅延 | processingStartTime - submissionTime | Yes |
| batchInfo.processingDelay | 処理時間 | processingEndTime - processingStartTime | Yes |
| batchInfo.totalDelay | 総遅延 | schedulingDelay + processingDelay | Yes |
| batchInfo.numRecords | 入力レコード数合計 | streamIdToInputInfo集計 | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| ジョブ完了 | JobScheduler.handleJobCompletion | バッチの全ジョブ完了（jobSet.hasCompleted=true） | 最後のジョブ完了時に発火 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 全ジョブ未完了 | jobSet.hasCompletedがfalseの場合（バッチ内に未完了ジョブあり）はイベントは発火されない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[JobHandler.run完了] --> B[eventLoop.post JobCompleted]
    B --> C[processEvent: handleJobCompletion呼び出し]
    C --> D[jobSet.handleJobCompletion job]
    D --> E[job.setEndTime completedTime]
    E --> F[StreamingListenerOutputOperationCompleted発火]
    F --> G{jobSet.hasCompleted?}
    G -->|No| H[終了 他ジョブ待ち]
    G -->|Yes| I[listenerBus.post StreamingListenerBatchCompleted]
    I --> J[jobSets.remove jobSet.time]
    J --> K[jobGenerator.onBatchCompletion]
    K --> L[WrappedStreamingListenerEventにラップ]
    L --> M[SparkListenerBus.post 非同期]
    M --> N[listener.onBatchCompleted]
```

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

### 参照テーブル一覧

該当なし。

### 更新テーブル一覧

該当なし。

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

WrappedStreamingListenerEventのlogEventがfalseに設定されているため、Sparkイベントログには記録されない。

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | リスナーのonBatchCompletedメソッドで例外発生 | ListenerBusが例外をキャッチしてログ出力。他のリスナーへの配信は継続 |
| ジョブ失敗 | バッチ内のジョブが失敗した場合 | ジョブ失敗でもバッチ完了としてイベントが発火される。ただし失敗時はreportError後にイベントが送信される |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（バッチ間隔に依存） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。

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

- BatchInfoには入力ストリーム情報が含まれるが、データ内容自体は含まれない
- OutputOperationInfoにはfailureReasonが含まれる可能性があり、エラーメッセージに機密情報が含まれる場合がある
- イベントはSparkプロセス内部でのみ配信される

## 備考

- バッチ完了後、jobSetsからジョブセットが削除され、jobGenerator.onBatchCompletionが呼ばれてチェックポイント処理が開始される
- ジョブ失敗時もBatchCompletedが発火されるが、失敗情報はOutputOperationInfoのfailureReasonに記録される
- RateControllerがリスナーとして登録されている場合、BatchCompleted情報に基づいてバックプレッシャのレート調整が行われる
- StatsReportListenerがこのイベントをリッスンし、最近のバッチ統計情報を表示する

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StreamingListener.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/StreamingListener.scala` | StreamingListenerBatchCompletedのcase class定義（39行目）。batchInfo: BatchInfoフィールドを確認 |
| 1-2 | BatchInfo.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala` | BatchInfoの定義（35-42行目）とschedulingDelay（49行目）、processingDelay（55-56行目）、totalDelay（62-63行目）の各計算プロパティ |

**読解のコツ**: BatchCompleted時点ではprocessingStartTimeとprocessingEndTimeが共に設定済みのため、schedulingDelay, processingDelay, totalDelayのすべてが計算可能。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | JobScheduler.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala` | handleJobCompletionメソッド（196-219行目）。jobSet.hasCompletedの判定とイベント発火、その後のクリーンアップ処理を確認 |

**主要処理フロー**:
1. **197行目**: `val jobSet = jobSets.get(job.time)` - ジョブセット取得
2. **198行目**: `jobSet.handleJobCompletion(job)` - ジョブ完了処理
3. **199行目**: `job.setEndTime(completedTime)` - 終了時刻設定
4. **200行目**: OutputOperationCompleted発火
5. **203行目**: `if (jobSet.hasCompleted)` - 全ジョブ完了判定
6. **204行目**: `listenerBus.post(StreamingListenerBatchCompleted(jobSet.toBatchInfo))` - イベント発火
7. **211行目**: `jobSets.remove(jobSet.time)` - ジョブセット削除
8. **212行目**: `jobGenerator.onBatchCompletion(jobSet.time)` - チェックポイント処理

#### Step 3: リスナーバスの配信とリスナー実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StreamingListenerBus.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/StreamingListenerBus.scala` | doPostEvent（62-63行目）でlistener.onBatchCompletedを呼び出し |
| 3-2 | StreamingListener.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/StreamingListener.scala` | StatsReportListener（109-132行目）がBatchCompletedをリッスンして統計情報を表示 |

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

```
JobHandler.run()  [ジョブ実行完了]
    |
    +-- eventLoop.post(JobCompleted(job, clock.getTimeMillis()))
            |
            +-- processEvent(JobCompleted)
                    |
                    +-- handleJobCompletion(job, completedTime)
                            |
                            +-- jobSet.handleJobCompletion(job)
                            +-- job.setEndTime(completedTime)
                            +-- listenerBus.post(OutputOperationCompleted)
                            |
                            +-- [jobSet.hasCompleted の場合のみ]
                                    |
                                    +-- listenerBus.post(StreamingListenerBatchCompleted(jobSet.toBatchInfo))
                                    |       |
                                    |       +-- sparkListenerBus.post(WrappedStreamingListenerEvent(event))
                                    |               |
                                    |               +-- listener.onBatchCompleted(event)
                                    |
                                    +-- jobSets.remove(jobSet.time)
                                    +-- jobGenerator.onBatchCompletion(jobSet.time)
```

### データフロー図

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

JobCompleted(job, time)    --> handleJobCompletion()          --> StreamingListenerBatchCompleted
                                    |                                  |
                                    v                                  v
                               jobSet.handleJobCompletion()     StreamingListener
                               job.setEndTime()                   .onBatchCompleted()
                                    |                                  |
                                    v                                  v
                               jobSet.toBatchInfo()             RateController
                               [全フィールド設定済み]               (レート調整)
                                                                      |
                                                                      v
                                                                StatsReportListener
                                                                  (統計表示)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StreamingListener.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/StreamingListener.scala` | ソース | イベントクラス定義（39行目）、リスナーインタフェース（91行目）、StatsReportListener（109-132行目） |
| BatchInfo.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/BatchInfo.scala` | ソース | BatchInfoデータクラス定義 |
| JobScheduler.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala` | ソース | イベント発火元handleJobCompletion（196-219行目） |
| StreamingListenerBus.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/StreamingListenerBus.scala` | ソース | イベント配信バス（62-63行目） |
| RateController.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/RateController.scala` | ソース | バックプレッシャのレート制御リスナー |
| StreamingJobProgressListener.scala | `streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala` | ソース | UI用リスナー実装 |
