# 機能設計書 21-DataFrame/Dataset API

## 概要

本ドキュメントは、Apache SparkのDataFrame/Dataset APIの機能設計について記述する。DataFrame/Dataset APIは、Spark SQLの中核を成す型安全な分散データ操作APIであり、構造化データの変換・集約・結合等の操作を宣言的に記述可能にする。

### 本機能の処理概要

DataFrame/Dataset APIは、Sparkにおける構造化データ処理の主要なプログラミングインターフェースである。Dataset[T]は型安全なコレクションであり、DataFrame（Dataset[Row]の型エイリアス）はスキーマ付きの非型安全なビューを提供する。これらのAPIを通じて、SQLライクな操作を関数型またはリレーショナル操作として記述できる。

**業務上の目的・背景**：大規模データの分散処理において、低レベルのRDD APIでは開発者の負担が大きく、またクエリ最適化の恩恵を受けられない。DataFrame/Dataset APIは、宣言的なデータ操作を可能にし、Catalystオプティマイザによる自動最適化とTungstenエンジンによる高効率実行を実現する。これにより、開発生産性とクエリ性能の両方を向上させる。

**機能の利用シーン**：ETLパイプラインの構築、データ分析・集計処理、機械学習の前処理、ストリーミングデータの構造化処理、外部データソース（Parquet, JSON, JDBC等）からのデータ読み込みと書き出し等、Sparkを用いるほぼ全てのデータ処理シーンで利用される。

**主要な処理内容**：
1. SparkSessionを通じたDataFrame/Datasetの生成（read API、createDataFrame等）
2. 変換操作（select, filter, map, flatMap, groupBy, agg, join, union等）
3. アクション操作（count, collect, show, write等）
4. スキーマ管理（schema, printSchema, as[T]による型変換）
5. カラム操作（Column型を通じた式の構築とカラム参照）
6. 集約操作（RelationalGroupedDataset, KeyValueGroupedDataset）
7. 欠損値処理（DataFrameNaFunctions）
8. 統計関数（DataFrameStatFunctions）
9. データの読み込み・書き出し（DataFrameReader, DataFrameWriter）

**関連システム・外部連携**：Catalystオプティマイザ（論理計画最適化）、Tungstenエンジン（物理実行）、各種データソース（Parquet, ORC, CSV, JSON, JDBC等）、Hiveメタストア（テーブルメタデータ）、Spark Connect（リモートクライアント）

**権限による制御**：テーブルアクセスに対するACL制御はカタログ層で管理される。DataFrame API自体には権限制御機構はないが、基盤となるカタログやデータソースのセキュリティ機構と連携する。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 14 | SQL All Executions（SQL実行一覧） | 結果表示画面 | DataFrame/Dataset操作の実行をSQL実行として追跡・表示 |
| 15 | SQL Execution Detail（SQL実行詳細） | 結果表示画面 | 物理実行計画のDAGグラフとノード情報の表示 |

## 機能種別

データ操作API / CRUD操作 / 計算処理 / データ連携

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| source | String | No | データソースフォーマット（parquet, json, csv等） | 登録済みデータソース名であること |
| schema | StructType | No | 入力データスキーマ | 有効なStructType構造であること |
| options | Map[String, String] | No | データソースオプション | データソース固有の検証 |
| path | String | 条件付き | データファイルパス | 有効なファイルシステムパスであること |

### 入力データソース

- ファイルシステム上のデータファイル（Parquet, ORC, CSV, JSON等）
- JDBC経由のRDBMS
- Hiveメタストアのテーブル
- RDD（createDataFrame経由）
- ローカルコレクション（Seq, List等からの変換）
- ストリーミングソース（Kafka等）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Dataset[T] | Dataset[T] | 型安全な分散データセット |
| DataFrame | Dataset[Row] | スキーマ付きの行コレクション |
| Row | Row | 個別行データ（collectアクション時） |
| スカラー値 | Any | count, first等のアクション結果 |

### 出力先

- メモリ上のコレクション（collect、take等のアクション結果）
- ファイルシステム（write APIによるファイル出力）
- JDBC経由のRDBMS
- コンソール（show メソッド）
- ストリーミングシンク

## 処理フロー

### 処理シーケンス

```
1. SparkSession.read / createDataFrame でDataFrame生成
   └─ DataFrameReaderが指定フォーマットのDataSourceを解決し、LogicalPlanを構築
2. 変換操作の適用（select, filter, join等）
   └─ 各操作が新しいLogicalPlanノードを追加（遅延評価）
3. アクション操作の実行（collect, count, write等）
   └─ QueryExecutionが呼び出され、Catalystオプティマイザが計画を最適化
4. 物理実行計画の生成
   └─ SparkPlannerが論理計画を物理計画に変換
5. WholeStageCodegenによるコード生成（該当する場合）
   └─ 物理計画からJavaバイトコードを動的生成
6. RDDの生成と実行
   └─ SparkContextを通じてクラスタ上でタスクを実行
7. 結果の収集・出力
   └─ ドライバーへの結果返却またはデータソースへの書き出し
```

### フローチャート

```mermaid
flowchart TD
    A[SparkSession.read/createDataFrame] --> B[LogicalPlan構築]
    B --> C[変換操作の適用]
    C --> D{アクション呼び出し?}
    D -->|No| C
    D -->|Yes| E[QueryExecution開始]
    E --> F[Analyzer: 未解決属性の解決]
    F --> G[Optimizer: ルール・コストベース最適化]
    G --> H[SparkPlanner: 物理計画生成]
    H --> I[WholeStageCodegen]
    I --> J[RDD生成・タスク実行]
    J --> K[結果収集/出力]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-21-01 | 遅延評価 | 変換操作はアクションが呼ばれるまで実行されない | 全ての変換操作 |
| BR-21-02 | 不変性 | 各変換操作は新しいDatasetを返し、元のDatasetは変更されない | 全ての変換操作 |
| BR-21-03 | スキーマ推論 | スキーマ未指定時にデータソースからスキーマを自動推論する | JSON, CSV等のデータソース読み込み時 |
| BR-21-04 | Encoder要件 | Dataset[T]の生成にはEncoder[T]が必要 | as[T]による型変換時 |
| BR-21-05 | Null安全性 | Nullable属性はスキーマで管理され、型安全性を保証する | 全てのカラム操作 |

### 計算ロジック

DataFrame/Dataset APIの計算ロジックはCatalystオプティマイザに委譲される。主な最適化には、述語プッシュダウン、カラムプルーニング、定数畳み込み、結合順序最適化等がある。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| read.table | カタログテーブル | SELECT | メタストアからテーブルメタデータを取得し、データを読み込む |
| write.saveAsTable | カタログテーブル | INSERT/CREATE | データをテーブルとして保存する |
| write.insertInto | カタログテーブル | INSERT | 既存テーブルにデータを挿入する |

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

#### カタログテーブル

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| SELECT | 全カラム | テーブル名による指定 | カラムプルーニングが適用される |
| INSERT | 全カラム | DataFrameのスキーマに基づく | スキーマ互換性チェックあり |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| AnalysisException | 分析エラー | 存在しないカラム参照、型不一致 | カラム名・型の確認 |
| UNRESOLVED_COLUMN | 未解決カラム | selectで存在しないカラム名を指定 | 正しいカラム名を指定 |
| DATATYPE_MISMATCH | 型不一致 | 演算で型が合わない | 明示的キャストの適用 |
| SparkException | 実行時エラー | タスク実行中の障害 | データ・設定の確認 |
| ENCODER_NOT_FOUND | Encoder未検出 | as[T]でEncoderが見つからない | Encoder暗黙変換のインポート |

### リトライ仕様

タスクレベルのリトライはTaskSchedulerにより管理される。デフォルトでspark.task.maxFailures=4回のリトライが行われる。

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

DataFrame API自体にはトランザクション機構はない。データソースへの書き込み時には、OutputCommitCoordinatorによりタスクの出力コミットが調整される。SaveModeにより、Overwrite/Append/ErrorIfExists/Ignoreの動作を制御できる。

## パフォーマンス要件

- Catalystオプティマイザによる自動クエリ最適化
- WholeStageCodegenによるJITコンパイル最適化
- Tungstenメモリ管理によるGC負荷軽減
- カラムナフォーマットでの効率的なデータ処理
- データローカリティを考慮したタスクスケジューリング

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

- テーブルアクセスはカタログのACLにより制御
- JDBC接続時の認証情報はオプションで指定（パスワードのマスク処理あり）
- Spark UIでのSQLクエリ表示時にセンシティブ情報がリダクションされる

## 備考

- DataFrameはDataset[Row]の型エイリアスとして定義されている
- Spark 2.0以降、DataFrameとDatasetは統合されたAPIとして提供されている
- Spark Connectクライアントからもリモートで同じAPIが利用可能

---

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

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

### 推奨読解順序

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

まず、DataFrameとDatasetの基本的なデータ構造とスキーマ表現を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Row.scala | `sql/api/src/main/scala/org/apache/spark/sql/Row.scala` | DataFrameの行データ表現、フィールドアクセス方式 |
| 1-2 | types/StructType.scala | `sql/api/src/main/scala/org/apache/spark/sql/types/StructType.scala` | スキーマ定義構造、カラム型情報 |
| 1-3 | Encoder.scala | `sql/api/src/main/scala/org/apache/spark/sql/Encoder.scala` | Dataset[T]のシリアライズ/デシリアライズ定義 |
| 1-4 | Column.scala | `sql/api/src/main/scala/org/apache/spark/sql/Column.scala` | カラム式の構築、演算子定義 |

**読解のコツ**: Column.scalaでは、`Column.fn()`メソッド（58-68行目）がUnresolvedFunctionノードを生成する仕組みを理解することが重要。全ての式操作は内部的にColumnNodeとして表現される。

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

SparkSessionがDataFrame/Dataset APIのエントリーポイントとなる。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SparkSession.scala | `sql/api/src/main/scala/org/apache/spark/sql/SparkSession.scala` | read, createDataFrame等のファクトリメソッド |
| 2-2 | DataFrameReader.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameReader.scala` | データソースからのDataFrame生成フロー |

**主要処理フロー**:
1. **38-48行目**（DataFrameReader.scala）: format()でデータソース名を設定
2. **57-64行目**（DataFrameReader.scala）: schema()でユーザー指定スキーマを設定
3. **63-72行目**（SparkSession.scala）: sparkContext, sharedState, sessionStateの参照構造

#### Step 3: Dataset操作の実装を理解する

Dataset.scalaが変換・アクション操作の定義の中心となる。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Dataset.scala | `sql/api/src/main/scala/org/apache/spark/sql/Dataset.scala` | 変換操作（select, filter, join等）とアクション操作（collect, count等）の定義 |
| 3-2 | RelationalGroupedDataset.scala | `sql/api/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala` | groupBy後の集約操作 |
| 3-3 | KeyValueGroupedDataset.scala | `sql/api/src/main/scala/org/apache/spark/sql/KeyValueGroupedDataset.scala` | 型安全なgroupByKey操作 |
| 3-4 | DataFrameWriter.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala` | データ書き出しAPI |

**主要処理フロー**:
- **125行目**（Dataset.scala）: abstract class Dataset[T]の定義、sparkSessionとencoderが基本フィールド
- **131行目**（Dataset.scala）: queryExecutionフィールドがClassicOnlyで定義
- **143行目**（Dataset.scala）: toDF()でDataFrameへの変換
- **164行目**（Dataset.scala）: as[U]で型変換

#### Step 4: クエリ実行フローを理解する

QueryExecutionがDatasetの操作を実行する際の中核となる。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | QueryExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala` | 論理計画から物理計画への変換フロー |
| 4-2 | DataFrameNaFunctions.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameNaFunctions.scala` | 欠損値処理（fill, drop, replace） |
| 4-3 | DataFrameStatFunctions.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameStatFunctions.scala` | 統計関数（approxQuantile, corr, cov） |

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

```
SparkSession
    │
    ├─ DataFrameReader
    │      ├─ format() / schema() / option()
    │      └─ load() / parquet() / json() / csv()
    │             └─ DataSource.resolveRelation()
    │                    └─ LogicalPlan生成
    │
    ├─ Dataset[T]
    │      ├─ 変換操作（遅延評価）
    │      │      ├─ select() → Project(LogicalPlan)
    │      │      ├─ filter() → Filter(LogicalPlan)
    │      │      ├─ join() → Join(LogicalPlan)
    │      │      ├─ groupBy() → RelationalGroupedDataset
    │      │      │      └─ agg() → Aggregate(LogicalPlan)
    │      │      └─ union() → Union(LogicalPlan)
    │      │
    │      └─ アクション操作（即時実行）
    │             ├─ collect() → QueryExecution.executedPlan.executeCollect()
    │             ├─ count() → QueryExecution → Aggregate
    │             └─ show() → take() → フォーマット表示
    │
    └─ DataFrameWriter
           ├─ mode() / format() / option()
           └─ save() / saveAsTable() / insertInto()
                  └─ DataSource.write()
```

### データフロー図

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

データソース ──────▶ DataFrameReader ──────▶ LogicalPlan
(Parquet/JSON/                                     │
 CSV/JDBC/Hive)                                    ▼
                                           Analyzer（解析）
                                                   │
                                                   ▼
                                           Optimizer（最適化）
                                                   │
                                                   ▼
                                           SparkPlanner（物理計画）
                                                   │
                                                   ▼
                                           WholeStageCodegen
                                                   │
                                                   ▼
                                           RDD実行 ──────────▶ 結果/ファイル/DB
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Dataset.scala | `sql/api/src/main/scala/org/apache/spark/sql/Dataset.scala` | ソース | DataFrameとDatasetの主要API定義（3301行） |
| Column.scala | `sql/api/src/main/scala/org/apache/spark/sql/Column.scala` | ソース | カラム式の定義と演算子 |
| SparkSession.scala | `sql/api/src/main/scala/org/apache/spark/sql/SparkSession.scala` | ソース | APIエントリーポイント |
| DataFrameReader.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameReader.scala` | ソース | データ読み込みAPI |
| DataFrameWriter.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala` | ソース | データ書き出しAPI |
| DataFrameWriterV2.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala` | ソース | V2データソース向け書き出しAPI |
| Row.scala | `sql/api/src/main/scala/org/apache/spark/sql/Row.scala` | ソース | 行データ表現 |
| Encoder.scala | `sql/api/src/main/scala/org/apache/spark/sql/Encoder.scala` | ソース | 型エンコーダインターフェース |
| RelationalGroupedDataset.scala | `sql/api/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala` | ソース | groupBy後の集約操作 |
| KeyValueGroupedDataset.scala | `sql/api/src/main/scala/org/apache/spark/sql/KeyValueGroupedDataset.scala` | ソース | 型安全なグループ操作 |
| DataFrameNaFunctions.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameNaFunctions.scala` | ソース | 欠損値処理 |
| DataFrameStatFunctions.scala | `sql/api/src/main/scala/org/apache/spark/sql/DataFrameStatFunctions.scala` | ソース | 統計関数 |
| QueryExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala` | ソース | クエリ実行管理 |
