# 機能設計書 90-pandas API on Spark

## 概要

本ドキュメントは、pandasと互換性のあるAPIを提供し、既存のpandasコードをSparkの分散処理エンジン上で実行可能にするpandas API on Sparkの機能設計を記述する。

### 本機能の処理概要

pandas API on Spark（旧Koalas）は、pandasの DataFrame/Series API と互換性のあるインターフェースを提供し、既存のpandasコードを最小限の変更でSparkの分散処理エンジン上で実行可能にする。内部的にはSpark DataFrameに変換して処理を行い、pandasライクなAPIを透過的に提供する。

**業務上の目的・背景**：pandasはPythonデータ分析の事実上の標準ライブラリであるが、単一マシンのメモリに収まるデータサイズに制限される。pandas API on Sparkは、pandasの使い慣れたAPIを維持しながら、Sparkの分散処理能力を活用することで、大規模データセットに対するpandasライクな操作を可能にする。データサイエンティストの学習コストを最小限に抑えつつ、スケーラビリティを獲得できる。

**機能の利用シーン**：pandasで開発したデータ処理コードの大規模データへのスケールアウト、Jupyter Notebookでの大規模データ探索、pandasユーザーがSparkを使い始める際のブリッジ、ETLパイプラインでのpandas互換処理。

**主要な処理内容**：
1. DataFrame API：pandas.DataFrame互換の分散DataFrameを提供
2. Series API：pandas.Series互換の分散Seriesを提供
3. Index API：pandas.Index互換のインデックス管理を提供
4. GroupBy API：グループ化集約のpandas互換APIを提供
5. I/O API：read_csv、read_parquet、read_sql等のデータ読み書きAPIを提供
6. pandas変換：from_pandas/to_pandasでpandas DataFrameとの相互変換

**関連システム・外部連携**：pandas、PyArrow、Spark SQL Engine

**権限による制御**：Sparkの認証・認可機構に依存

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能はPython APIであり直接関連する画面はない |

## 機能種別

API / データ処理フレームワーク

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| data | dict/list/ndarray/DataFrame | Yes | DataFrameの元データ | - |
| index | Index | No | インデックス | - |
| columns | list | No | カラム名 | - |
| dtype | str/type | No | データ型 | - |

### 入力データソース

Python辞書、リスト、NumPy配列、pandas DataFrame、CSVファイル、Parquetファイル、SQLクエリ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| DataFrame | pyspark.pandas.DataFrame | pandas互換の分散DataFrame |
| Series | pyspark.pandas.Series | pandas互換の分散Series |
| Index | pyspark.pandas.Index | インデックスオブジェクト |
| pandas.DataFrame | pandas.DataFrame | to_pandas()で変換したローカルDataFrame |

### 出力先

Pythonアプリケーション（メモリ上）、CSVファイル、Parquetファイル等

## 処理フロー

### 処理シーケンス

```
1. pandas API on Spark初期化
   └─ pyspark.pandas パッケージインポート、pandas/pyarrowバージョン検証
2. DataFrame生成
   └─ ps.DataFrame(data) or ps.read_csv() / ps.read_parquet()
3. 内部Spark DataFrame構築
   └─ pandas DataFrameをApache Arrow経由でSpark DataFrameに変換
4. pandas API操作
   └─ df.groupby(), df.merge(), df.apply()等のpandas互換メソッド
5. Spark SQL変換
   └─ pandas API呼び出しをSpark SQLの変換/アクションに変換
6. 結果取得
   └─ to_pandas()でローカルpandas DataFrameに変換
```

### フローチャート

```mermaid
flowchart TD
    A[import pyspark.pandas as ps] --> B{バージョン検証}
    B -->|OK| C[ps.DataFrame生成]
    B -->|NG| D[ImportError]
    C --> E[内部Spark DataFrame構築]
    E --> F[pandas API操作]
    F --> G{操作タイプ}
    G -->|変換| H[遅延評価（Spark Transformation）]
    G -->|アクション| I[即時評価（Spark Action）]
    H --> F
    I --> J[結果返却]
    J --> K{pandas変換要?}
    K -->|Yes| L[to_pandas / Arrow転送]
    K -->|No| M[pandas API DataFrame返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-90-01 | pandas最小バージョン | require_minimum_pandas_version()でpandasの最小バージョンを検証 | モジュールインポート時 |
| BR-90-02 | PyArrow最小バージョン | require_minimum_pyarrow_version()でPyArrowの最小バージョンを検証 | モジュールインポート時 |
| BR-90-03 | SPARK_TESTINGモード | テスト環境でバージョン不足の場合は警告を出して終了する | SPARK_TESTING環境変数設定時 |
| BR-90-04 | 使用状況ロガー | PYSPARK_PANDAS_USAGE_LOGGER環境変数で使用状況ログモジュールを指定可能 | 環境変数設定時 |

### 計算ロジック

pandas API操作は内部的にSpark SQLの式ツリーに変換される。例えばDataFrame.groupby().agg()はSpark SQLのGROUP BY + 集約関数に変換される。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| read_sql | 任意 | SELECT | SQLクエリ結果をDataFrameとして読み込み |
| read_sql_table | 任意 | SELECT | テーブル全体をDataFrameとして読み込み |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ImportError | 依存パッケージ不足 | pandas/pyarrowが未インストールまたはバージョン不足 | pip installで適切なバージョンを導入 |
| PySparkValueError | 値エラー | 不正なパラメータが渡された場合 | パラメータを修正 |
| NotImplementedError | 未実装 | pandasにはあるがpyspark.pandasでは未実装の操作 | Spark DataFrameのAPIを直接使用 |

### リトライ仕様

特定のリトライ機構はない。Spark SQLエンジンのリトライ機構に依存する。

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

該当なし（Spark SQLのトランザクション制御に依存）。

## パフォーマンス要件

- Apache Arrow経由のpandas-Spark DataFrame変換で高速なデータ転送を実現
- Sparkの遅延評価により不要な計算を回避
- pandas_udfによるベクトル化処理でUDFの実行効率を向上

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

- Sparkの認証・認可機構に依存
- read_sql等のデータベースアクセスはJDBCデータソースのセキュリティ設定に依存

## 備考

pandas API on SparkはSpark 3.2でpyspark.pandasとして統合された（旧Koalas）。pandasのすべてのAPIが実装されているわけではなく、MissingPandasLikeGeneralFunctions等で未実装APIのスタブが提供されている。

---

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

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

### 推奨読解順序

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

pandas API on Sparkの全体構造とエクスポートAPIを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | __init__.py | `python/pyspark/pandas/__init__.py` | **18-100行目**: パッケージ初期化、pandas/PyArrowバージョン検証、エクスポートAPI一覧（__all__）、使用状況ロガー |
| 1-2 | config.py | `python/pyspark/pandas/config.py` | pandas API on Spark固有の設定オプション |

**読解のコツ**: `__all__`リストがパブリックAPIの一覧を定義する。`_auto_patch_spark()`が初期化時に実行される。MissingPandasLike*クラスが未実装APIのスタブを提供する。

#### Step 2: DataFrame実装を理解する

pandas互換DataFrameの実装を追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | frame.py | `python/pyspark/pandas/frame.py` | **18-80行目**: Spark DataFrameをラップするpandas互換DataFrame実装。NumPy/pandas/PySpark SQLの各種インポート |

**主要処理フロー**:
1. **21行目**: "A wrapper class for Spark DataFrame to behave like pandas DataFrame"
2. **54-61行目**: numpy, pandas, pandas.api.typesからの各種インポート
3. **71-80行目**: PySpark SQL Column, DataFrame, functionsの利用

#### Step 3: Seriesとインデックスを理解する

Series/Index実装を追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | series.py | `python/pyspark/pandas/series.py` | Series実装 |
| 3-2 | indexes/base.py | `python/pyspark/pandas/indexes/base.py` | 基本Index実装 |
| 3-3 | indexes/multi.py | `python/pyspark/pandas/indexes/multi.py` | MultiIndex実装 |
| 3-4 | groupby.py | `python/pyspark/pandas/groupby.py` | GroupBy実装 |

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

```
import pyspark.pandas as ps
    |
    +-- __init__.py [パッケージ初期化]
    |       +-- require_minimum_pandas_version()
    |       +-- require_minimum_pyarrow_version()
    |       +-- _auto_patch_spark()
    |
    +-- ps.DataFrame(data)
    |       +-- frame.py -> DataFrame クラス
    |       |       +-- Internal Spark DataFrame
    |       |       +-- pandas互換メソッド（groupby, merge, apply等）
    |       |       |       +-- Spark SQL変換
    |       |       |       +-- pandas_udf（Python UDF）
    |       |       |
    |       |       +-- to_pandas() -> Arrow -> pandas.DataFrame
    |       |
    |       +-- series.py -> Series クラス
    |       +-- indexes/ -> Index クラス群
    |
    +-- ps.read_csv() / ps.read_parquet() / ps.sql()
            +-- DataFrameReader -> Spark DataFrame -> ps.DataFrame
```

### データフロー図

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

pandas.DataFrame ──▶ Arrow変換 ──▶ Spark DataFrame ──▶ ps.DataFrame
CSV/Parquet ─────▶ SparkReader ──▶ Spark DataFrame ──▶ ps.DataFrame
                                          |
                                    pandas API操作
                                    (Spark SQL変換)
                                          |
                                          v
pandas.DataFrame ◀── Arrow変換 ◀── Spark collect() ◀── to_pandas()
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| __init__.py | `python/pyspark/pandas/__init__.py` | ソース | パッケージ初期化・API定義 |
| frame.py | `python/pyspark/pandas/frame.py` | ソース | DataFrame実装 |
| series.py | `python/pyspark/pandas/series.py` | ソース | Series実装 |
| groupby.py | `python/pyspark/pandas/groupby.py` | ソース | GroupBy実装 |
| base.py (indexes) | `python/pyspark/pandas/indexes/base.py` | ソース | 基本Index実装 |
| multi.py (indexes) | `python/pyspark/pandas/indexes/multi.py` | ソース | MultiIndex実装 |
| datetimes.py (indexes) | `python/pyspark/pandas/indexes/datetimes.py` | ソース | DatetimeIndex実装 |
| config.py | `python/pyspark/pandas/config.py` | ソース | 設定管理 |
| generic.py | `python/pyspark/pandas/generic.py` | ソース | Frame/Series共通基底 |
| internal.py | `python/pyspark/pandas/internal.py` | ソース | 内部Spark DataFrame管理 |
| accessors.py | `python/pyspark/pandas/accessors.py` | ソース | アクセサ（dt, str等） |
| strings.py | `python/pyspark/pandas/strings.py` | ソース | 文字列操作 |
| datetimes.py | `python/pyspark/pandas/datetimes.py` | ソース | 日時操作 |
| missing/ | `python/pyspark/pandas/missing/` | ソース | 未実装APIスタブ |
| spark/ | `python/pyspark/pandas/spark/` | ソース | Spark固有操作 |
| data_type_ops/ | `python/pyspark/pandas/data_type_ops/` | ソース | データ型別演算 |
| correlation.py | `python/pyspark/pandas/correlation.py` | ソース | 相関計算 |
| sql_formatter.py | `python/pyspark/pandas/sql_formatter.py` | ソース | SQLフォーマッタ |
