# 機能設計書 2-RDD（Resilient Distributed Dataset）

## 概要

本ドキュメントは、Apache SparkのRDD（Resilient Distributed Dataset）機能の設計を記述する。RDDはSparkの基本的な分散データ抽象であり、耐障害性を持つ不変な分散コレクションとして、並列データ処理の基盤を提供する。

### 本機能の処理概要

RDDは、クラスタ上に分散されたデータの論理的な集合を表現する抽象クラスであり、変換（Transformation）とアクション（Action）の2種類の操作をサポートする。

**業務上の目的・背景**：大規模データの分散処理において、データの耐障害性（ノード障害時の自動復旧）と効率的な並列計算を両立させる必要がある。RDDはリネージ（変換履歴）に基づく再計算により耐障害性を実現しつつ、メモリ上でのキャッシュにより反復計算のパフォーマンスを向上させる。データレプリケーションではなく、計算のリネージ追跡による障害復旧という独自のアプローチを採用している。

**機能の利用シーン**：バッチデータ処理、ETLパイプライン、機械学習の反復計算、対話的データ分析など、あらゆるSparkワークロードの基盤として利用される。高レベルAPI（DataFrame/Dataset）も内部的にRDDを利用している。

**主要な処理内容**：
1. パーティション分割されたデータの並列計算（compute）
2. 変換操作（map, filter, flatMap, groupByKey, join等）による新RDDの生成
3. アクション操作（count, collect, reduce, save等）による結果の返却
4. ストレージレベル指定によるメモリ/ディスクへのキャッシュ（persist/cache）
5. チェックポイントによるリネージの切断と信頼性の確保
6. 依存関係（narrow/wide）の管理によるスケジューリング最適化

**関連システム・外部連携**：Hadoop HDFS（入出力）、各種ファイルシステム、JDBC（JdbcRDD）、外部データソース

**権限による制御**：RDD自体は権限制御を持たないが、入出力先のファイルシステムやデータソースの権限に従う。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 7 | Storage（ストレージ一覧） | 参照画面 | 永続化RDDの一覧表示、ストレージレベル・キャッシュ状態の確認 |
| 8 | RDD Detail（RDD詳細） | 参照画面 | 特定RDDのストレージ詳細（データ分布・パーティション情報）の表示 |

## 機能種別

分散データ処理基盤 / 変換・集約処理 / データキャッシュ管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| _sc | SparkContext | Yes | RDDを生成するSparkContext | null不可 |
| deps | Seq[Dependency[_]] | Yes | 親RDDへの依存関係リスト | - |
| partitioner | Option[Partitioner] | No | キーバリューRDDのパーティショナー | - |
| storageLevel | StorageLevel | No | キャッシュ時のストレージレベル | NONE/MEMORY_ONLY/MEMORY_AND_DISK等 |

### 入力データソース

- SparkContext.parallelize(): Scalaコレクションからの生成
- SparkContext.textFile(): テキストファイル（HDFS、S3、ローカル等）
- SparkContext.hadoopFile(): Hadoop InputFormatからの読み込み
- 親RDDからの変換操作による生成

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| partitions | Array[Partition] | RDDのパーティション配列 |
| dependencies | Seq[Dependency[_]] | 親RDDへの依存関係 |
| iterator | Iterator[T] | パーティションごとのデータイテレータ |
| preferredLocations | Seq[String] | パーティションの推奨実行場所 |

### 出力先

- collect/take: ドライバーメモリに結果を返却
- saveAsTextFile: 外部ファイルシステムに出力
- saveAsObjectFile: シリアライズされたオブジェクトファイルとして出力
- foreach: 各要素に対する副作用実行

## 処理フロー

### 処理シーケンス

```
1. RDD生成
   └─ SparkContext経由またはTransformationにより新RDDインスタンス生成
2. 変換チェーン構築（遅延評価）
   └─ map/filter等のTransformationで新しいRDDを生成、依存関係を記録
3. アクション実行トリガー
   └─ count/collect等のActionがSparkContext.runJobを呼び出し
4. DAGスケジューリング
   └─ DAGSchedulerがRDD依存関係グラフからステージを構築
5. パーティション計算
   └─ 各Executor上でRDD.compute()がパーティション単位で実行
6. 結果集約
   └─ ドライバーに結果を返却、またはファイルシステムに出力
```

### フローチャート

```mermaid
flowchart TD
    A[RDD生成] --> B[Transformation適用]
    B --> B
    B --> C[Action呼び出し]
    C --> D[DAGScheduler.submitJob]
    D --> E[ステージ分割]
    E --> F{キャッシュ済み?}
    F -->|Yes| G[キャッシュからデータ取得]
    F -->|No| H[RDD.compute実行]
    H --> I{チェックポイントあり?}
    I -->|Yes| J[チェックポイントからデータ取得]
    I -->|No| K[親RDDからデータ計算]
    G --> L[結果返却]
    J --> L
    K --> L
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | 不変性 | RDDは一度生成されると変更不可。変換操作は新しいRDDを生成する | 常時 |
| BR-02 | 遅延評価 | Transformationは即座に実行されず、Actionが呼ばれた時点で実行される | 常時 |
| BR-03 | パーティションインデックス一致 | partitions配列の各要素のindexプロパティがその配列インデックスと一致すること | 常時 |
| BR-04 | リネージベース耐障害性 | ノード障害時は依存関係グラフに基づきパーティションを再計算 | 障害発生時 |
| BR-05 | ネストRDD禁止 | RDD内にRDDをネストすることは非サポート（SPARK-5063） | 常時（警告） |

### 計算ロジック

RDDの5つの主要プロパティ:
1. `getPartitions`: パーティション配列の定義
2. `compute(split, context)`: パーティション計算ロジック
3. `getDependencies`: 親RDDへの依存関係
4. `partitioner`: キーバリューRDDのパーティション戦略（Optional）
5. `getPreferredLocations`: データローカリティ（Optional）

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | RDD自体はデータベース操作を直接行わない。JdbcRDDサブクラスがJDBC経由でRDBMSへアクセスする。 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| SparkCoreErrors | 状態エラー | SparkContextがnull（デシリアライズ後） | RDDはSparkContext内で使用すること |
| SparkException | 実行エラー | パーティション計算中の例外 | TaskSchedulerが一定回数リトライ後、ステージ失敗 |
| OutOfMemoryError | リソースエラー | メモリ不足によるパーティション計算失敗 | ストレージレベルの変更、パーティション数の増加 |

### リトライ仕様

パーティション計算の失敗はTaskSchedulerにより`spark.task.maxFailures`（デフォルト4回）までリトライされる。

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

RDDは不変データ構造であるため、トランザクション管理は不要。出力操作（saveAs系）は各パーティション単位で書き込みを行い、OutputCommitCoordinatorが重複書き込みを防止する。

## パフォーマンス要件

- パーティション数はデータサイズとクラスタリソースに応じて調整（推奨: タスクあたり128MB程度）
- persist(MEMORY_ONLY)によるキャッシュで反復計算のパフォーマンス向上
- wide dependency（シャッフル）を最小化することで性能最適化

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

- RDDデータはシリアライズされてExecutor間を転送される
- チェックポイントデータはHDFS等の外部ファイルシステムに保存されるため、ファイルシステムのアクセス制御に依存
- spark.authenticate有効時はノード間通信が認証される

## 備考

- RDDはSpark 1.x時代のプライマリAPIであり、Spark 2.0以降はDataFrame/Dataset APIの利用が推奨される
- ただし内部的にはDataFrame/DatasetもRDDを基盤としている
- 各種RDD実装（MapPartitionsRDD, ShuffledRDD, CoGroupedRDD等）はrddディレクトリ配下に存在する

---

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

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

### 推奨読解順序

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

RDDの抽象クラスと基本的なデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | RDD.scala | `core/src/main/scala/org/apache/spark/rdd/RDD.scala` | 抽象クラスの5つのプロパティ（partitions, compute, dependencies, partitioner, preferredLocations） |
| 1-2 | Dependency.scala | `core/src/main/scala/org/apache/spark/Dependency.scala` | NarrowDependencyとShuffleDependencyの違い |
| 1-3 | Partition.scala | `core/src/main/scala/org/apache/spark/Partition.scala` | パーティションインターフェース |
| 1-4 | Partitioner.scala | `core/src/main/scala/org/apache/spark/Partitioner.scala` | HashPartitioner、RangePartitioner |

**読解のコツ**: RDD.scalaの56-83行目のScaladocに5つの主要プロパティが記載されている。全てのRDD実装はこの5プロパティをオーバーライドする。`@transient`アノテーションが付いたフィールドはシリアライズ時にnullとなる点に注意。

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

RDD生成の起点となるSparkContextのメソッドを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SparkContext.scala | `core/src/main/scala/org/apache/spark/SparkContext.scala` | parallelize(), textFile(), hadoopFile()等のRDD生成メソッド |
| 2-2 | ParallelCollectionRDD.scala | `core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala` | Scalaコレクションからの基本的なRDD生成 |

#### Step 3: 変換操作を理解する

Transformation操作がどのように新しいRDDを生成するかを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | RDD.scala | `core/src/main/scala/org/apache/spark/rdd/RDD.scala` | map, filter, flatMap等の変換メソッド定義 |
| 3-2 | MapPartitionsRDD.scala | `core/src/main/scala/org/apache/spark/rdd/MapPartitionsRDD.scala` | 最も基本的なRDD実装。map/filterの内部実装 |
| 3-3 | ShuffledRDD.scala | `core/src/main/scala/org/apache/spark/rdd/ShuffledRDD.scala` | シャッフルを伴うRDD（groupByKey等の結果） |
| 3-4 | PairRDDFunctions.scala | `core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala` | キーバリューペア操作（groupByKey, reduceByKey, join等） |

#### Step 4: アクション操作とジョブ実行を理解する

Action操作がジョブ実行をトリガーする仕組みを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | RDD.scala | `core/src/main/scala/org/apache/spark/rdd/RDD.scala` | collect, count, reduce等のアクションメソッド |
| 4-2 | SparkContext.scala | `core/src/main/scala/org/apache/spark/SparkContext.scala` | runJobメソッド：アクションからDAGSchedulerへの橋渡し |

#### Step 5: キャッシュ/チェックポイントを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | RDD.scala | `core/src/main/scala/org/apache/spark/rdd/RDD.scala` | persist(), cache(), checkpoint()メソッド |
| 5-2 | RDDCheckpointData.scala | `core/src/main/scala/org/apache/spark/rdd/RDDCheckpointData.scala` | チェックポイントの管理 |
| 5-3 | ReliableCheckpointRDD.scala | `core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala` | 信頼性のあるチェックポイント（HDFS） |

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

```
SparkContext
    |
    +-- parallelize() / textFile() / hadoopFile()
    |       └── RDD[T] (ParallelCollectionRDD / HadoopRDD)
    |
    +-- RDD.map() / filter() / flatMap()
    |       └── MapPartitionsRDD[U]
    |
    +-- RDD.groupByKey() / reduceByKey()
    |       └── ShuffledRDD[K, V, C]
    |
    +-- RDD.collect() / count()
            └── SparkContext.runJob()
                    └── DAGScheduler.submitJob()
                            └── TaskScheduler.submitTasks()
                                    └── Executor.launchTask()
                                            └── RDD.compute()
```

### データフロー図

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

テキストファイル     ───▶  HadoopRDD            ───▶  MapPartitionsRDD
(HDFS/S3/Local)            |                           |
                           +-- compute()               +-- map/filter
                                                       |
Scalaコレクション   ───▶  ParallelCollectionRDD ───▶  ShuffledRDD
                           |                           |
                           +-- compute()               +-- groupByKey
                                                       |
                                                  ───▶  collect/save
                                                       [ドライバー/ファイル]
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| RDD.scala | `core/src/main/scala/org/apache/spark/rdd/RDD.scala` | ソース | RDD抽象クラス定義 |
| MapPartitionsRDD.scala | `core/src/main/scala/org/apache/spark/rdd/MapPartitionsRDD.scala` | ソース | map/filter等の基本変換RDD |
| ShuffledRDD.scala | `core/src/main/scala/org/apache/spark/rdd/ShuffledRDD.scala` | ソース | シャッフルRDD |
| PairRDDFunctions.scala | `core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala` | ソース | キーバリューRDD操作 |
| CoGroupedRDD.scala | `core/src/main/scala/org/apache/spark/rdd/CoGroupedRDD.scala` | ソース | 複数RDDの結合 |
| UnionRDD.scala | `core/src/main/scala/org/apache/spark/rdd/UnionRDD.scala` | ソース | RDD結合 |
| HadoopRDD.scala | `core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala` | ソース | Hadoopファイル読み込み |
| ParallelCollectionRDD.scala | `core/src/main/scala/org/apache/spark/rdd/ParallelCollectionRDD.scala` | ソース | コレクションからのRDD生成 |
| Dependency.scala | `core/src/main/scala/org/apache/spark/Dependency.scala` | ソース | 依存関係定義 |
| Partition.scala | `core/src/main/scala/org/apache/spark/Partition.scala` | ソース | パーティション定義 |
| Partitioner.scala | `core/src/main/scala/org/apache/spark/Partitioner.scala` | ソース | パーティショナー定義 |
| RDDCheckpointData.scala | `core/src/main/scala/org/apache/spark/rdd/RDDCheckpointData.scala` | ソース | チェックポイント管理 |
| RDDOperationScope.scala | `core/src/main/scala/org/apache/spark/rdd/RDDOperationScope.scala` | ソース | 操作スコープ管理 |
| StorageLevel.scala | `core/src/main/scala/org/apache/spark/storage/StorageLevel.scala` | ソース | ストレージレベル定義 |
