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

pipelineの基本

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

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つがある。

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

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

関数からtransformerを作成する

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

import numpy as np
from sklearn.preprocessing import FunctionTransformer

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

カスタムtransformerを定義する

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

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)

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

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

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

from sklearn.utils.estimator_checks import check_estimator

check_estimator(OutlierClipper())

set_output APIへの対応

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

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が使えるようになる。

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

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

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

FeatureUnion

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

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

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

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)

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