# 通知設計書 73-QueryProgressEvent

## 概要

本ドキュメントは、Apache SparkのStructured Streamingにおいて、ストリーミングクエリの進捗が更新された際に発火されるイベント通知「QueryProgressEvent」の設計を記述する。

### 本通知の処理概要

QueryProgressEventは、Structured Streamingのクエリがバッチ処理を完了するたびに発火されるイベント通知である。StreamingQueryProgressオブジェクトを内包し、入力レート、処理レート、ウォーターマーク情報、ソース・シンクのメトリクスなど、クエリの詳細な進捗情報を含む。

**業務上の目的・背景**：ストリーミングアプリケーションの継続的な運用監視において、各バッチの処理状況を把握することは不可欠である。QueryProgressEventにより、処理速度の監視、レイテンシの追跡、バックプレッシャの検知、データソースの健全性確認などのリアルタイムモニタリングが可能となる。処理遅延やスループット低下のアラート発報にも利用される。

**通知の送信タイミング**：ProgressReporter.updateProgressメソッドにおいて、マイクロバッチの処理が完了し進捗情報が更新された後に非同期で発火される。QueryStartedEventとは異なり、非同期配信であるため、リスナーがイベントを処理する時点ではクエリの状態が変化している可能性がある。

**通知の受信者**：StreamingQueryListenerインタフェースを実装し、SparkSessionのStreamingQueryManagerに登録されたすべてのリスナーが受信する。Streaming UIのStatusListenerも受信者の一つである。

**通知内容の概要**：StreamingQueryProgressオブジェクトが含まれ、id, runId, name, timestamp, batchId, batchDuration, durationMs, eventTime, stateOperators, sources, sink, observedMetricsなどの包括的な進捗メトリクスが格納される。

**期待されるアクション**：リスナーは通知を受けて、ダッシュボードの更新、メトリクスの外部システムへのエクスポート、閾値超過時のアラート発報、処理遅延の検知と対応などを実行することが期待される。

## 通知種別

アプリ内通知（StreamingQueryListenerBus + Spark LiveListenerBus経由の非同期イベント通知）

## 送信仕様

### 基本情報

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

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

StreamingQueryListenerBusに登録されたすべてのStreamingQueryListenerに対して配信される。doPostEvent内でactiveQueryRunIdsに含まれるrunIdのイベントのみがリスナーに配信される（shouldReportチェック）。

## 通知テンプレート

### メール通知の場合

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

### 本文テンプレート

```json
{
  "progress": {
    "id": "{id}",
    "runId": "{runId}",
    "name": "{name}",
    "timestamp": "{timestamp}",
    "batchId": {batchId},
    "batchDuration": {batchDuration},
    "durationMs": { ... },
    "eventTime": { ... },
    "stateOperators": [ ... ],
    "sources": [ ... ],
    "sink": { ... }
  }
}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| progress | クエリ進捗情報オブジェクト | StreamingQueryProgress | Yes |
| progress.id | クエリの永続一意ID | StreamExecution.id | Yes |
| progress.runId | 実行ごとの一意ID | StreamExecution.runId | Yes |
| progress.batchId | バッチID | ProgressContext.currentBatchId | Yes |
| progress.durationMs | 各処理段階の所要時間 | ProgressContext内で計測 | Yes |
| progress.sources | ソースごとの入力メトリクス | 各SparkDataStreamから取得 | Yes |
| progress.sink | シンクのメトリクス | Sinkから取得 | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| バッチ完了 | マイクロバッチの処理完了 | バッチ内でデータが処理された場合 | ProgressReporter.updateProgressで発火 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| データなしバッチ | データがないトリガーではQueryProgressEventではなくQueryIdleEventが発火される場合がある |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[マイクロバッチ処理完了] --> B[ProgressContext.finishTrigger]
    B --> C[StreamingQueryProgress生成]
    C --> D[ProgressReporter.updateProgress]
    D --> E[progressBufferに追加]
    E --> F[QueryProgressEvent生成]
    F --> G[postEvent SparkSession.streams.postListenerEvent]
    G --> H[StreamingQueryListenerBus.post]
    H --> I[SparkListenerBus.post 非同期]
    I --> J[doPostEvent: listener.onQueryProgress]
```

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

### 参照テーブル一覧

該当なし。本イベントはメモリ上の進捗情報から生成される。

### 更新テーブル一覧

該当なし。

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

Sparkイベントログが有効な場合、SparkListenerEventとしてイベントログに記録される。

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | リスナーのonQueryProgressメソッドで例外発生 | ListenerBusが例外をキャッチしてログ出力。他のリスナーへの配信は継続 |
| 進捗情報不完全 | メトリクス取得中のエラー | 部分的なメトリクス情報でProgressを生成 |

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（トリガー間隔に依存） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。バッチ完了時に発火される。

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

- StreamingQueryProgressにはクエリの物理プラン情報やソース/シンクの接続情報が含まれる可能性がある
- イベントログに記録される場合、接続文字列などの機密情報がログに残る可能性がある
- JSON形式でのシリアライズ機能を持つ

## 備考

- QueryProgressEventはStreamingQueryListener.Event traitを継承
- Since 2.1.0で導入
- 非同期配信のため、イベント受信時のクエリ状態と進捗情報が表す時点にずれがある可能性がある
- progressBufferには直近の進捗情報が保持され、recentProgressメソッドで取得可能
- StreamingQueryProgressのJSON表現にはJacksonとjson4sが使用される

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StreamingQueryListener.scala | `sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala` | QueryProgressEventクラス（208-215行目）の構造。progressフィールド（StreamingQueryProgress型）を確認 |

**読解のコツ**: QueryProgressEventは単一のprogressフィールドのみを持つシンプルなラッパーであり、実質的な情報はStreamingQueryProgressクラスに格納されている。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ProgressReporter.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ProgressReporter.scala` | updateProgressメソッド（93-100行目）でのイベント生成と配信。addNewProgressでバッファに追加後、postEventで配信 |

**主要処理フロー**:
1. **95行目**: `lastNoExecutionProgressEventTime = triggerClock.getTimeMillis()` - タイムスタンプ更新
2. **97行目**: `addNewProgress(newProgress)` - progressBufferに追加
3. **98行目**: `postEvent(new QueryProgressEvent(newProgress))` - イベント生成・ポスト

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StreamingQueryListenerBus.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryListenerBus.scala` | postメソッド（71-81行目）でのQueryProgressEventの処理。SparkListenerBusへの非同期ポスト（78-79行目） |
| 3-2 | StreamingQueryListenerBus.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryListenerBus.scala` | doPostEvent（133-135行目）でshouldReportチェック後にlistener.onQueryProgressを呼び出し |

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

```
MicroBatchExecution.runBatch()
    |
    +-- ProgressContext.finishTrigger()
            |
            +-- StreamingQueryProgress生成
                    |
                    +-- ProgressReporter.updateProgress(newProgress)
                            |
                            +-- addNewProgress(newProgress)
                            |
                            +-- postEvent(new QueryProgressEvent(newProgress))
                                    |
                                    +-- SparkSession.streams.postListenerEvent(event)
                                            |
                                            +-- StreamingQueryListenerBus.post(event)
                                                    |
                                                    +-- sparkListenerBus.post(event)  [非同期]
                                                            |
                                                            +-- doPostEvent(listener, event)
                                                                    |
                                                                    +-- listener.onQueryProgress(event)
```

### データフロー図

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

バッチ処理結果メトリクス  --->  ProgressContext         --->  StreamingQueryProgress
ソース入力情報                   .finishTrigger()               |
シンク出力情報                                                   v
ステートオペレータ情報            ProgressReporter        --->  QueryProgressEvent
                                .updateProgress()               |
                                                                v
                                StreamingQueryListenerBus  -->  StreamingQueryListener
                                .post()                         .onQueryProgress()
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StreamingQueryListener.scala | `sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala` | ソース | QueryProgressEventクラス定義（208-215行目） |
| ProgressReporter.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/ProgressReporter.scala` | ソース | イベント発火元（93-100行目） |
| StreamingQueryListenerBus.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryListenerBus.scala` | ソース | イベント配信バス（71-81行目、133-135行目） |
| StreamingQueryStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala` | ソース | UI用リスナー実装 |
| StreamExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecution.scala` | ソース | ストリーミングクエリ実行基盤 |
