scikit-learn pipelineによる前処理を概観する

Kaggleのテーブルコンペでsklearnのpipelineを使うと見通しよく前処理ができたので備忘としてまとめておく。

# pipelineの基本

https://scikit-learn.org/stable/modules/compose.html#pipeline

`Pipeline`クラスを使うことで、transformerや（`fit` / `transform`を備えたクラス）predictor（`fit` / `predict`を備えたクラス）などをつなげた機械学習パイプラインを作ることができる。

```python
from sklearn.decomposition import PCA
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

pipeline = Pipeline([("scaler", StandardScaler()), ("pca", PCA())])
X_transformed = pipeline.fit_transform(X)
```

パイプラインとして前処理を実装することには以下の嬉しさがある。

- 複雑な前処理フローをステップの組み合わせとして表現できるため見通しよく実装できる
- 前処理の手続きが一つのインスタンスに集約されるのでpickle化して推論環境で利用しやすい

状態に依存しないstatelessな変換処理を行う場合は関数による実装で事足りるが、学習データに依存した前処理を行いたい場合にはパイプラインを使うメリットが大きい。

以下では、前処理パイプラインの各ステップを構成するtransformerの作り方について見ていく。

# transformerの作成

パイプラインの構成要素であるtransformerを作る方法としては大きく以下の4つがある。

1. 組み込みのtransformerを利用する (ex. `StandardScaler`)
2. 関数からtransformerを作成する：`FunctionTransformer`
3. カスタムtransformerを定義する
4. 複数のtransformerを組み合わせたtransformerを作成する
    - 複数の変換を並列に実行する：`FeatureUnion`
    - カラムごとに別々の変換を施す：`ColumnTransformer`

以下では2-4について説明する。

## 関数からtransformerを作成する

https://scikit-learn.org/stable/modules/preprocessing.html#function-transformer

関数からtransformerインスタンスを生成する。状態を持たない関数を元に作るため、`fit`では実質的には何も行われない。

```python
import numpy as np
from sklearn.preprocessing import FunctionTransformer

log_transformer = FunctionTransformer(np.log1p)
X_transformed = log_transformer.fit_transform(X)
```

## カスタムtransformerを定義する

https://scikit-learn.org/stable/developers/develop.html#rolling-your-own-estimator

`BaseEstimator`と`TransformerMixin`を継承してsklearn準拠のカスタムクラスを定義できる。

```python
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.utils.validation import check_is_fitted, validate_data

class OutlierClipper(TransformerMixin, BaseEstimator):  # Mixinクラスを左に
    """列ごとに外れ値をクリッピングするtransformer"""
    def __init__(self, lower_quantile=0.01, upper_quantile=0.99):  # 1
        self.lower_quantile = lower_quantile  # 2
        self.upper_quantile = upper_quantile

    def fit(self, X, y=None):
        X = validate_data(self, X)
        self.lower_bounds_ = np.quantile(X, self.lower_quantile, axis=0)  # 3
        self.upper_bounds_ = np.quantile(X, self.upper_quantile, axis=0)
        return self

    def transform(self, X):
        check_is_fitted(self)
        X = validate_data(self, X, reset=False)
        X_clipped = np.clip(X, self.lower_bounds_, self.upper_bounds_)
        return X_clipped
        
clipper = OutlierClipper()
X_transformed = clipper.fit_transform(X)
```

クラス定義は以下の作法に従う（コード上の番号に対応）。

1. 入力データに依存しないパラメータは`__init__`のキーワード引数、依存するパラメータは`fit`のキーワード引数として定義する
2. `__init__`に与えたキーワード引数はそのまま同名のパブリック属性として設定する
    - プライベートにしない
    - `__init__`内で値を変更しない
    - 属性名を引数名ｋ変えない
3. `fit`を呼んだ後に決定される属性で外部からのアクセスを許可したいものには、名前の後ろに`_`を付ける (ex. `self.lower_bounds_`, `self.uppder_bounds_`)

なお、定義したカスタムクラスがsklearn準拠になっているかどうかは`check_estimator`によりチェックできる。

```python
from sklearn.utils.estimator_checks import check_estimator

check_estimator(OutlierClipper())
```

:::note info
**[set_output APIへの対応](https://scikit-learn.org/stable/developers/develop.html#developer-api-for-set-output)**

sklearnデフォルトの仕様では`transform`メソッドはNumPy配列を返すが、`set_output`を通してデータフレームを返すようにすることができる。

```python
scaler = StandardScaler().set_output(transform="pandas")
X_transformed = scaler.fit_transform(X)
print(type(X_transformed))
# <class 'pandas.core.frame.DataFrame'>
```

この機能をカスタムクラスでもサポートするためには以下の2つの方法がある。

**マニュアル定義**
出力データのカラム名を`get_feature_name_out`メソッド上で定義する。

**Mixinの利用**
特別なtransformerに対しては以下のような便利なMixinが用意されている。

- `OneToOneFeatureMixin`
    - 入出力のカラムが一対一に対応するような変換に対応するMixin (ex. `StandardScaler`)
    - 出力データのカラム名が入力データのものと同一になる
- `ClassNamePrefixFeaturesOutMixin`
    - 独自の特徴量名が付与される変換に対応するMixin (ex. `PCA`)
    - 出力データのカラム名が`[classname0, classname1, …]`のようになる

例えば上記の外れ値をクリッピングするカスタムクラスは入出力が一対一のため、`OneToOneFeatureMixin`を継承元に加えることで自動的にset_output APIが使えるようになる。

```python
class OutlierClipper(OneToOneFeatureMixin, TransformerMixin, BaseEstimator):
    ....
```
:::

## 複数のtransformerを組み合わせたtransformerを作成する

`Pipeline`は複数のtransformerを直列に繋いでいくものだが、複数のtransformerを並列に繋いで新たなtransformerを作成することもできる。この機能を用いてより複雑なパイプラインを構成することができる。また、並列化することで並列処理の恩恵を受けられるため、パフォーマンスの面でも嬉しい。

### FeatureUnion

https://scikit-learn.org/1.5/modules/compose.html#featureunion-composite-feature-spaces

複数のtransformerを並列に処理して列方向に結合したデータを出力する。

```python
from sklearn.decomposition import PCA, KernelPCA
from sklearn.pipeline import FeatureUnion

feature_union = FeatureUnion([("linear_pca", PCA()), ("kernel_pca", KernelPCA())])
X_transformed = feature_union.fit_transform(X)
```

### ColumnTransformer

https://scikit-learn.org/1.5/modules/compose.html#columntransformer-for-heterogeneous-data

異なるカラムごとに別々のtransformerにより変換を施す。

```python
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder, StandardScaler

column_transformer = ColumnTransformer(
    [
        ("num", StandardScaler(), ["age", "height"]),  # transformerと適用するカラム名を指定
        ("cat", OneHotEncoder(), ["gender"]),
    ]
)
X_transformed = column_transformer.fit_transform(X)
```

以上のような機能を組み合わせることで、込み入った前処理を見通しよく実装することができる。