# 機能設計書 9-ブロードキャスト変数

## 概要

本ドキュメントは、Apache Sparkのブロードキャスト変数機能の設計を記述する。TorrentBroadcast機構により、大きな読み取り専用データを効率的にクラスタ全体に配布する。

### 本機能の処理概要

TorrentBroadcastは、BitTorrentライクな分散配布メカニズムを用いて、ドライバーが保持する読み取り専用データをクラスタ全体のExecutorに効率的に配布するブロードキャスト変数の実装である。

**業務上の目的・背景**：分散処理において、全タスクで共通に参照するデータ（ルックアップテーブル、機械学習モデル、設定情報等）をExecutorに配布する需要がある。単純にタスクごとにデータを送信すると、ドライバーがボトルネックとなる。TorrentBroadcastはデータを小ブロックに分割し、Executor間でP2P的にブロックを交換することで、ドライバーの負荷を分散する。

**機能の利用シーン**：join操作でのブロードキャストハッシュジョイン、全タスクで参照する参照テーブルの配布、機械学習モデルのExecutorへの配布など。

**主要な処理内容**：
1. ドライバーでのデータのシリアライズとブロック分割（writeBlocks）
2. ブロックのBlockManager登録
3. Executor側でのブロック取得と再構成（readBlocks/readBroadcastBlock）
4. チェックサムによるブロックの整合性検証
5. 圧縮によるネットワーク転送量の削減
6. ブロードキャスト変数の破棄（destroy/unpersist）

**関連システム・外部連携**：BlockManager（ブロック保存・取得）、SparkContext（ブロードキャスト変数生成API）

**権限による制御**：ブロードキャスト変数は読み取り専用であり、権限制御は持たない。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 該当なし | - | - | ブロードキャスト変数に特化した画面は存在しないが、Storageページにブロードキャストブロックが表示されうる |

## 機能種別

データ配布 / P2P分散転送 / キャッシュ管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| obj | T | Yes | ブロードキャストするオブジェクト | シリアライズ可能 |
| id | Long | Yes | ブロードキャスト変数の一意識別子 | 0以上 |
| serializedOnly | Boolean | Yes | シリアライズ済みデータのみ保持するか | - |
| spark.broadcast.blockSize | Int | No | ブロックサイズ（KB） | デフォルト4MB（4096KB） |
| spark.broadcast.compress | Boolean | No | 圧縮有効化 | デフォルトtrue |
| spark.broadcast.checksum | Boolean | No | チェックサム検証有効化 | デフォルトfalse |

### 入力データソース

SparkContext.broadcast()メソッドからの呼び出し

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Broadcast[T] | Broadcast[T] | ブロードキャスト変数ラッパー |
| value | T | ブロードキャスト変数の値（getValue()で取得） |
| numBlocks | Int | データが分割されたブロック数 |

### 出力先

各Executorの BlockManager（メモリ/ディスク）

## 処理フロー

### 処理シーケンス

```
1. ブロードキャスト変数生成（ドライバー側）
   └─ TorrentBroadcastインスタンスの生成
2. データのシリアライズとブロック分割（writeBlocks）
   └─ オブジェクトをシリアライズし、blockSizeごとにチャンク分割
3. ブロックのBlockManager登録
   └─ 各ブロック（BroadcastBlockId(id, "piece" + i)）をMEMORY_AND_DISK_SERで保存
4. チェックサム計算（オプション）
   └─ 各ブロックのAdler32チェックサムを計算
5. Executor側でのブロック取得（readBlocks）
   └─ ランダム順でブロックを取得（ローカル→リモート）
6. データ再構成（readBroadcastBlock）
   └─ 全ブロックを結合してデシリアライズし、元のオブジェクトを復元
7. Executor側でのキャッシュ
   └─ 復元したオブジェクトをSoftReference/WeakReferenceで保持
```

### フローチャート

```mermaid
flowchart TD
    A[SparkContext.broadcast] --> B[TorrentBroadcast生成]
    B --> C[writeBlocks: データシリアライズ]
    C --> D[blockSizeごとにチャンク分割]
    D --> E[各ブロックをBlockManagerに登録]
    E --> F{チェックサム有効?}
    F -->|Yes| G[Adler32チェックサム計算]
    F -->|No| H[ドライバー側完了]
    G --> H
    H --> I[Executor: getValue呼び出し]
    I --> J{ローカルキャッシュあり?}
    J -->|Yes| K[キャッシュから返却]
    J -->|No| L[readBroadcastBlock]
    L --> M[readBlocks: ブロック取得]
    M --> N{ローカルにブロックあり?}
    N -->|Yes| O[ローカルから取得]
    N -->|No| P[リモートから取得]
    O --> Q[全ブロック結合・デシリアライズ]
    P --> Q
    Q --> R[SoftReference/WeakReferenceでキャッシュ]
    R --> K
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | P2P配布 | Executorが取得したブロックは他のExecutorにも提供される | 常時 |
| BR-02 | ランダム順取得 | ブロックの取得順序はランダム化され、ホットスポットを防止 | readBlocks時 |
| BR-03 | serializedOnly最適化 | 内部ブロードキャスト（serializedOnly=true）はドライバーのBlockManagerに保存しない（ローカルモード除く） | serializedOnly=true時 |
| BR-04 | 参照種別 | serializedOnly=trueの場合はWeakReference、falseの場合はSoftReferenceで値を保持 | getValue時 |
| BR-05 | チェックサム検証 | spark.broadcast.checksum=trueで各ブロックのAdler32チェックサムを検証 | 設定有効時 |

### 計算ロジック

ブロック分割: `blockSize = spark.broadcast.blockSize * 1024`（バイト単位）。デフォルト4MB。オブジェクトはシリアライズ後、blockSizeごとにチャンクに分割される。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | ブロードキャスト変数はデータベース操作を行わない |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| SparkException | 保存失敗 | BlockManagerへのブロック保存失敗 | ブロードキャスト変数のクリーンアップと例外送出 |
| SparkException | 取得失敗 | 全てのソースからブロック取得に失敗 | タスク失敗→リトライ |
| SparkException | チェックサム不一致 | ブロックのチェックサムが一致しない | ブロックの再取得 |

### リトライ仕様

ブロック取得失敗時はBlockManagerのリモートフェッチリトライメカニズムに従う。

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

ブロードキャスト変数の書き込みはドライバーでのみ行われ、Executorは読み取り専用アクセスのみ。writeBlocksの失敗時は、登録済みブロックをremoveBroadcastでクリーンアップする（178-184行目）。

## パフォーマンス要件

- ブロックサイズ（デフォルト4MB）によりネットワーク転送の並列性を制御
- 圧縮によりネットワーク転送量を削減
- P2P配布によりドライバーの負荷を分散

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

- ブロードキャストデータはシリアライズされた状態でBlockManager間を転送
- spark.authenticate有効時は通信が認証・暗号化される

## 備考

- TorrentBroadcastはSpark唯一のブロードキャスト実装（HttpBroadcastはSpark 2.0で削除）
- ContextCleanerにより不要なブロードキャスト変数のブロックが自動的にクリーンアップされる
- serializedOnly=trueの内部ブロードキャストはドライバーメモリ圧迫を防ぐためドライバーBlockManagerへの保存をスキップ（SPARK-39983）

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Broadcast.scala | `core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala` | Broadcast抽象クラスの定義 |
| 1-2 | BroadcastBlockId | `core/src/main/scala/org/apache/spark/storage/BlockId.scala` | ブロードキャストブロックIDの定義 |

**読解のコツ**: TorrentBroadcast.scalaの38-59行目のScaladocがBitTorrentライクな配布メカニズムを説明している。ドライバーがデータを小チャンクに分割し、Executor間でP2P的に交換する仕組みを理解することが重要。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | TorrentBroadcast.scala | `core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala` | コンストラクタとwriteBlocks() |

**主要処理フロー**:
- **60-61行目**: クラス定義（obj, id, serializedOnly）
- **84-95行目**: setConf: compressionCodec, blockSize, checksumEnabled設定
- **100行目**: writeBlocks(): データのシリアライズとブロック分割
- **139-186行目**: writeBlocks()の実装詳細
- **105-118行目**: getValue(): キャッシュ確認→readBroadcastBlock()

#### Step 3: 読み取りメカニズムを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TorrentBroadcast.scala | `core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala` | readBlocks(), readBroadcastBlock() |

**主要処理フロー**:
- **189-199行目**: readBlocks(): ランダム順でブロック取得開始
- readBroadcastBlock(): 全ブロック結合→デシリアライズ→キャッシュ

#### Step 4: ブロードキャストファクトリを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | TorrentBroadcastFactory.scala | `core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcastFactory.scala` | ファクトリによるインスタンス生成 |
| 4-2 | BroadcastManager.scala | `core/src/main/scala/org/apache/spark/broadcast/BroadcastManager.scala` | ブロードキャストマネージャ |

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

```
SparkContext.broadcast(value)
    |
    +-- BroadcastManager.newBroadcast()
            +-- TorrentBroadcastFactory.newBroadcast()
                    +-- TorrentBroadcast(obj, id, serializedOnly=false)
                            |
                            +-- setConf() [圧縮、ブロックサイズ設定]
                            +-- writeBlocks(obj)
                                    +-- blockifyObject() [シリアライズ + 分割]
                                    +-- BlockManager.putBytes() [各ブロック保存]
                                    +-- calcChecksum() [チェックサム計算]

[Executor側]
TorrentBroadcast.getValue()
    |
    +-- readBroadcastBlock()
            +-- readBlocks()
            |       +-- BlockManager.getLocalBytes() [ローカル取得]
            |       +-- BlockManager.getRemoteBytes() [リモート取得]
            +-- unBlockifyObject() [結合 + デシリアライズ]
            +-- BlockManager.putSingle() [キャッシュ保存]
```

### データフロー図

```
[ドライバー]                [処理]                         [Executor]

オブジェクト        ───▶  シリアライズ             ───▶  ブロック1..N
                          + ブロック分割                   (BlockManager)
                          + チェックサム計算

                          BlockManager              ───▶  ブロック取得（P2P）
                          (ブロック保存)                    |
                                                          +-- ローカル/リモート
                                                          +-- デシリアライズ
                                                          +-- オブジェクト復元
                                                          +-- SoftRef/WeakRefキャッシュ
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| TorrentBroadcast.scala | `core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala` | ソース | ブロードキャスト実装 |
| Broadcast.scala | `core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala` | ソース | Broadcast抽象クラス |
| TorrentBroadcastFactory.scala | `core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcastFactory.scala` | ソース | ファクトリ |
| BroadcastManager.scala | `core/src/main/scala/org/apache/spark/broadcast/BroadcastManager.scala` | ソース | ブロードキャストマネージャ |
| BroadcastFactory.scala | `core/src/main/scala/org/apache/spark/broadcast/BroadcastFactory.scala` | ソース | ファクトリインターフェース |
| BlockManager.scala | `core/src/main/scala/org/apache/spark/storage/BlockManager.scala` | ソース | ブロック保存・取得 |
| BlockId.scala | `core/src/main/scala/org/apache/spark/storage/BlockId.scala` | ソース | BroadcastBlockId定義 |
| ContextCleaner.scala | `core/src/main/scala/org/apache/spark/ContextCleaner.scala` | ソース | ブロードキャスト変数のクリーンアップ |
