# 機能設計書 36-ストリーム処理エンジン

## 概要

本ドキュメントは、Apache SparkのStructured Streamingにおけるストリーム処理エンジンに関する機能設計書である。マイクロバッチおよび連続処理モードによるストリーミングクエリの実行管理、進行状況の追跡、障害復旧を統括するコアエンジンを詳細に記述する。

### 本機能の処理概要

**業務上の目的・背景**：リアルタイムデータ処理の需要増大に対し、バッチ処理と同一のAPIでストリーム処理を記述できるStructured Streamingエンジンを提供する。開発者はDataFrame/Dataset APIやSQLを使用してストリーミングクエリを定義でき、Exactly-once処理保証やイベント時間処理、ウォーターマーク管理といった複雑なストリーム処理要件をエンジンが透過的に管理する。

**機能の利用シーン**：リアルタイムETLパイプライン、ストリーミングデータの集約・ウィンドウ処理、イベント駆動アーキテクチャにおけるデータ処理、リアルタイムダッシュボード用データフィード。

**主要な処理内容**：
1. StreamExecution：ストリーミングクエリの実行ライフサイクル管理（開始・停止・再開）
2. MicroBatchExecution：マイクロバッチモードでのクエリ実行（オフセット管理、バッチ計画、実行）
3. AsyncProgressTrackingMicroBatchExecution：非同期進行状況追跡によるマイクロバッチ実行
4. IncrementalExecution：増分実行計画の最適化（ステートフルオペレーターのバッチ情報注入）
5. ProgressReporter：処理進捗の追跡とメトリクス報告
6. WatermarkTracker/WatermarkPropagator：イベント時間ウォーターマークの追跡と伝播
7. TriggerExecutor：トリガー間隔に基づく実行スケジューリング

**関連システム・外部連携**：データソースV2 API（MicroBatchStream/ContinuousStream）、StateStore、CheckpointManager、Kafkaソース/シンク

**権限による制御**：チェックポイントディレクトリへのアクセス権限が必要。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 16 | Streaming Query List（ストリーミングクエリ一覧） | 主機能 | StreamingQueryStoreからクエリのステータス・入力レート・処理レート情報を取得して表示 |
| 17 | Streaming Query Statistics（ストリーミングクエリ統計） | 主機能 | マイクロバッチの実行統計を取得してD3.jsグラフ表示 |

## 機能種別

計算処理 / データ連携 / リアルタイム処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| plan | LogicalPlan | Yes | ストリーミングクエリの論理実行計画 | ストリーミングソースを含む |
| trigger | Trigger | Yes | 実行トリガー（ProcessingTime/Once/AvailableNow/Continuous） | 有効なTrigger |
| outputMode | OutputMode | Yes | 出力モード（Append/Complete/Update） | ソースとシンクに適合 |
| checkpointLocation | String | Yes | チェックポイントディレクトリパス | 書き込み可能パス |
| sink | StreamWriteSupport | Yes | 出力先（ストリーミングシンク） | 有効なシンク |

### 入力データソース

- ストリーミングデータソース（Kafka、FileStream、Rate等のMicroBatchStream/ContinuousStream実装）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| StreamingQuery | StreamingQuery | 実行中のストリーミングクエリハンドル |
| StreamingQueryProgress | StreamingQueryProgress | バッチごとの処理進捗レポート |
| メトリクス | 各種メトリクス | 入力レート、処理レート、バッチ実行時間等 |

### 出力先

- ストリーミングシンク（Kafka、ファイル、メモリ、コンソール等）
- Web UI（クエリ統計）
- StreamingQueryListener（イベント通知）

## 処理フロー

### 処理シーケンス

```
1. クエリ開始
   └─ StreamingQueryManager.startQuery() でStreamExecutionを生成・起動
2. マイクロバッチループ
   └─ MicroBatchExecution.runActivatedStream() でトリガー間隔でバッチ実行
3. オフセット取得
   └─ 各ソースからlatestOffset()で最新オフセットを取得
4. バッチ計画構築
   └─ IncrementalExecution で増分実行計画を最適化
5. バッチ実行
   └─ SparkSessionのrunJob()でバッチのジョブを実行
6. オフセットコミット
   └─ OffsetSeqLogにコミット済みオフセットを記録
7. 進捗報告
   └─ ProgressReporterが処理統計を集計・報告
8. ウォーターマーク更新
   └─ WatermarkTracker/WatermarkPropagatorでイベント時間ウォーターマークを更新
```

### フローチャート

```mermaid
flowchart TD
    A[writeStream.start] --> B[StreamingQueryManager]
    B --> C[MicroBatchExecution生成]
    C --> D[runActivatedStream]
    D --> E{トリガー待機}
    E --> F[latestOffset取得]
    F --> G{新データあり?}
    G -->|Yes| H[バッチ計画構築]
    G -->|No| E
    H --> I[IncrementalExecution]
    I --> J[SparkJob実行]
    J --> K[オフセットコミット]
    K --> L[進捗報告]
    L --> M[ウォーターマーク更新]
    M --> E
    D --> N{停止要求?}
    N -->|Yes| O[クエリ停止]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | Exactly-once保証 | チェックポイントとオフセットコミットにより、各レコードは厳密に1回処理される | マイクロバッチモード |
| BR-02 | 出力モード制約 | Appendモードでは集約なしクエリ、Completeモードでは集約必須 | クエリ解析時 |
| BR-03 | ウォーターマーク遅延 | withWatermark()で指定された遅延時間を超えたイベントは破棄される | ウィンドウ集約時 |
| BR-04 | トリガー制御 | ProcessingTimeトリガーは指定間隔でバッチ実行、Onceは1回のみ | クエリ実行時 |

### 計算ロジック

- **バッチID**: 0から始まる連番、各バッチ完了時にインクリメント
- **ウォーターマーク計算**: max(eventTime) - delayThreshold、全パーティション間の最小値
- **進捗レポート**: inputRowsPerSecond, processedRowsPerSecond, batchDuration等を算出

## データベース操作仕様

### 操作別データベース影響一覧

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| チェックポイント | OffsetSeqLog | WRITE | バッチオフセット情報の永続化 |
| チェックポイント | CommitLog | WRITE | コミット完了情報の永続化 |
| ステート | StateStore | READ/WRITE | ステートフルオペレーターの状態読み書き |

### テーブル別操作詳細

チェックポイントディレクトリ内のOffsetSeqLog/CommitLogに対するファイルI/O操作。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| STREAM_FAILED | StreamingQueryException | ストリーミングクエリの実行中エラー | エラー原因を修正して再起動 |
| CHECKPOINT_NOT_FOUND | AnalysisException | チェックポイントディレクトリが存在しない | パスを確認 |
| INCOMPATIBLE_SCHEMA | AnalysisException | チェックポイントとクエリのスキーマ不整合 | チェックポイントをクリアまたはスキーマ修正 |

### リトライ仕様

バッチ内のジョブはSparkの標準タスクリトライに従う。バッチ全体が失敗した場合、次のトリガーで同じバッチが再実行される。

## トランザクション仕様

各マイクロバッチはアトミックに処理される。オフセットコミットが完了するまではバッチは完了とみなされず、障害復旧時にはコミット済みオフセットから再開される。

## パフォーマンス要件

- マイクロバッチの実行遅延はトリガー間隔以内が理想
- ステートフルオペレーターのステートサイズが増大するとバッチ処理時間に影響
- ウォーターマーク設定によりステートの自動クリーンアップが行われる

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

- チェックポイントディレクトリのアクセス権限管理が必要
- ストリーミングソース/シンクの認証情報はSparkの設定管理に従う

## 備考

- Structured Streamingのストリーム処理エンジンは `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` 配下に実装
- MicroBatchExecutionが主要な実行エンジンであり、ContinuousExecutionは実験的機能
- AsyncProgressTrackingMicroBatchExecutionは非同期オフセット追跡による低遅延処理を提供

---

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

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

### 推奨読解順序

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

ストリーミングクエリの実行状態を表すデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StreamProgress.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` | ソース別オフセット進行状況管理 |
| 1-2 | StreamExecutionContext.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` | バッチ実行コンテキスト |
| 1-3 | StreamingRelation.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` | ストリーミングソースの論理計画表現 |

**読解のコツ**: StreamProgressはMap[SparkDataStream, Offset]でソースごとの処理済みオフセットを管理する。バッチIDと組み合わせてExactly-once処理を実現。

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

ストリーミングクエリの起動と実行ループを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | StreamExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` | ストリーミングクエリの基底クラス |
| 2-2 | MicroBatchExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` | マイクロバッチ実行エンジン |

#### Step 3: 増分実行と進捗管理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | IncrementalExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` | 増分実行計画の最適化 |
| 3-2 | ProgressReporter.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` | 処理進捗の追跡・報告 |
| 3-3 | WatermarkTracker.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/runtime/` | ウォーターマーク追跡 |

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

```
DataStreamWriter.start()
    │
    └─ StreamingQueryManager.startQuery()
           └─ MicroBatchExecution / ContinuousExecution
                  │
                  ├─ runActivatedStream()
                  │      ├─ TriggerExecutor.execute()
                  │      │      └─ runBatch()
                  │      │             ├─ latestOffset() [各ソース]
                  │      │             ├─ IncrementalExecution
                  │      │             ├─ runJob()
                  │      │             └─ commitOffsets()
                  │      │
                  │      └─ ProgressReporter.finishTrigger()
                  │
                  └─ WatermarkTracker.updateWatermark()
```

### データフロー図

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

ストリーミングソース      MicroBatchExecution              シンク
(Kafka/File/Rate)  ──▶  latestOffset → runBatch     ──▶  (Kafka/File/Console)
                              │
                              ▼
                     IncrementalExecution
                     (増分実行計画最適化)
                              │
                              ▼
                     StateStore (状態読み書き)
                     OffsetSeqLog (オフセット永続化)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StreamExecution.scala | `sql/core/.../streaming/runtime/` | ソース | ストリーミングクエリ基底クラス |
| MicroBatchExecution.scala | `sql/core/.../streaming/runtime/` | ソース | マイクロバッチ実行エンジン |
| AsyncProgressTrackingMicroBatchExecution.scala | `sql/core/.../streaming/runtime/` | ソース | 非同期進行状況追跡 |
| IncrementalExecution.scala | `sql/core/.../streaming/runtime/` | ソース | 増分実行計画最適化 |
| ProgressReporter.scala | `sql/core/.../streaming/runtime/` | ソース | 処理進捗報告 |
| WatermarkTracker.scala | `sql/core/.../streaming/runtime/` | ソース | ウォーターマーク追跡 |
| WatermarkPropagator.scala | `sql/core/.../streaming/runtime/` | ソース | ウォーターマーク伝播 |
| TriggerExecutor.scala | `sql/core/.../streaming/runtime/` | ソース | トリガー実行スケジューラ |
| StreamingQueryWrapper.scala | `sql/core/.../streaming/runtime/` | ソース | クエリラッパー |
| StreamMetadata.scala | `sql/core/.../streaming/runtime/` | ソース | ストリームメタデータ |
