# 機能設計書 41-レシーバー管理

## 概要

本ドキュメントは、Apache Spark DStreams Streamingにおけるレシーバー管理機能の設計を記述する。外部データソースからデータを受信するレシーバーの起動・監視・障害復旧を管理するサブシステムについて、その構成・処理フロー・データフローを詳細に説明する。

### 本機能の処理概要

レシーバー管理機能は、DStreams（Discretized Streams）ストリーム処理において、外部データソースからデータを取り込むためのレシーバーコンポーネントのライフサイクルを管理する機能である。

**業務上の目的・背景**：リアルタイムデータ処理アプリケーションでは、Kafka、ソケット、HDFS等の外部データソースからデータを継続的に受信する必要がある。レシーバー管理機能は、このデータ受信プロセスを安定的に維持し、障害時にも自動復旧を行うことで、ストリーミングアプリケーションの信頼性を確保する。

**機能の利用シーン**：DStreamベースのストリーミングアプリケーションにおいて、外部データソースからのリアルタイムデータ取り込みが必要な場面で利用される。具体的には、`StreamingContext.receiverStream()`や各種InputDStream（SocketInputDStream、KafkaInputDStream等）を介してレシーバーが起動される。

**主要な処理内容**：
1. レシーバーの起動とReceiverTrackerへの登録
2. 受信データのブロック化とBlockManagerへの格納
3. レシーバーの状態監視（Initialized / Started / Stopped）
4. 障害発生時の自動リスタート（設定可能な遅延時間付き）
5. レートリミッティングによるバックプレッシャー制御
6. Write Ahead Log（WAL）によるデータ永続化（オプション）
7. 古いブロックのクリーンアップ

**関連システム・外部連携**：BlockManager（データ格納）、ReceiverTracker（ドライバー側管理）、WriteAheadLog（障害復旧）、RPC通信フレームワーク（ドライバーとの通信）

**権限による制御**：特段のロール制御はないが、WAL利用時にはチェックポイントディレクトリへのHDFS書き込み権限が必要。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 18 | Streaming Overview（DStreamsストリーミング概要） | 補助機能 | レシーバーの稼働状況・受信レート情報をテーブル表示 |

## 機能種別

データ連携 / ストリームデータ受信管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| storageLevel | StorageLevel | Yes | 受信データの保存レベル | 有効なStorageLevelであること |
| spark.streaming.receiverRestartDelay | Int | No | レシーバー再起動時の遅延（ms）、デフォルト2000 | 正の整数 |
| spark.streaming.receiver.maxRate | Long | No | レシーバーの最大受信レート（レコード/秒） | 0以上の整数 |
| spark.streaming.backpressure.initialRate | Long | No | バックプレッシャーの初期レート | 正の整数 |
| spark.streaming.receiver.blockStoreTimeout | Int | No | WAL利用時のブロック保存タイムアウト（秒）、デフォルト30 | 正の整数 |
| spark.streaming.blockInterval | Long | No | ブロック生成間隔（ms） | 正の整数 |
| spark.streaming.blockQueueSize | Int | No | プッシュ待ちブロックキューサイズ、デフォルト10 | 正の整数 |

### 入力データソース

外部データソース（ソケット、Kafka、Kinesis等）からのストリームデータ。具体的なデータソースはReceiverの実装クラスに依存する。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| StreamBlockId | StreamBlockId | 受信データブロックの一意識別子 |
| ReceivedBlockInfo | ReceivedBlockInfo | ブロックのメタデータ（streamId、レコード数、格納結果） |
| ReceivedBlockStoreResult | ReceivedBlockStoreResult | ブロック格納結果（blockId、レコード数、WALハンドル） |

### 出力先

- BlockManager（メモリ/ディスク）へのデータブロック格納
- Write Ahead Log（オプション、HDFS上のチェックポイントディレクトリ）
- ReceiverTracker（ドライバー）へのブロック情報報告

## 処理フロー

### 処理シーケンス

```
1. ReceiverSupervisorImpl初期化
   └─ ReceivedBlockHandler選択（WAL有効ならWriteAheadLogBasedBlockHandler、無効ならBlockManagerBasedBlockHandler）
   └─ ReceiverTracker用RPCエンドポイント設定
   └─ デフォルトBlockGenerator生成

2. レシーバー起動（start）
   └─ BlockGenerator起動（blockIntervalTimerとblockPushingThread開始）
   └─ ReceiverTrackerへのレシーバー登録（RegisterReceiver RPC）
   └─ Receiver.onStart()呼び出し

3. データ受信ループ
   └─ Receiver.store()でデータ投入
   └─ BlockGenerator.addData()でバッファに蓄積
   └─ RateLimiter.waitToPush()でレート制御
   └─ blockIntervalTimer定期実行でバッファをブロック化
   └─ blockPushingThreadでブロックをBlockManager/WALに格納
   └─ ReceiverTrackerへブロック情報報告（AddBlock RPC）

4. レシーバー停止（stop）
   └─ Receiver.onStop()呼び出し
   └─ ReceiverTrackerへの登録解除（DeregisterReceiver RPC）
   └─ BlockGenerator停止（データ追加停止→ブロック生成停止→キュードレイン）
   └─ WALクローズ

5. 障害復旧（restart）
   └─ 非同期スレッドでstopReceiver→遅延→startReceiver
```

### フローチャート

```mermaid
flowchart TD
    A[ReceiverSupervisorImpl初期化] --> B[BlockGenerator起動]
    B --> C[ReceiverTracker登録]
    C --> D{登録成功?}
    D -->|Yes| E[Receiver.onStart]
    D -->|No| F[停止]
    E --> G[データ受信ループ]
    G --> H{store呼び出し}
    H --> I[addData: バッファ蓄積]
    I --> J{blockInterval経過?}
    J -->|Yes| K[バッファをブロック化]
    J -->|No| I
    K --> L[ブロックをBlockManager格納]
    L --> M[ReceiverTrackerへ報告]
    M --> G
    G --> N{障害発生?}
    N -->|Yes| O{restart or stop?}
    O -->|restart| P[遅延後に再起動]
    O -->|stop| F
    P --> C
    N -->|No| G
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-41-01 | レート制限 | spark.streaming.receiver.maxRateを上限としてGuava RateLimiterで受信レートを制御する | maxRateが設定されている場合 |
| BR-41-02 | WALストレージレベル調整 | WAL有効時はデシリアライズ=false、レプリケーション=1に強制変更 | WALが有効な場合 |
| BR-41-03 | 再起動遅延 | レシーバー再起動はspark.streaming.receiverRestartDelay（デフォルト2000ms）の遅延後に実行 | restart呼び出し時 |
| BR-41-04 | ブロック生成間隔 | spark.streaming.blockInterval（デフォルト200ms）ごとにバッファをブロック化 | BlockGenerator動作中 |
| BR-41-05 | ブロックキューサイズ | blocksForPushingキューがspark.streaming.blockQueueSize（デフォルト10）に達するとブロック化スレッドがブロッキング | キュー満杯時 |

### 計算ロジック

レートリミッターの初期レートは `min(spark.streaming.backpressure.initialRate, spark.streaming.receiver.maxRate)` で決定される。動的レート更新時は `min(newRate, maxRateLimit)` が適用される。

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

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

本機能はRDBMSへの直接操作は行わない。ストレージ操作はBlockManagerおよびHDFS（WAL）を介して行われる。

| 操作 | 対象ストレージ | 操作種別 | 概要 |
|-----|-------------|---------|------|
| データブロック格納 | BlockManager | PUT | 受信データブロックをメモリ/ディスクに格納 |
| WAL書き込み | HDFS | WRITE | ブロックデータをWrite Ahead Logに書き込み |
| 古いブロック削除 | WAL / BlockManager | DELETE | 閾値時刻より古いブロックをクリーンアップ |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| SparkException | 致命的エラー | WAL有効時にチェックポイントディレクトリ未設定 | streamingContext.checkpoint()を設定 |
| SparkException | 致命的エラー | BlockManagerへのブロック格納失敗 | ストレージレベル・メモリ設定を確認 |
| SparkException | 致命的エラー | ReceiverTrackerへのブロック追加失敗 | ドライバーとの接続状態を確認 |
| SparkException | 致命的エラー | BlockGeneratorが未起動/停止済み状態でデータ追加 | ライフサイクル管理を確認 |
| NonFatal | 復旧可能エラー | Receiver.onStart()で例外発生 | レシーバー停止後にエラー報告 |

### リトライ仕様

レシーバーの`restart()`メソッドにより、設定可能な遅延時間後に自動リスタートが実行される。リスタートは非同期スレッドで行われ、`stopReceiver` -> `Thread.sleep(delay)` -> `startReceiver` の順序で処理される。

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

WAL有効時は、BlockManagerへの格納とWALへの書き込みが並列実行され、両方が完了した場合にのみ成功と判定される。タイムアウト（デフォルト30秒）を超えた場合は例外がスローされる。

## パフォーマンス要件

- ブロック生成間隔（blockInterval）はストリーミングバッチ間隔より小さく設定する必要がある
- RateLimiterによりレシーバーのスループットを制御可能
- WAL有効時はBlockManagerとWAL書き込みを並列実行（2スレッド）してレイテンシを最小化

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

- WAL利用時はHDFS上のチェックポイントディレクトリへの書き込み権限が必要
- ReceiverTrackerとの通信はSpark RPCフレームワークを介して行われ、Sparkのセキュリティ設定（認証・暗号化）に従う

## 備考

- DStreams APIはStructured Streamingに対する旧APIであり、新規開発ではStructured Streamingの使用が推奨される
- カスタムレシーバーの実装にはReceiver抽象クラスを継承し、onStart()とonStop()を実装する

---

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

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

### 推奨読解順序

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

まず、レシーバー管理で使用されるデータ構造とメッセージプロトコルを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ReceivedBlock.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlock.scala` | 受信ブロックの3つの型（ArrayBufferBlock, IteratorBlock, ByteBufferBlock）を理解する |
| 1-2 | ReceiverMessage.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverMessage.scala` | ドライバーからレシーバーへのメッセージ型（StopReceiver, CleanupOldBlocks, UpdateRateLimit）を理解する |

**読解のコツ**: `ReceivedBlock`はsealed traitでパターンマッチにより3種類のブロック形式を安全に処理する。`ReceiverMessage`もsealed traitでSerializableを継承しRPC通信で使用される。

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

レシーバーの基底クラスとユーザー向けAPIを把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | Receiver.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/Receiver.scala` | 抽象クラスの全体構造、store()メソッド群、restart()/stop()メソッドを理解する |

**主要処理フロー**:
1. **86行目**: `Receiver[T]`抽象クラス定義。`storageLevel`パラメータで保存レベルを指定
2. **102行目**: `onStart()` - ユーザーが実装するデータ受信開始メソッド（非ブロッキング必須）
3. **108行目**: `onStop()` - ユーザーが実装するリソース解放メソッド
4. **118-180行目**: `store()`メソッド群 - データをSupervisorに委譲して格納
5. **195-219行目**: `restart()`メソッド群 - 非同期でレシーバーを再起動
6. **260-279行目**: private変数群 - `_supervisor`の遅延バインディング

#### Step 3: レシーバースーパーバイザーの抽象層を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ReceiverSupervisor.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisor.scala` | レシーバーのライフサイクル管理（start/stop/restart）の状態遷移を理解する |

**主要処理フロー**:
- **43-46行目**: ReceiverState列挙型（Initialized, Started, Stopped）
- **62行目**: `defaultRestartDelay` - 再起動遅延のデフォルト値（2000ms）
- **130-133行目**: `start()` - onStart()呼び出し後にstartReceiver()
- **136-142行目**: `stop()` - stopReceiver→onStop→シャットダウン
- **145-160行目**: `startReceiver()` - ReceiverTrackerへの登録とReceiver.onStart()呼び出し
- **191-204行目**: `restartReceiver()` - Future内で非同期にstop→sleep→start

#### Step 4: レシーバースーパーバイザーの具体実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | ReceiverSupervisorImpl.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisorImpl.scala` | ReceivedBlockHandlerの選択、RPCエンドポイント、pushAndReportBlockの処理フローを理解する |

**主要処理フロー**:
- **55-68行目**: `receivedBlockHandler`の初期化 - WAL有効時とBlockManager直接格納時の分岐
- **72行目**: `trackerEndpoint` - ReceiverTrackerへのRPC参照取得
- **75-92行目**: RPCエンドポイント設定 - StopReceiver/CleanupOldBlocks/UpdateRateLimitメッセージ処理
- **100-113行目**: `defaultBlockGeneratorListener` - ブロック生成時のコールバック定義
- **119-121行目**: `pushSingle()` - 単一データをBlockGeneratorに投入
- **151-166行目**: `pushAndReportBlock()` - ブロック格納とReceiverTrackerへの報告
- **190-194行目**: `onReceiverStart()` - RegisterReceiver RPCでドライバーに登録

#### Step 5: ブロック生成メカニズムを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | BlockGenerator.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/BlockGenerator.scala` | 5つの状態遷移（Initialized→Active→StoppedAddingData→StoppedGeneratingBlocks→StoppedAll）と2つのスレッド（タイマー・プッシャー）を理解する |

**主要処理フロー**:
- **86行目**: `Block`ケースクラス - StreamBlockIdとArrayBufferの組
- **99-103行目**: `GeneratorState`列挙型 - 5段階の状態遷移
- **105行目**: `blockIntervalMs` - ブロック生成間隔設定
- **119-129行目**: `start()` - タイマーとプッシュスレッド起動
- **165-180行目**: `addData()` - レート制限後にバッファにデータ追加
- **237-259行目**: `updateCurrentBuffer()` - 定期的にバッファをブロック化してキューに投入
- **262-294行目**: `keepPushingBlocks()` - キューからブロックを取り出してリスナー経由で格納

#### Step 6: レート制限とブロックハンドラを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 6-1 | RateLimiter.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/RateLimiter.scala` | Guava RateLimiterラッパーによるスループット制御を理解する |
| 6-2 | ReceivedBlockHandler.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlockHandler.scala` | BlockManagerBasedとWriteAheadLogBasedの2つの格納戦略を理解する |

**主要処理フロー（RateLimiter）**:
- **41行目**: `maxRateLimit` - spark.streaming.receiver.maxRate設定値
- **44-46行目**: `waitToPush()` - Guava RateLimiter.acquire()でブロッキング
- **59-66行目**: `updateRate()` - 動的レート更新（maxRateLimitを上限とする）

**主要処理フロー（ReceivedBlockHandler）**:
- **69-106行目**: `BlockManagerBasedBlockHandler` - BlockManagerに直接格納
- **125-225行目**: `WriteAheadLogBasedBlockHandler` - BlockManagerとWALに並列書き込み
- **174-215行目**: `storeBlock()` - シリアライズ→Future並列実行→zip結合→結果返却

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

```
Receiver.store(data)
    |
    +-- ReceiverSupervisorImpl.pushSingle(data)
    |       |
    |       +-- BlockGenerator.addData(data)
    |               |
    |               +-- RateLimiter.waitToPush()
    |               +-- currentBuffer += data
    |
    +-- [Timer] BlockGenerator.updateCurrentBuffer(time)
    |       |
    |       +-- BlockGeneratorListener.onGenerateBlock(blockId)
    |       +-- blocksForPushing.put(block)
    |
    +-- [Thread] BlockGenerator.keepPushingBlocks()
            |
            +-- BlockGeneratorListener.onPushBlock(blockId, buffer)
                    |
                    +-- ReceiverSupervisorImpl.pushAndReportBlock()
                            |
                            +-- ReceivedBlockHandler.storeBlock(blockId, block)
                            |       |
                            |       +-- BlockManagerBasedBlockHandler.storeBlock()
                            |       |       +-- BlockManager.putIterator/putBytes()
                            |       |
                            |       +-- WriteAheadLogBasedBlockHandler.storeBlock()
                            |               +-- [Future] BlockManager.putBytes()
                            |               +-- [Future] WriteAheadLog.write()
                            |
                            +-- trackerEndpoint.askSync(AddBlock(blockInfo))
```

### データフロー図

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

外部データソース ───> Receiver.store()
                         |
                         v
                    BlockGenerator
                    (バッファリング)
                         |
                    [blockInterval]
                         |
                         v
                    ブロック化
                         |
                    +---------+----------+
                    |                    |
                    v                    v
              BlockManager          WAL(HDFS)
              (メモリ/ディスク)     (オプション)
                    |
                    v
              ReceiverTracker ───> ブロック情報
              (ドライバー)          (ReceivedBlockInfo)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Receiver.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/Receiver.scala` | ソース | レシーバー基底抽象クラス |
| ReceiverSupervisor.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisor.scala` | ソース | スーパーバイザー抽象クラス（ライフサイクル管理） |
| ReceiverSupervisorImpl.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverSupervisorImpl.scala` | ソース | スーパーバイザー具体実装（RPC・ブロック管理） |
| BlockGenerator.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/BlockGenerator.scala` | ソース | データブロック生成器（バッファリング・定期ブロック化） |
| ReceivedBlockHandler.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlockHandler.scala` | ソース | ブロック格納ハンドラ（BlockManager/WAL） |
| ReceivedBlock.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlock.scala` | ソース | 受信ブロックデータ型定義 |
| ReceiverMessage.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceiverMessage.scala` | ソース | レシーバー制御メッセージ型定義 |
| RateLimiter.scala | `streaming/src/main/scala/org/apache/spark/streaming/receiver/RateLimiter.scala` | ソース | レート制限機構 |
