# 機能設計書 7-ブロックストレージ管理

## 概要

本ドキュメントは、Apache Sparkのブロックストレージ管理機能の設計を記述する。BlockManagerはデータブロック（RDDパーティション、シャッフルデータ、ブロードキャスト変数等）のメモリ・ディスク上での保存と取得を管理する。

### 本機能の処理概要

BlockManagerは、各Executor上でデータブロックのライフサイクルを管理し、メモリストア・ディスクストアへの保存、リモートノードからのブロック取得、レプリケーション、デコミッション時のブロックマイグレーションを行う中核ストレージコンポーネントである。

**業務上の目的・背景**：分散処理において、RDDのキャッシュ、シャッフルの中間データ、ブロードキャスト変数等のデータブロックを効率的に管理する必要がある。BlockManagerはメモリとディスクの階層ストレージを提供し、データローカリティの最大化とI/Oの最小化を実現する。

**機能の利用シーン**：RDDのpersist/cache時、シャッフルデータの書き込み/読み取り時、ブロードキャスト変数の配布時など、データブロックの保存・取得が必要な全ての場面で利用される。

**主要な処理内容**：
1. ブロックの保存（putSingle, putBytes, putIterator）
2. ブロックの取得（getLocalValues, getLocalBytes, getRemoteValues, getRemoteBytes）
3. メモリストアとディスクストアの階層管理
4. ブロックロック管理（BlockInfoManager）
5. リモートブロックの取得（シャッフルデータフェッチ）
6. ブロックレプリケーション
7. ブロックのエビクション（ストレージメモリ不足時）
8. デコミッション時のブロックマイグレーション

**関連システム・外部連携**：MemoryManager（メモリ割り当て）、DiskBlockManager（ディスクI/O）、BlockManagerMaster（ドライバー側の管理）、ExternalShuffleService

**権限による制御**：BlockManager自体は権限制御を持たないが、セキュリティマネージャにより通信の認証・暗号化が行われる。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 7 | Storage（ストレージ一覧） | 主機能 | RDDのストレージレベル・キャッシュパーティション数・メモリ/ディスク使用量を取得して表示 |
| 8 | RDD Detail（RDD詳細） | 主機能 | RDDのExecutor別データ分布を取得して表示 |

## 機能種別

データストレージ管理 / キャッシュ管理 / ブロックI/O

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| blockId | BlockId | Yes | ブロック一意識別子 | null不可 |
| data | Any / ByteBuffer / Iterator | Yes | 保存するデータ | - |
| level | StorageLevel | Yes | ストレージレベル（MEMORY_ONLY, MEMORY_AND_DISK等） | 有効なStorageLevel |
| tellMaster | Boolean | Yes | マスターへの登録通知要否 | - |
| keepReadLock | Boolean | No | 読み取りロックの保持要否 | デフォルトfalse |

### 入力データソース

- RDD.iterator(): キャッシュされたRDDパーティションの取得
- ShuffleWriter: シャッフルマップ出力の書き込み
- TorrentBroadcast: ブロードキャスト変数のブロック保存

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| BlockResult | BlockResult | 取得したブロックデータとメトリクス |
| Boolean | Boolean | ブロック保存成功フラグ |
| BlockStatus | BlockStatus | ブロックの状態情報（ストレージレベル、サイズ等） |

### 出力先

呼び出し元への返却、BlockManagerMasterへの状態報告

## 処理フロー

### 処理シーケンス

```
1. ブロック保存要求（put系メソッド）
   └─ BlockInfoManagerでライトロックを取得
2. ストレージレベルに基づく保存先決定
   └─ メモリ/ディスク/両方への保存
3. メモリストアへの保存
   └─ MemoryManagerからストレージメモリを割り当て
4. ディスクストアへの保存（必要な場合）
   └─ DiskBlockManagerで物理ファイルに書き込み
5. マスターへの報告
   └─ BlockManagerMasterEndpointにブロック状態を送信
6. レプリケーション（必要な場合）
   └─ ストレージレベルの複製数に応じてリモートノードに複製
7. ブロック取得要求（get系メソッド）
   └─ ローカル→リモートの順で検索・取得
```

### フローチャート

```mermaid
flowchart TD
    A[put要求] --> B[ライトロック取得]
    B --> C{メモリ保存?}
    C -->|Yes| D[MemoryStore.putBytes/putIterator]
    C -->|No| E[DiskStore.putBytes]
    D --> F{メモリ保存成功?}
    F -->|Yes| G{ディスクにも保存?}
    F -->|No| H{ディスクフォールバック?}
    H -->|Yes| E
    H -->|No| I[保存失敗]
    G -->|Yes| E
    G -->|No| J[マスター報告]
    E --> J
    J --> K{レプリケーション必要?}
    K -->|Yes| L[リモートノードに複製]
    K -->|No| M[ロック解放]
    L --> M
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | ストレージレベル制御 | MEMORY_ONLY, MEMORY_AND_DISK, MEMORY_ONLY_SER等のレベルに応じた保存戦略 | put時 |
| BR-02 | ローカル優先検索 | ブロック取得はローカル→リモートの順で検索 | get時 |
| BR-03 | エビクション | メモリ不足時はLRUに基づきブロックをエビクト | メモリ割り当て失敗時 |
| BR-04 | ロック管理 | read/writeロックによる並行制御 | put/get時 |
| BR-05 | デコミッション | Executorデコミッション時にブロックを他ノードにマイグレーション | デコミッション時 |

### 計算ロジック

ブロックID体系:
- RDDBlockId(rddId, splitIndex): RDDパーティション
- ShuffleBlockId(shuffleId, mapId, reduceId): シャッフルデータ
- BroadcastBlockId(broadcastId, field): ブロードキャスト変数
- TaskResultBlockId(taskId): タスク結果

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | BlockManagerはファイルシステムベースのストレージを使用。データベースは使用しない。 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| BlockNotFoundException | ブロック未検出 | 要求ブロックがローカル・リモートとも存在しない | 再計算（RDDリネージ経由） |
| SparkException | 保存失敗 | メモリ/ディスクへの書き込み失敗 | ストレージレベル変更、ディスク容量確認 |
| BlockSavedOnDecommissionedBMException | デコミッション | デコミッション済みBlockManagerへの保存試行 | 別のBlockManagerを選択 |

### リトライ仕様

リモートブロック取得はShuffleBlockFetcherIteratorにより自動リトライされる。リトライ回数はspark.shuffle.io.maxRetries設定で制御。

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

ブロック操作はBlockInfoManagerのロックメカニズムにより保護される。ライトロックは排他的、リードロックは共有的。

## パフォーマンス要件

- メモリストアI/O: マイクロ秒オーダー
- ディスクストアI/O: ミリ秒オーダー（ディスク性能に依存）
- リモートブロック取得: ネットワーク帯域に依存

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

- BlockManager間のデータ転送はNetty通信基盤を使用
- spark.authenticate有効時は通信が認証・暗号化される
- External Shuffle Service使用時はシャッフルデータが永続化される

## 備考

- BlockManagerはドライバーとExecutorの両方で動作する
- ドライバー上のBlockManagerMasterがグローバルなブロック位置情報を管理する
- Decommission機能によりExecutor退避時のデータロスを防止する

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | BlockId.scala | `core/src/main/scala/org/apache/spark/storage/BlockId.scala` | ブロックID体系の定義 |
| 1-2 | BlockInfoManager.scala | `core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala` | ブロックメタデータとロック管理 |
| 1-3 | StorageLevel.scala | `core/src/main/scala/org/apache/spark/storage/StorageLevel.scala` | ストレージレベル定義 |

**読解のコツ**: BlockIdのサブクラス（RDDBlockId, ShuffleBlockId, BroadcastBlockId等）がブロックの種類を表現する。apply/unapplyメソッドでブロックID文字列との相互変換が行われる。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | BlockManager.scala | `core/src/main/scala/org/apache/spark/storage/BlockManager.scala` | putSingle(), getLocalValues(), getRemoteBytes()等の主要メソッド |

#### Step 3: メモリ/ディスクストアを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | MemoryStore.scala | `core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala` | メモリ上のブロック保存 |
| 3-2 | DiskStore.scala | `core/src/main/scala/org/apache/spark/storage/DiskStore.scala` | ディスク上のブロック保存 |
| 3-3 | DiskBlockManager.scala | `core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala` | ディスクファイル管理 |

#### Step 4: マスター連携を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | BlockManagerMaster.scala | `core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala` | ドライバー側管理API |
| 4-2 | BlockManagerMasterEndpoint.scala | `core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala` | ドライバー側RPCエンドポイント |

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

```
RDD.iterator() / ShuffleWriter / TorrentBroadcast
    |
    +-- BlockManager.putSingle() / putBytes() / putIterator()
    |       +-- BlockInfoManager.lockForWriting()
    |       +-- MemoryStore.putBytes() / putIterator()
    |       |       +-- MemoryManager.acquireStorageMemory()
    |       +-- DiskStore.putBytes()
    |       |       +-- DiskBlockManager.getFile()
    |       +-- BlockManagerMaster.updateBlockInfo()
    |       +-- replicate() [レプリケーション時]
    |
    +-- BlockManager.getLocalValues() / getRemoteBytes()
            +-- BlockInfoManager.lockForReading()
            +-- MemoryStore.getValues() / getBytes()
            +-- DiskStore.getBytes()
            +-- ShuffleBlockFetcherIterator [リモート取得時]
```

### データフロー図

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

データブロック       ───▶  BlockManager.put()       ───▶  MemoryStore/DiskStore
(RDDパーティション等)       |                               |
                            +-- ストレージレベル判定         +-- BlockManagerMaster報告
                            +-- メモリ割り当て
                            +-- レプリケーション

ブロックID          ───▶  BlockManager.get()       ───▶  BlockResult
                            |                               (データ + メトリクス)
                            +-- ローカル検索
                            +-- リモート検索
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| BlockManager.scala | `core/src/main/scala/org/apache/spark/storage/BlockManager.scala` | ソース | ブロック管理のメインクラス |
| BlockId.scala | `core/src/main/scala/org/apache/spark/storage/BlockId.scala` | ソース | ブロックID定義 |
| BlockInfoManager.scala | `core/src/main/scala/org/apache/spark/storage/BlockInfoManager.scala` | ソース | ブロックメタデータ・ロック管理 |
| BlockManagerMaster.scala | `core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala` | ソース | ドライバー側管理 |
| BlockManagerMasterEndpoint.scala | `core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala` | ソース | ドライバー側RPCエンドポイント |
| DiskBlockManager.scala | `core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala` | ソース | ディスクファイル管理 |
| DiskStore.scala | `core/src/main/scala/org/apache/spark/storage/DiskStore.scala` | ソース | ディスクストア |
| MemoryStore.scala | `core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala` | ソース | メモリストア |
| BlockManagerDecommissioner.scala | `core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala` | ソース | デコミッション管理 |
| ShuffleBlockFetcherIterator.scala | `core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala` | ソース | リモートシャッフルブロック取得 |
| BlockReplicationPolicy.scala | `core/src/main/scala/org/apache/spark/storage/BlockReplicationPolicy.scala` | ソース | レプリケーションポリシー |
| StorageLevel.scala | `core/src/main/scala/org/apache/spark/storage/StorageLevel.scala` | ソース | ストレージレベル定義 |
