# 機能設計書 42-Pipeline API

## 概要

本ドキュメントは、Apache Spark MLlibにおけるPipeline API機能の設計を記述する。機械学習ワークフローをパイプラインとして構成し、前処理からモデル訓練・評価までの一連のステージを統合管理する仕組みについて詳細に説明する。

### 本機能の処理概要

Pipeline APIは、機械学習ワークフローを複数のステージ（Estimator/Transformer）の連鎖として定義・実行するためのフレームワークである。

**業務上の目的・背景**：機械学習プロジェクトでは、特徴量エンジニアリング、モデル訓練、予測という複数のステップを一貫したワークフローとして管理する必要がある。Pipeline APIは、これらのステップを標準化されたインターフェースで統合し、再現性の高い機械学習パイプラインの構築を可能にする。

**機能の利用シーン**：データ前処理（トークナイゼーション、特徴量変換）からモデル訓練、ハイパーパラメータチューニング（CrossValidator/TrainValidationSplit）まで、一連のML処理を構成する際に使用される。

**主要な処理内容**：
1. PipelineStage（Estimator/Transformer）の順序付き配列としてパイプラインを定義
2. Pipeline.fit()で入力データセットに対してステージを順次実行し、PipelineModelを生成
3. PipelineModel.transform()でフィット済みモデルによるデータ変換を実行
4. パイプラインとモデルのシリアライズ/デシリアライズ（永続化・復元）
5. スキーマ検証による入出力整合性の事前チェック

**関連システム・外部連携**：MLlib内の各種Estimator（分類器・回帰器等）およびTransformer（特徴量変換器等）と連携する。

**権限による制御**：特段のロール制御はない。

## 関連画面

本機能に直接関連する画面はない。

## 機能種別

計算処理 / 機械学習ワークフロー管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| stages | Array[PipelineStage] | Yes | パイプラインのステージ配列 | 重複ステージ不可、各ステージはEstimatorまたはTransformerであること |
| dataset | Dataset[_] | Yes | fit()メソッドの入力データセット | スキーマがステージの要件を満たすこと |

### 入力データソース

MLlibのDataFrame/Datasetとして提供されるトレーニングデータ。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| PipelineModel | PipelineModel | フィット済みのパイプラインモデル（Transformer配列を内包） |
| transformed DataFrame | DataFrame | PipelineModel.transform()による変換結果 |

### 出力先

メモリ内のPipelineModelオブジェクト、またはMLWriter経由でファイルシステムに永続化。

## 処理フロー

### 処理シーケンス

```
1. Pipeline.fit(dataset)
   └─ transformSchema()でスキーマ検証
   └─ 最後のEstimatorのインデックスを探索
   └─ ステージを順次実行:
       ├─ Estimatorの場合: estimator.fit(curDataset)でモデル生成、次ステージ用にtransform
       └─ Transformerの場合: そのまま次ステージへ
   └─ PipelineModelを生成して返却

2. PipelineModel.transform(dataset)
   └─ transformSchema()でスキーマ検証
   └─ stages.foldLeft(dataset)で各Transformerを順次適用
   └─ 変換済みDataFrameを返却

3. Pipeline/PipelineModel永続化
   └─ メタデータ保存（UID、ステージUID配列）
   └─ 各ステージを個別にシリアライズ保存
```

### フローチャート

```mermaid
flowchart TD
    A[Pipeline.fit開始] --> B[transformSchema検証]
    B --> C[最後のEstimatorインデックス探索]
    C --> D{ステージループ}
    D --> E{Estimator?}
    E -->|Yes| F[estimator.fit でモデル生成]
    F --> G{最後のEstimator前?}
    G -->|Yes| H[transformer.transform でデータ変換]
    G -->|No| I[transformersに追加]
    H --> I
    E -->|No| J[Transformerとしてそのまま追加]
    J --> I
    I --> D
    D -->|完了| K[PipelineModel生成]
    K --> L[返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-42-01 | 重複ステージ禁止 | パイプライン内に同一ステージインスタンスを含めることはできない | transformSchema時 |
| BR-42-02 | 最後のEstimator以降はTransformのみ | 最後のEstimator以降のステージはTransformerのみ許可 | fit()実行時 |
| BR-42-03 | ステージ数最低0 | ステージ数0のパイプラインはidentity transformerとして動作 | 常時 |
| BR-42-04 | MLWritable要件 | Pipeline永続化時は全ステージがMLWritableを実装している必要がある | write時 |

### 計算ロジック

パイプラインのfit()では、`indexOfLastEstimator`以前のステージに対してのみ`transform()`が呼び出される（最後のEstimatorステージ自体のtransformは省略される）。

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

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

本機能はデータベースへの直接操作は行わない。パイプライン永続化はファイルシステム（HDFS/ローカル）を使用する。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalArgumentException | バリデーションエラー | パイプライン内に重複ステージが存在 | ステージインスタンスを重複なくする |
| IllegalArgumentException | バリデーションエラー | EstimatorでもTransformerでもないステージが含まれる | 正しいPipelineStage実装を使用 |
| UnsupportedOperationException | 永続化エラー | MLWritableを実装していないステージが含まれる | 全ステージにMLWritableを実装 |

### リトライ仕様

特になし。

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

パイプラインのfit()は逐次処理であり、途中のステージで例外が発生した場合はそこまでの処理結果は破棄される。

## パフォーマンス要件

パイプラインの実行時間は各ステージのfit()/transform()処理時間の合計に依存する。

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

パイプラインモデルの永続化時に、モデルパラメータが平文でファイルシステムに保存されるため、機密データを含むモデルの場合はアクセス制御を適用する必要がある。

## 備考

- Pipeline APIはscikit-learnのPipeline概念を参考に設計されている
- CrossValidatorやTrainValidationSplitと組み合わせてハイパーパラメータチューニングが可能

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Pipeline.scala | `mllib/src/main/scala/org/apache/spark/ml/Pipeline.scala` | PipelineStage抽象クラス（42-80行目）、Estimator/Transformer/Modelの関係性を理解する |

**読解のコツ**: SparkのML Pipelineは`PipelineStage`を基底とし、`Estimator[M]`（fit→Model）と`Transformer`（transform→DataFrame）の2種類のステージで構成される。`Params`トレイトによりパラメータ管理が統一されている。

#### Step 2: Pipelineクラスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | Pipeline.scala | `mllib/src/main/scala/org/apache/spark/ml/Pipeline.scala` | Pipeline.fit()メソッド（133-170行目）の逐次処理ロジックを理解する |

**主要処理フロー**:
1. **105行目**: `stages`パラメータ定義（PipelineStage配列）
2. **133-170行目**: `fit()` - スキーマ検証→最後のEstimator探索→ステージ逐次実行→PipelineModel生成
3. **139-145行目**: `indexOfLastEstimator`の探索ループ
4. **148-167行目**: ステージ実行ループ（Estimator.fit/Transformer.transform分岐）
5. **180-185行目**: `transformSchema()` - 全ステージのスキーマを順次検証

#### Step 3: PipelineModelを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Pipeline.scala | `mllib/src/main/scala/org/apache/spark/ml/Pipeline.scala` | PipelineModel.transform()（331-336行目）のfoldLeft処理を理解する |

**主要処理フロー**:
- **320-323行目**: PipelineModel定義（uid + Transformer配列）
- **331-336行目**: `transform()` - stages.foldLeftでTransformerを順次適用
- **339-341行目**: `transformSchema()` - 全ステージのスキーマ変換

#### Step 4: 永続化を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | Pipeline.scala | `mllib/src/main/scala/org/apache/spark/ml/Pipeline.scala` | SharedReadWrite（225-313行目）による保存・読み込みロジックを理解する |

**主要処理フロー**:
- **230-238行目**: `validateStages()` - 全ステージのMLWritable検証
- **257-273行目**: `saveImpl()` - メタデータ保存 + ステージ個別保存
- **289-304行目**: `load()` - メタデータ読み込み + ステージ個別復元

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

```
Pipeline.fit(dataset)
    |
    +-- transformSchema(dataset.schema)
    |       +-- stage1.transformSchema()
    |       +-- stage2.transformSchema()
    |       +-- ...
    |
    +-- [ステージループ]
    |       +-- Estimator.fit(curDataset) -> Transformer(Model)
    |       +-- Transformer.transform(curDataset) -> next curDataset
    |
    +-- new PipelineModel(uid, transformers)

PipelineModel.transform(dataset)
    |
    +-- transformSchema(dataset.schema)
    +-- stages.foldLeft(dataset)(transformer.transform)
```

### データフロー図

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

Dataset ──────> Pipeline.fit()
                    |
                    +-- Stage1(Transformer).transform()
                    |       |
                    |       v
                    +-- Stage2(Estimator).fit() -> Model2
                    |       |
                    |       v
                    +-- Model2.transform()
                    |       |
                    |       v
                    +-- Stage3(Estimator).fit() -> Model3
                    |
                    v
               PipelineModel ──────> 永続化(HDFS/ローカル)
                    |
                    v
               PipelineModel.transform(newData) -> DataFrame
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Pipeline.scala | `mllib/src/main/scala/org/apache/spark/ml/Pipeline.scala` | ソース | Pipeline/PipelineModel/PipelineStageの定義 |
| Estimator.scala | `mllib/src/main/scala/org/apache/spark/ml/Estimator.scala` | ソース | Estimator基底クラス（fit→Model） |
| Transformer.scala | `mllib/src/main/scala/org/apache/spark/ml/Transformer.scala` | ソース | Transformer基底クラス（transform→DataFrame） |
| Model.scala | `mllib/src/main/scala/org/apache/spark/ml/Model.scala` | ソース | Model基底クラス（TransformerかつHasParent） |
| Params.scala | `mllib/src/main/scala/org/apache/spark/ml/param/Params.scala` | ソース | パラメータ管理トレイト |
| MLWritable.scala | `mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala` | ソース | 永続化インターフェース |
