# 通知設計書 72-QueryStartedEvent

## 概要

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

### 本通知の処理概要

QueryStartedEventは、Structured Streamingのクエリが`DataStreamWriter.start()`で開始された際に発火されるイベント通知である。クエリの一意識別子（id, runId）、ユーザー指定の名前、開始タイムスタンプ、ジョブタグなどの情報を含み、登録されたStreamingQueryListenerに配信される。

**業務上の目的・背景**：ストリーミングアプリケーションの運用監視において、クエリの開始を検知することは基本的な要件である。QueryStartedEventにより、クエリの開始をリアルタイムに検知し、監視ダッシュボードの更新、クエリ管理テーブルへの登録、アラート設定の初期化などのアクションを実行できる。restart時にもrunIdが新たに生成されるため、再起動の検知にも利用可能である。

**通知の送信タイミング**：StreamExecution.runStream()内で、クエリのアクティベーション処理中に同期的に発火される。`DataStreamWriter.start()`が呼び出し元に制御を返す前に、すべてのリスナーのonQueryStartedが呼び出されることが保証されている。

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

**通知内容の概要**：クエリの永続ID（id: UUID）、実行ID（runId: UUID）、ユーザー指定名（name: String、nullの場合あり）、開始タイムスタンプ（timestamp: String、ISO8601形式）、ジョブタグ（jobTags: Set[String]）が含まれる。

**期待されるアクション**：リスナーは通知を受けて、クエリ監視の開始、UI表示の更新、外部システムへのクエリ開始通知の送信などを実行することが期待される。ただし、本イベントは同期配信されるため、リスナーの処理がクエリの開始をブロックする点に注意が必要である。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 同期（QueryStartedEventのみ同期配信。start()が返る前にすべてのリスナーに配信完了） |
| 優先度 | 高 |
| リトライ | 無し |

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

StreamingQueryListenerBusに登録されたすべてのStreamingQueryListenerに対して配信される。QueryStartedEventは特別扱いされ、ローカルリスナーへの同期配信とSparkListenerBusへの非同期ポストの両方が行われる。activeQueryRunIdsにrunIdが追加され、以降のイベントがこのクエリに関連づけられる。

## 通知テンプレート

### メール通知の場合

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

### 本文テンプレート

```json
{
  "id": "{id}",
  "runId": "{runId}",
  "name": "{name}",
  "timestamp": "{timestamp}",
  "jobTags": ["{tag1}", "{tag2}"]
}
```

### 添付ファイル

該当なし。

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| id | クエリの永続一意ID（再起動しても不変） | StreamExecution.id | Yes |
| runId | 実行ごとの一意ID（再起動で変更） | StreamExecution.runId | Yes |
| name | ユーザー指定のクエリ名 | StreamExecution.name | No（null可） |
| timestamp | クエリ開始タイムスタンプ（ISO8601形式） | triggerClock.getTimeMillis() | Yes |
| jobTags | ジョブタグのセット | SparkContext.getJobTags() | No（空集合可） |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| API呼び出し | DataStreamWriter.start() | クエリスレッドの起動・アクティベーション処理中 | StreamExecution.runStream内のpostEventで発火 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| クエリの初期化失敗 | StreamExecutionの初期化中に例外が発生した場合、QueryStartedEventは発火されない可能性がある |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[DataStreamWriter.start 呼び出し] --> B[StreamExecution生成]
    B --> C[クエリスレッド起動]
    C --> D[runStream実行]
    D --> E[QueryStartedEvent生成]
    E --> F[StreamingQueryListenerBus.post]
    F --> G[activeQueryRunIdsにrunId追加]
    G --> H[SparkListenerBusにpost]
    H --> I[ローカルリスナーにpostToAll 同期]
    I --> J[start メソッド戻り]
```

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

### 参照テーブル一覧

該当なし。本イベントはメモリ上のストリーミングクエリ情報から生成される。

### 更新テーブル一覧

該当なし。

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

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

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| リスナー例外 | リスナーのonQueryStartedメソッドで例外発生 | ListenerBusが例外をキャッチしてログ出力。クエリ開始は続行 |
| クエリ開始ブロック | リスナーの処理が長時間かかる | start()メソッドの返却がブロックされる。リスナー側で長時間処理を避ける必要がある |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし。クエリ開始時に即座に配信される。

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

- クエリ名にはユーザー指定の任意文字列が含まれる可能性がある
- イベントはSparkプロセス内部およびイベントログに記録される
- JSON形式でのシリアライズ機能を持ち、Spark Connect経由でリモートクライアントにも配信可能

## 備考

- QueryStartedEventはStreamingQueryListener.Event traitを継承し、SparkListenerEventも継承している
- Since 2.1.0で導入。jobTagsフィールドはより新しいバージョンで追加
- 同期配信されるのはQueryStartedEventのみ。他のイベント（Progress, Idle, Terminated）は非同期配信
- fromJsonメソッドを持ち、JSON文字列からの復元が可能（Spark History Server等で使用）

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StreamingQueryListener.scala | `sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala` | QueryStartedEventクラス（164-186行目）のフィールド定義（id, runId, name, timestamp, jobTags）とjsonメソッドの実装を確認 |
| 1-2 | StreamingQueryListener.scala | `sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala` | Event trait（147行目）とStreamingQueryListenerの抽象メソッド（56行目: onQueryStarted）を確認 |

**読解のコツ**: QueryStartedEventはcompanion object内のclass（inner class）である。private[sql]コンストラクタとpublicコンストラクタの両方を持ち、後方互換のためjobTagsなしのコンストラクタ（173-175行目）もある。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | StreamExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecution.scala` | runStream内でのQueryStartedEvent生成とpostEvent呼び出し（298-302行目付近）。同期配信の保証メカニズム（277-280行目のコメント） |

**主要処理フロー**:
1. **298行目**: `val startTimestamp = triggerClock.getTimeMillis()` - タイムスタンプ取得
2. **299-302行目**: `postEvent(new QueryStartedEvent(id, runId, name, ...))` - イベント生成・ポスト

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StreamingQueryListenerBus.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryListenerBus.scala` | postメソッド（71-81行目）でQueryStartedEventの特別扱い：activeQueryRunIdsへの追加、SparkListenerBusへのpost、ローカルpostToAll |
| 3-2 | StreamingQueryListenerBus.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryListenerBus.scala` | doPostEventメソッド（118-147行目）でのイベント種別に応じたリスナーメソッド呼び出し |

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

```
DataStreamWriter.start()
    |
    +-- StreamExecution(生成)
            |
            +-- runStream()
                    |
                    +-- postEvent(new QueryStartedEvent(...))
                            |
                            +-- StreamingQueryListenerBus.post(event)
                                    |
                                    +-- activeQueryRunIds += runId
                                    |
                                    +-- sparkListenerBus.post(event)  [非同期]
                                    |
                                    +-- postToAll(event)  [同期]
                                            |
                                            +-- doPostEvent(listener, event)
                                                    |
                                                    +-- listener.onQueryStarted(event)
```

### データフロー図

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

StreamExecution              StreamingQueryListenerBus          StreamingQueryListener
  id, runId, name     --->     post(QueryStartedEvent)    --->    .onQueryStarted()
  timestamp, jobTags              |                                    |
                                  v                                    v
                              SparkListenerBus              StreamingQueryStatusListener
                              (非同期配信)                      (UI更新)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StreamingQueryListener.scala | `sql/api/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryListener.scala` | ソース | QueryStartedEventクラス定義（164-186行目） |
| StreamExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamExecution.scala` | ソース | イベント発火元（298-302行目） |
| StreamingQueryListenerBus.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/StreamingQueryListenerBus.scala` | ソース | イベント配信バス（71-81行目） |
| StreamingQueryStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala` | ソース | UI用リスナー実装 |
| StreamingQueryListenerSuite.scala | `sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryListenerSuite.scala` | テスト | リスナーイベントのテスト |
