# 機能設計書 40-DStreamストリーム処理

## 概要

本ドキュメントは、Apache SparkのDStream（Discretized Streams）APIによる従来型ストリーム処理機能に関する機能設計書である。RDDベースのマイクロバッチ方式でリアルタイムデータ処理を実行するDStream APIの仕組みを詳細に記述する。

### 本機能の処理概要

**業務上の目的・背景**：DStreamはSpark初期から提供されてきたストリーム処理APIであり、RDDの抽象化の上にストリーミング処理を構築する。マイクロバッチ方式により、各バッチ間隔でデータを収集し、RDD操作として処理を実行する。Spark 3.4.0以降は非推奨（deprecated）となり、Structured Streamingへの移行が推奨されているが、レガシーシステムとの互換性維持のため引き続き提供されている。

**機能の利用シーン**：既存のDStreamベースのストリーミングアプリケーションの保守・運用、Structured Streamingへの移行前の過渡期における利用。

**主要な処理内容**：
1. StreamingContext：DStreamストリーム処理のエントリーポイント、バッチ間隔の設定と処理の開始・停止
2. DStream：離散化ストリームの抽象化、各バッチ間隔でRDDを生成
3. DStreamGraph：入力DStreamと出力DStreamの有向グラフ管理
4. JobScheduler：バッチごとのジョブ生成とスケジューリング
5. ReceiverTracker：データ受信レシーバーの管理（レシーバー管理機能と連携）
6. チェックポイント：DStreamの状態と設定の永続化

**関連システム・外部連携**：SparkContext、各種入力ソース（TCP、Kafka等）、Web UI（DStreams Streaming Overview画面）

**権限による制御**：SparkContextの権限に従う。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 18 | Streaming Overview（DStreamsストリーミング概要） | 主機能 | StreamingJobProgressListenerから各バッチの処理時間・スケジューリング遅延・入力レートを取得して表示 |
| 19 | Batch Detail（バッチ詳細） | 主機能 | バッチ内のOutput Operation・Spark Job情報を取得して表示 |

## 機能種別

計算処理 / リアルタイム処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| sparkContext | SparkContext | Yes | 基盤となるSparkContext | アクティブなContext |
| batchDuration | Duration | Yes | バッチ間隔（ミリ秒） | 正の値 |
| checkpoint | String | No | チェックポイントディレクトリ | 有効なパス |

### 入力データソース

- TCP/UDPソケット
- ファイルシステム（HDFS、ローカル等）
- カスタムレシーバー
- Kafkaコネクタ（connector/kafka-0-10/）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| DStream[T] | DStream[T] | 離散化ストリーム（各バッチでRDD[T]を生成） |
| RDD[T] | RDD[T] | バッチごとのRDDデータ |

### 出力先

- foreachRDD操作による任意の出力先
- saveAsTextFiles等のファイル出力
- DStreamのprint操作によるコンソール出力

## 処理フロー

### 処理シーケンス

```
1. StreamingContext作成
   └─ StreamingContext(sparkContext, batchDuration) でコンテキスト生成
2. 入力DStream定義
   └─ socketTextStream/textFileStream/receiverStream等で入力ソースを定義
3. 変換処理定義
   └─ map/flatMap/filter/window/reduceByKeyAndWindow等の変換を定義
4. 出力操作定義
   └─ foreachRDD/print/saveAsTextFiles等で出力操作を定義
5. ストリーム処理開始
   └─ ssc.start() でJobSchedulerを起動
6. バッチ実行ループ
   └─ バッチ間隔ごとにDStreamGraph.generateJobs()でジョブ生成→実行
7. 終了待機
   └─ ssc.awaitTermination() で終了を待機
```

### フローチャート

```mermaid
flowchart TD
    A[StreamingContext生成] --> B[入力DStream定義]
    B --> C[変換処理定義]
    C --> D[出力操作定義]
    D --> E[ssc.start]
    E --> F[JobScheduler起動]
    F --> G{バッチ間隔待機}
    G --> H[DStreamGraph.generateJobs]
    H --> I[RDD生成]
    I --> J[SparkJob実行]
    J --> K[進捗報告]
    K --> G
    G --> L{停止要求?}
    L -->|Yes| M[ssc.stop]
    L -->|No| G
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | マイクロバッチ方式 | データはバッチ間隔ごとに収集され、各バッチはRDDとして処理される | 全DStream処理 |
| BR-02 | DStream変換の遅延評価 | 変換は出力操作が定義されるまで実行されない（遅延評価） | DStream定義時 |
| BR-03 | 非推奨 | Spark 3.4.0以降deprecated、Structured Streamingへの移行推奨 | 全DStream API |
| BR-04 | StreamingContext唯一性 | 1つのJVMプロセス内でアクティブなStreamingContextは1つのみ | start()時 |

### 計算ロジック

- **バッチ生成**: バッチ間隔（batchDuration）ごとにDStreamGraph内の全入力DStreamがRDDを生成
- **ジョブスケジューリング**: JobSchedulerがバッチごとにジョブを生成し、SparkContextのジョブスケジューラに投入
- **ウィンドウ操作**: window(windowDuration, slideDuration)でスライディングウィンドウ内のRDDを統合

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | DStream自体はデータベース操作なし（foreachRDD内でユーザー定義の操作可能） |

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

DStream APIはデータベース操作を直接提供しない。ユーザーがforeachRDD内でデータベースアクセスを実装する。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| STREAMING_CONTEXT_ACTIVE | SparkException | 同一JVM内で複数のStreamingContextが活動 | 既存のContextを停止 |
| RECEIVER_ERROR | SparkException | レシーバーの障害 | レシーバー設定を確認 |
| BATCH_TIMEOUT | StreamingException | バッチ処理時間がバッチ間隔を超過 | バッチ間隔の延長またはリソース追加 |

### リトライ仕様

レシーバーはReceiverTrackerにより自動的に再起動される。タスク失敗時はSparkの標準リトライメカニズムに従う。

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

DStreamのチェックポイント機能により、バッチ処理の進行状況を永続化し、障害復旧時に再開可能。ただし、Structured StreamingのようなExactly-Once保証は提供されない（At-Least-Once）。

## パフォーマンス要件

- バッチ処理時間がバッチ間隔以内であることが安定稼働の前提
- スケジューリング遅延の監視が重要（Web UIの Scheduling Delay メトリクス）
- ウィンドウ操作は大量のRDDを保持するためメモリに注意

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

- SparkContextのセキュリティ設定に従う
- チェックポイントディレクトリのアクセス権限管理が必要

## 備考

- DStream APIは `streaming/` ディレクトリ配下に実装
- Spark 3.4.0以降、`@deprecated`アノテーションが付与されている
- StreamingContextクラスの`@deprecated`アノテーションのメッセージ: "DStream is deprecated. Migrate to Structured Streaming."

---

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

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

### 推奨読解順序

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

DStreamの基本構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | DStream.scala | `streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala` | DStreamの基底クラス、generatedRDDs管理、compute()メソッド |
| 1-2 | DStreamGraph.scala | `streaming/src/main/scala/org/apache/spark/streaming/DStreamGraph.scala` | 入力/出力DStreamのグラフ管理、generateJobs() |

**読解のコツ**: DStreamは各バッチ間隔でRDDを生成する「RDDの連続列」として理解するとよい。generatedRDDs（HashMap[Time, RDD]）がバッチ時刻とRDDの対応を保持する。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | StreamingContext.scala | `streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala` | メインエントリーポイント、start()/stop()/awaitTermination() |

**主要処理フロー**:
- **68-73行目**: `@deprecated`アノテーションとクラス定義
- **69行目**: `StreamingContext` が `SparkContext`, `Checkpoint`, `Duration` を受け取る
- **80行目**: `this(sparkContext, batchDuration)` コンストラクタ

#### Step 3: ジョブスケジューリングを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | JobScheduler.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala` | バッチごとのジョブ生成・スケジューリング |
| 3-2 | StreamingJobProgressListener.scala | `streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala` | Web UI向け進捗リスナー |

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

```
StreamingContext
    │
    ├─ start()
    │      └─ JobScheduler.start()
    │             ├─ ReceiverTracker.start()
    │             └─ JobGenerator.start()
    │                    └─ Timer(batchDuration)
    │
    ├─ [バッチ間隔ごと]
    │      └─ DStreamGraph.generateJobs(time)
    │             ├─ DStream.getOrCompute(time)
    │             │      └─ DStream.compute(time) → RDD[T]
    │             └─ OutputDStream.generateJob(time)
    │                    └─ SparkContext.runJob()
    │
    └─ stop()
           └─ JobScheduler.stop()
```

### データフロー図

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

外部ソース               ReceiverTracker               DStream[T]
(TCP/Kafka/File)  ──▶  → DStream.compute(time)    ──▶  → RDD[T]
                              │
                              ▼
RDD[T]                  DStream変換                    出力DStream
(各バッチ)         ──▶  map/filter/window          ──▶  → foreachRDD
                              │
                              ▼
                        SparkJob実行                   外部ストレージ
                        SparkContext.runJob()     ──▶  (HDFS/DB/Console)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StreamingContext.scala | `streaming/src/main/scala/org/apache/spark/streaming/` | ソース | エントリーポイント |
| DStream.scala | `streaming/src/main/scala/org/apache/spark/streaming/dstream/` | ソース | DStream基底クラス |
| DStreamGraph.scala | `streaming/src/main/scala/org/apache/spark/streaming/` | ソース | DStreamグラフ管理 |
| JobScheduler.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/` | ソース | ジョブスケジューラ |
| StreamingJobProgressListener.scala | `streaming/src/main/scala/org/apache/spark/streaming/ui/` | ソース | Web UI進捗リスナー |
| Checkpoint.scala | `streaming/src/main/scala/org/apache/spark/streaming/` | ソース | チェックポイント管理 |
| ReceiverTracker.scala | `streaming/src/main/scala/org/apache/spark/streaming/scheduler/` | ソース | レシーバー管理 |
