# 機能設計書 89-PySpark API

## 概要

本ドキュメントは、PythonからSparkの全機能（RDD、DataFrame、SQL、MLlib、Streaming）にアクセスするためのPySpark APIの機能設計を記述する。

### 本機能の処理概要

PySparkは、PythonプログラミングからApache Sparkの分散データ処理機能にアクセスするためのPython APIである。SparkContext、RDD、DataFrame/Dataset、SQL、MLlib、Streaming等のSpark全機能をPythonから利用可能にし、Py4J（Javaゲートウェイ）またはSpark Connect経由でJVM上のSparkエンジンと通信する。

**業務上の目的・背景**：Pythonはデータサイエンス・機械学習分野で最も広く使用されるプログラミング言語である。PySparkにより、Pythonエコシステム（NumPy、pandas、scikit-learn等）との親和性を保ちながら、Sparkの大規模分散データ処理能力を活用できる。データサイエンティストが使い慣れたPython環境からビッグデータ処理を実行でき、プロトタイプから本番環境への移行を容易にする。

**機能の利用シーン**：Jupyter Notebookでのインタラクティブデータ探索、機械学習パイプラインの構築と実行、大規模ETLジョブの開発、Pythonスクリプトによるバッチ処理。

**主要な処理内容**：
1. SparkContext/SparkSession提供：PythonからSparkクラスタへの接続エントリーポイント
2. RDD API：分散コレクション操作のPython API（map, filter, reduce等）
3. DataFrame/SQL API：構造化データ操作とSQL実行のPython API
4. MLlib API：機械学習アルゴリズムのPython API
5. Streaming API：ストリーム処理のPython API
6. UDF：Python関数をSpark UDFとして登録・実行（pandas UDF含む）
7. Py4J/Spark Connect通信：Python-JVM間の透過的な通信

**関連システム・外部連携**：JVM（Py4J経由）、Spark Connect Server（gRPC経由）、Apache Arrow（データ転送）、pandas、NumPy

**権限による制御**：Sparkの認証・認可機構をPython API経由で利用

## 関連画面

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

## 機能種別

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

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| master | str | No | Sparkクラスタマスター URL | - |
| appName | str | No | アプリケーション名 | - |
| sparkHome | str | No | SPARKインストールディレクトリ | - |
| conf | SparkConf | No | Spark設定 | - |
| remote | str | No | Spark Connect接続文字列 | sc://形式 |

### 入力データソース

Pythonアプリケーションコード、Jupyter Notebook、pysparkシェル

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| SparkContext | pyspark.SparkContext | Sparkクラスタ接続オブジェクト |
| SparkSession | pyspark.sql.SparkSession | SQLセッションオブジェクト |
| RDD | pyspark.RDD | 分散データセット |
| DataFrame | pyspark.sql.DataFrame | 構造化データフレーム |

### 出力先

Pythonアプリケーション（メモリ上のデータ構造）

## 処理フロー

### 処理シーケンス

```
1. SparkSession/SparkContext生成
   └─ SparkSession.builder.getOrCreate() / SparkContext()
2. Py4Jゲートウェイ起動（通常モード）
   └─ JavaGateway経由でJVM上のSparkSessionと接続
3. Spark Connect接続（リモートモード）
   └─ gRPC経由でSpark Connectサーバーに接続
4. DataFrame/RDD操作
   └─ Python APIがJVM APIにプロキシ
5. UDF実行
   └─ Python関数をPickle/Arrow経由でJVMに転送・実行
6. 結果取得
   └─ collect()等でJVM→Pythonにデータ転送（Arrow最適化）
```

### フローチャート

```mermaid
flowchart TD
    A[Python Application] --> B{接続モード}
    B -->|Classic| C[Py4J JavaGateway]
    B -->|Connect| D[Spark Connect gRPC]
    C --> E[JVM SparkSession]
    D --> F[Remote SparkSession]
    E --> G[DataFrame/RDD操作]
    F --> G
    G --> H{UDF?}
    H -->|Yes| I[Python Worker Process]
    H -->|No| J[JVM内実行]
    I --> K[結果マージ]
    J --> K
    K --> L[Python結果返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-89-01 | リモートモード判定 | is_remote_only()でSpark Connect専用モードを判定し、利用可能なAPIを制限する | モジュール初期化時 |
| BR-89-02 | 後方互換モジュール | pyspark.rdd等の旧モジュールパスをsys.modulesで後方互換パスとして登録する | インポート時 |
| BR-89-03 | pandas最小バージョン | require_minimum_pandas_version()でpandasの最小バージョン要件を検証する | pandas連携機能使用時 |
| BR-89-04 | Arrow最小バージョン | require_minimum_pyarrow_version()でPyArrowの最小バージョン要件を検証する | Arrow連携機能使用時 |

### 計算ロジック

特になし（各種計算はJVM側のSpark SQLエンジンで実行される）。

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

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

本機能自体はデータベース操作を直接行わない。DataFrame APIを通じて間接的にデータソースアクセスが行われる。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ImportError | 依存パッケージ不足 | pandas/pyarrow等が未インストールの場合 | pip installで導入 |
| Py4JJavaError | JVM例外 | JVM側で例外が発生した場合 | Javaスタックトレースを確認 |
| ConnectionError | 接続エラー | Spark Connectサーバーに接続できない場合 | 接続設定を確認 |

### リトライ仕様

Spark Connect経由の場合、クライアントのリトライポリシーが適用される。Py4J経由の場合は直接的なリトライ機構はない。

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

Python API層でのトランザクション管理は行わない。Spark SQLエンジン側のトランザクション制御に依存する。

## パフォーマンス要件

- Apache Arrowによる効率的なPython-JVM間データ転送
- pandas UDFによるベクトル化された関数実行
- Spark Connect経由ではgRPC + Protobufによる軽量通信

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

- Spark ConnectモードではgRPCを介したセキュア通信
- Py4Jモードではローカルソケットを使用（ネットワーク露出なし）

## 備考

PySparkはSpark 0.7以降で提供されている。Spark 3.4以降ではSpark Connect経由のリモート接続もサポートされ、is_remote_only()による条件付きモジュール読み込みが導入された。

---

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

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

### 推奨読解順序

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

PySparkの全体構造とモジュール構成を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | __init__.py | `python/pyspark/__init__.py` | **18-76行目**: PySparkのパブリッククラス一覧、is_remote_only()による条件付きインポート、後方互換モジュールパスの登録 |

**読解のコツ**: PySparkはPythonパッケージであり、__init__.pyがモジュール公開APIを定義する。`is_remote_only()`がTrue場合、RDD等のクラシックAPIは利用不可となる。

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

SparkSession/SparkContextの生成を追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | conf.py | `python/pyspark/conf.py` | SparkConf - Spark設定のPython API |
| 2-2 | java_gateway.py | `python/pyspark/java_gateway.py` | Py4J JavaGatewayの起動と管理 |

**主要処理フロー**:
1. **49-55行目** (__init__.py): `is_remote_only()`チェック
2. **55-66行目**: Classic APIのインポートとsys.modules登録

#### Step 3: 各サブモジュールを理解する

主要なサブモジュール構成を追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | python/pyspark/sql/ | `python/pyspark/sql/` | DataFrame/SQL API |
| 3-2 | python/pyspark/ml/ | `python/pyspark/ml/` | MLlib API |
| 3-3 | python/pyspark/streaming/ | `python/pyspark/streaming/` | Streaming API |
| 3-4 | python/pyspark/core/ | `python/pyspark/core/` | RDD等のコアAPI |

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

```
Python Application
    |
    +-- pyspark.__init__.py [モジュール初期化]
    |       +-- is_remote_only() [モード判定]
    |       +-- SparkConf / SparkContext / SparkSession
    |
    +-- Classic Mode (Py4J)
    |       +-- java_gateway.py -> Py4J Gateway
    |       +-- JVM SparkSession / SparkContext
    |       +-- Python Worker (UDF実行)
    |
    +-- Connect Mode (gRPC)
            +-- pyspark.sql.connect.session
            +-- gRPC -> Spark Connect Server
            +-- Arrow データ転送
```

### データフロー図

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

Python Code ─────▶ PySpark API ──▶ Py4J/gRPC ──▶ JVM Spark Engine
                                                        |
Python結果 ◀── Arrow/Pickle ◀── Py4J/gRPC ◀── 実行結果
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| __init__.py | `python/pyspark/__init__.py` | ソース | PySparkパッケージ初期化 |
| conf.py | `python/pyspark/conf.py` | ソース | SparkConf実装 |
| java_gateway.py | `python/pyspark/java_gateway.py` | ソース | Py4Jゲートウェイ管理 |
| serializers.py | `python/pyspark/serializers.py` | ソース | シリアライゼーション |
| shell.py | `python/pyspark/shell.py` | ソース | pysparkシェル |
| daemon.py | `python/pyspark/daemon.py` | ソース | Python Workerデーモン |
| storagelevel.py | `python/pyspark/storagelevel.py` | ソース | StorageLevel定義 |
| accumulators.py | `python/pyspark/accumulators.py` | ソース | アキュムレータ実装 |
| taskcontext.py | `python/pyspark/taskcontext.py` | ソース | TaskContext実装 |
| profiler.py | `python/pyspark/profiler.py` | ソース | プロファイラ |
