# 通知設計書 78-StreamingListenerBatchStarted

## 概要

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

### 本通知の処理概要

StreamingListenerBatchStartedは、DStreamベースのSpark Streamingにおいて、バッチ内の最初のジョブが実際に処理を開始した際に発火されるイベント通知である。BatchInfoオブジェクトを内包し、バッチ時間、送信時刻、処理開始時刻などの情報が含まれる。バッチの送信（Submitted）と開始（Started）の間のスケジューリング遅延を計測するための重要なイベントである。

**業務上の目的・背景**：バッチの送信と処理開始の間には、ジョブキューの待機時間（スケジューリング遅延）が存在する。StreamingListenerBatchStartedにより、スケジューリング遅延の実測値を取得し、処理のボトルネック分析、リソース不足の検知、並行ジョブ数の最適化判断に利用できる。スケジューリング遅延が大きい場合は、numConcurrentJobsの増加やリソースの追加が必要であることを示唆する。

**通知の送信タイミング**：JobScheduler.handleJobStartメソッドにおいて、バッチの最初のジョブが開始された際（isFirstJobOfJobSet=true）にのみ発火される。handleJobStartの呼び出し後にイベントが送信されるため、正確なprocessingStartTimeが含まれる。

**通知の受信者**：StreamingListenerインタフェースを実装し、StreamingContext.addStreamingListener()で登録されたすべてのリスナーが受信する。onBatchStartedメソッドにはデフォルト実装（空メソッド）がある。

**通知内容の概要**：BatchInfoオブジェクトが含まれ、batchTime、streamIdToInputInfo、submissionTime、processingStartTime（設定済み）、processingEndTime（None）、outputOperationInfosが格納される。送信時点ではprocessingEndTimeは未設定。

**期待されるアクション**：リスナーは通知を受けて、スケジューリング遅延の計算・記録、処理開始のログ記録、UIの更新、バッチ処理時間の計測開始などを実行することが期待される。

## 通知種別

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

## 送信仕様

### 基本情報

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

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

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

## 通知テンプレート

### メール通知の場合

該当なし。

### 本文テンプレート

```
イベント型: StreamingListenerBatchStarted
バッチ時間: {batchTime}
送信時刻: {submissionTime}
処理開始時刻: {processingStartTime}
スケジューリング遅延: {schedulingDelay}ms
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| batchInfo | バッチ情報オブジェクト | JobSet.toBatchInfo | Yes |
| batchInfo.batchTime | バッチの時間 | JobSet.time | Yes |
| batchInfo.submissionTime | ジョブセット送信時刻 | JobSet.submissionTime | Yes |
| batchInfo.processingStartTime | 処理開始時刻 | JobSet.processingStartTime | Yes |
| batchInfo.schedulingDelay | スケジューリング遅延 | processingStartTime - submissionTime | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| ジョブ開始 | JobScheduler.handleJobStart | バッチの最初のジョブ（isFirstJobOfJobSet=true） | バッチ内の2番目以降のジョブ開始では発火しない |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| 最初のジョブでない | isFirstJobOfJobSetがfalseの場合（バッチの2番目以降のジョブ開始）はイベントは発火されない |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[JobHandler.run実行] --> B[eventLoop.post JobStarted]
    B --> C[processEvent: handleJobStart呼び出し]
    C --> D{isFirstJobOfJobSet?}
    D -->|No| E[jobSet.handleJobStart のみ]
    D -->|Yes| F[jobSet.handleJobStart]
    F --> G[listenerBus.post StreamingListenerBatchStarted]
    G --> H[WrappedStreamingListenerEventにラップ]
    H --> I[SparkListenerBus.post 非同期]
    I --> J[listener.onBatchStarted]
```

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

### 参照テーブル一覧

該当なし。

### 更新テーブル一覧

該当なし。

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

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

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | リスナーのonBatchStartedメソッドで例外発生 | ListenerBusが例外をキャッチしてログ出力。他のリスナーへの配信は継続 |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし。

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

- BatchInfoには入力ストリーム情報が含まれるが、データ内容自体は含まれない
- イベントはSparkプロセス内部でのみ配信される

## 備考

- handleJobStartの呼び出し後にイベントが送信される設計により、正確なprocessingStartTimeが確保される（186-188行目のコメント参照）
- isFirstJobOfJobSetは`!jobSet.hasStarted`で判定される
- 1つのバッチに複数のジョブが含まれる場合でも、BatchStartedイベントは1回のみ発火

---

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

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

### 推奨読解順序

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

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

**読解のコツ**: BatchStarted時点ではprocessingStartTimeが設定済みのため、schedulingDelayが計算可能。processingEndTimeはNoneのためprocessingDelay/totalDelayは計算不可。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | JobScheduler.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala` | handleJobStartメソッド（181-194行目）。isFirstJobOfJobSetの判定とイベント発火の条件を確認 |

**主要処理フロー**:
1. **182行目**: `val jobSet = jobSets.get(job.time)` - ジョブセット取得
2. **183行目**: `val isFirstJobOfJobSet = !jobSet.hasStarted` - 初回判定
3. **184行目**: `jobSet.handleJobStart(job)` - ジョブ開始処理
4. **186-188行目**: handleJobStart呼び出し後にprocessingStartTimeが設定される
5. **188行目**: `listenerBus.post(StreamingListenerBatchStarted(jobSet.toBatchInfo))` - イベント発火

#### Step 3: リスナーバスの配信メカニズムを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StreamingListenerBus.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/StreamingListenerBus.scala` | doPostEvent（60-61行目）でlistener.onBatchStartedを呼び出し |

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

```
JobHandler.run()
    |
    +-- eventLoop.post(JobStarted(job, clock.getTimeMillis()))
            |
            +-- processEvent(JobStarted)
                    |
                    +-- handleJobStart(job, startTime)
                            |
                            +-- jobSet.handleJobStart(job)
                            |
                            +-- [isFirstJobOfJobSet の場合のみ]
                            |       |
                            |       +-- listenerBus.post(StreamingListenerBatchStarted(jobSet.toBatchInfo))
                            |               |
                            |               +-- sparkListenerBus.post(WrappedStreamingListenerEvent(event))
                            |                       |
                            |                       +-- listener.onBatchStarted(event)
                            |
                            +-- job.setStartTime(startTime)
                            |
                            +-- listenerBus.post(StreamingListenerOutputOperationStarted(...))
```

### データフロー図

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

JobStarted(job, startTime) --> handleJobStart()           --> StreamingListenerBatchStarted
                                    |                              |
                                    v                              v
                               jobSet.handleJobStart()       StreamingListener
                               [processingStartTime設定]       .onBatchStarted()
                                    |
                                    v
                               jobSet.toBatchInfo()
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StreamingListener.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/StreamingListener.scala` | ソース | イベントクラス定義（42行目）、リスナーインタフェース（88行目） |
| 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` | ソース | イベント発火元handleJobStart（181-194行目） |
| StreamingListenerBus.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/StreamingListenerBus.scala` | ソース | イベント配信バス（60-61行目） |
