PipelineとColumnTransformerを理解する

これまでTitanicデータセットを使い、NumPy、pandas、統計、データ分析、ロジスティック回帰、決定木、Random Forest、特徴量エンジニアリングなどを一つずつ学んできた。

今回は、これまで個別に扱ってきた前処理を、機械学習モデルの学習処理に組み込む方法を学んだ。

今回の中心となるのは、scikit-learnのPipelineとColumnTransformerである。

今回の学習目標

今回の目標は、Titanicのデータを使って、次のような処理を一つのモデルとして構成することである。

X
↓
ColumnTransformer
├─ 数値列
│   ├─ 欠損値補完
│   └─ 標準化
│
└─ カテゴリ列
    ├─ 欠損値補完
    └─ One-Hot Encoding
↓
LogisticRegression

単にモデルを動かすだけではなく、「なぜこの処理で動くのか」を理解することを今回の学習テーマとした。

これまでのPipeline

これまで、ロジスティック回帰にStandardScalerを組み合わせたPipelineを使っていた。

model = Pipeline([
    ("scaler", StandardScaler()),
    ("classifier", LogisticRegression())
])

model.fit(X_train, y_train)

このコードで最初に疑問に感じたのが、model.fit()と書いただけなのに、なぜStandardScalerのfit_transform()まで実行されるのかという点である。

Pipelineは、複数の処理を順番につないで実行する仕組みである。

概念的には、次のように動く。

X_train
↓
StandardScaler.fit_transform()
↓
LogisticRegression.fit()

予測時は逆に、前処理についてはtransform()だけが実行される。

X_test
↓
StandardScaler.transform()
↓
LogisticRegression.predict()

ここで重要なのは、テストデータについて新しくScalerを学習させないことである。

StandardScalerなら訓練データから平均と標準偏差を覚え、その値を使ってテストデータを変換する。

Pipelineの「最後」は何でも同じではない

今回、Pipelineについてもう一つ疑問が生じた。

Pipelineの途中ではfit_transform()を実行し、最後ではfit()を実行する、と説明されることがある。しかし、これは正確には「最後だからfitだけ」という意味ではない。

重要なのは、最後のステップが何であるかである。

例えば、次のPipelineを考える。

Pipeline([
    ("imputer", SimpleImputer()),
    ("scaler", StandardScaler())
])

このPipelineの最後はStandardScalerである。StandardScalerはTransformerなので、Pipeline全体をfit()すると、概念的には次の処理になる。

SimpleImputer.fit_transform()
↓
StandardScaler.fit_transform()

一方、次のPipelineでは最後がLogisticRegressionである。

Pipeline([
    ("scaler", StandardScaler()),
    ("classifier", LogisticRegression())
])

この場合は、

StandardScaler.fit_transform()
↓
LogisticRegression.fit()

となる。

つまり、正確には次のように理解する必要がある。

Transformerとは何か

ここで「Transformerとは何なのか」という疑問が出てきた。

Transformerは、データを別の形に変換するためのオブジェクトである。

代表的なものには次のようなものがある。

• SimpleImputer：欠損値を補完する
• StandardScaler：データを標準化する
• OneHotEncoder：カテゴリデータを数値に変換する
• PCA：次元削減を行う

これらは基本的に、fit()で訓練データから変換に必要な情報を学習し、transform()でデータを変換する。

例えばSimpleImputerで中央値補完をする場合、

imputer = SimpleImputer(strategy="median")

とした後、訓練データに対してfit()すると、訓練データから中央値を計算して覚える。

その後、transform()すると、覚えた中央値を使って欠損値を補完する。

ここでも、テストデータから中央値を計算してはいけない。

ColumnTransformerを使う理由

今回のTitanicデータでは、すべての列を同じ方法で処理することはできない。

例えば、次のようなデータがある。

Pclass       数値
Age          数値
Fare         数値
FamilySize   数値
Sex          カテゴリ
Embarked     カテゴリ

数値列にはStandardScalerを使いたい。

一方、SexやEmbarkedのようなカテゴリ列にはOneHotEncoderを使いたい。

このように、列によって前処理を変えたい場合に使うのがColumnTransformerである。

数値列用Pipelineを作る

まず数値列についてのPipelineを作った。

numeric_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])

これは、

数値データ
↓
欠損値を中央値で補完
↓
標準化

という処理を一つのPipelineにまとめたものである。

今回の数値列は次の4つとした。

numeric_features = [
    "Pclass",
    "Age",
    "Fare",
    "FamilySize"
]

カテゴリ列用Pipelineを作る

次にカテゴリ列用のPipelineを作った。

categorical_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("encoder", OneHotEncoder(handle_unknown="ignore"))
])

こちらは、

カテゴリデータ
↓
欠損値を最頻値で補完
↓
One-Hot Encoding

という処理になる。

カテゴリ列には次の2つを使用した。

categorical_features = [
    "Sex",
    "Embarked"
]

OneHotEncoderを使う理由

カテゴリデータを単純に整数へ変換してはいけない場合がある。

例えばEmbarkedを、

S → 0
C → 1
Q → 2

と変換したとする。

すると、機械学習モデルから見ると「QはCより大きく、CはSより大きい」という数値的な関係が存在するように見えてしまう。

しかし、S、C、Qは単なるカテゴリであり、本来そこに大小関係はない。

そこでOneHotEncoderを使う。

S → [1, 0, 0]
C → [0, 1, 0]
Q → [0, 0, 1]

これならカテゴリ間に大小関係を持たせずに、機械学習モデルへ渡せる。

ColumnTransformerで処理を分岐する

数値用Pipelineとカテゴリ用Pipelineができたので、ColumnTransformerで組み合わせる。

preprocessor = ColumnTransformer([
    ("num", numeric_pipeline, numeric_features),
    ("cat", categorical_pipeline, categorical_features)
])

このコードは、かなり重要な構造を持っている。

X
│
├── 数値列
│     ↓
│  numeric_pipeline
│     ↓
│  Imputer
│     ↓
│  StandardScaler
│
└── カテゴリ列
      ↓
   categorical_pipeline
      ↓
   Imputer
      ↓
   OneHotEncoder
│
└── 処理結果を結合

つまりColumnTransformerは、列ごとに別々の処理を適用し、その結果を一つにまとめる役割を持っている。

最後にモデルと組み合わせる

最後に、ColumnTransformerそのものをPipelineに組み込んだ。

model = Pipeline([
    ("preprocessor", preprocessor),
    ("classifier", LogisticRegression())
])

これで、モデル全体が一つのPipelineになった。

学習時には概念的に次のように動く。

model.fit(X_train, y_train)
        ↓
ColumnTransformer
        │
        ├── 数値列
        │     ↓
        │  SimpleImputer.fit_transform()
        │     ↓
        │  StandardScaler.fit_transform()
        │
        └── カテゴリ列
              ↓
           SimpleImputer.fit_transform()
              ↓
           OneHotEncoder.fit_transform()
        ↓
結果を結合
        ↓
LogisticRegression.fit()

予測時は、

model.predict(X_test)
        ↓
ColumnTransformer.transform()
        ↓
数値列・カテゴリ列をそれぞれ変換
        ↓
LogisticRegression.predict()

となる。

実際にTitanicで実装した

最終的なコードは次のようになった。

import pandas as pd

from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.preprocessing import OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer
from sklearn.metrics import accuracy_score

df = pd.read_csv("train.csv")

df["FamilySize"] = df["SibSp"] + df["Parch"] + 1

features = [
    "Pclass",
    "Sex",
    "Age",
    "Fare",
    "FamilySize",
    "Embarked"
]

X = df[features]
y = df["Survived"]

X_train, X_test, y_train, y_test = train_test_split(
    X,
    y,
    test_size=0.2,
    random_state=42
)

numeric_features = [
    "Pclass",
    "Age",
    "Fare",
    "FamilySize"
]

categorical_features = [
    "Sex",
    "Embarked"
]

numeric_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])

categorical_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("encoder", OneHotEncoder(handle_unknown="ignore"))
])

preprocessor = ColumnTransformer([
    ("num", numeric_pipeline, numeric_features),
    ("cat", categorical_pipeline, categorical_features)
])

model = Pipeline([
    ("preprocessor", preprocessor),
    ("classifier", LogisticRegression())
])

model.fit(X_train, y_train)

y_pred = model.predict(X_test)

print(accuracy_score(y_test, y_pred))

実行結果

0.8044692737430168

これまで作ってきたモデルと同じように、テストデータに対して約80.4%のAccuracyとなった。

今回はスコアを大幅に上げることが目的ではない。これまで手作業で行っていた欠損値処理やスケーリング、カテゴリ変換を、機械学習の学習処理の中に正しく組み込めたことが重要である。

途中で特に混乱したところ

Pipelineの中にPipelineを入れている

今回、少し分かりにくかったのが、Pipelineの中に別のPipelineが存在する構造である。

model
└── preprocessor
    └── ColumnTransformer
        ├── numeric_pipeline
        │   ├── SimpleImputer
        │   └── StandardScaler
        │
        └── categorical_pipeline
            ├── SimpleImputer
            └── OneHotEncoder

最初は「PipelineとはScalerとモデルを組み合わせるもの」というイメージだったが、実際にはもっと汎用的な仕組みである。

処理を順番につなげるための仕組みと考えると理解しやすい。

Transformerと最終モデルの違い

もう一つ疑問になったのが、「Pipelineはどうやって最後のモデルと前処理を区別しているのか」という点である。

SimpleImputerやStandardScalerなどはTransformerであり、データを変換するための処理である。

一方、LogisticRegressionは学習データから分類モデルを作るEstimatorである。

実際にPythonで型を確認すると、

print(type(SimpleImputer()))
print(type(LogisticRegression()))

の結果は、

<class 'sklearn.impute._base.SimpleImputer'>
<class 'sklearn.linear_model._logistic.LogisticRegression'>

となった。

ただし、type()を見ただけでTransformerかどうかを判断しているわけではない。

scikit-learnでは、TransformerやClassifierなどの共通インターフェースが用意されている。

例えば、次のように確認できる。

from sklearn.base import TransformerMixin, ClassifierMixin

print(isinstance(SimpleImputer(), TransformerMixin))
print(isinstance(LogisticRegression(), TransformerMixin))

print(isinstance(SimpleImputer(), ClassifierMixin))
print(isinstance(LogisticRegression(), ClassifierMixin))

概念的には、次のような関係になる。

SimpleImputer
    ↓
Transformer

StandardScaler
    ↓
Transformer

LogisticRegression
    ↓
Classifier

したがって、今回のPipelineは、

Transformer
↓
Transformer
↓
最終モデル

という構造になっている。

今回の学習で理解できたこと

• Pipelineは前処理やモデルを順番につなぐ仕組みである
• Pipelineの途中にはTransformerを置く
• Transformerはfit()で変換に必要な情報を学習し、transform()でデータを変換する
• Pipeline全体をfit()すると、途中のTransformerは基本的にfit_transform()される
• 予測時はテストデータに対してtransform()だけを行う
• ColumnTransformerを使うと、列の種類によって異なる前処理を適用できる
• SimpleImputerで欠損値を補完できる
• OneHotEncoderでカテゴリデータを適切に数値化できる
• 前処理をPipelineに含めることで、データリークを防ぎやすくなる

今回の学習を振り返って

今回の内容は、コードそのものはそれほど長くないが、内部で何が起きているのかを理解するのに時間がかかった。

特に、

Pipeline
  ↓
ColumnTransformer
  ↓
Pipeline
  ↓
Transformer

という構造は、最初はかなり複雑に感じた。

ただ、処理を一つずつ追っていくと、それぞれが特別なことをしているわけではない。

「データを変換する処理を順番につなぐ」「列によって処理を分ける」「最後にモデルへ渡す」という役割分担になっている。

今回も、分からない部分についてはChatGPTを講師兼壁打ち相手として使い、「この理解で合っているか」「この処理では実際に何が呼ばれているのか」といった点を確認しながら進めた。

単にコードを生成してもらうのではなく、自分でコードを書いて実行し、出力を確認した上で疑問点を質問する、という使い方を意識している。

機械学習についてはまだ初学者の段階だが、少しずつ「コードを動かせる」状態から「なぜこのコードで動くのか説明できる」状態へ進めていきたい。

次回に向けて

次は、これまで扱ってきたTitanicの特徴量エンジニアリングを、今回学んだPipelineの中に組み込む方法を考えていきたい。

特に、Name列からTitleを抽出するような処理を、学習データとテストデータに対して一貫して適用する方法を理解することが次の課題になる。

Kaggleで上位を目指すためには、単にモデルを選ぶだけではなく、データからどのような特徴を作り出すかが重要になる。

今後は、PipelineやColumnTransformerを土台として、より実践的な特徴量エンジニアリングやモデル比較へ進んでいく予定である。