Pythonのオープンソースの機械学習ライブラリであるscikit-learnにはデータを分析するための便利なAPIが提供されており、さらにそれらを効率よく使用するためのPipelineという機能が用意されています。
本記事では、scikit-learnのAPIとPipelineの基本的な仕組み使い方について記載します。
scikit-learnの機械学習APIのデザイン
まずはscikit-learnの機械学習APIの基本的な概念である「Estimator」「Transformer」「Predictor」の概念について説明します。
Estimator
Estimatorは与えられたデータから「学習」する機能です。学習させるためのメソッドは「fit」となります。
教師あり学習の場合は、教師データのデータ（説明変数）とラベル（目的変数）を入力して学習します。
estimator.fit(X, y)
具体例（ロジスティック回帰）で示すと以下のようになります。
#ロジスティック回帰
from sklearn.linear_model import LogisticRegression
classifier = LogisticRegression()
#学習
classifier.fit(X, y)
教師なし学習、もしくは、データ変換のため統計値算出(例：欠損値補間のために平均値を算出）の場合は、データを入力して学習します。
estimator.fit(X)
具体例（K-means、SimpleImputer）で示すとは以下のようになります。
#Kmeans(教師なし学習）
from sklearn.cluster import KMeans
#クラスタ数は10
km = KMeans(nclusters=10)
#学習
km.fit(X)#欠損値補間
from sklearn.impute import SimpleImputer
imputer = SimpleImputer()
#学習（各特徴量の平均値算出）
imputer.fit(X)Transformer
TransformerはEstimatorを拡張したもので、与えられたデータを変換します。変換するためのメソッドは「transform」となります。
「transform」を実行するためには事前に「fit」が実行されている必要があります。
以下のようにデータを入力値として変換したデータを出力します。
X_transformed = transformer.transform(X)
具体例で示すと以下のようになります。（事前にfitの実行が必要）
#欠損値補間
from sklearn.impute import SimpleImputer
imputer = SimpleImputer()
#教師データで学習（各特徴量の平均値算出）
imputer.fit(X_train)
#欠損値補間(教師データ）
X_train_transformed = imputer.transform(X_train)
#欠損値補間（テストデータ）
X_test_transformed = imputer.transform(X_test)
さらに「fit」と「transform」を一緒に実行するための「fit_transform」のメソッドも持っています。
X_transformed = transformer.fit_transform(X)
具体例で示すと以下のようになります。
#欠損値補間
from sklearn.impute import SimpleImputer
#教師データで学習（各特徴量の平均値算出）
imputer = SimpleImputer()
#欠損値補間
X_transformed = imputer.fit_transform(X)Predictor
PredictorはEstimatorを拡張したもので、与えられたデータから結果を予測します。予測するためのメソッドは「predict」となります。
「predict」を実行するためには事前に「fit」が実行されている必要があります。
以下のようにデータを入力値として予測したデータを出力します。
y_predict = predictor.predict(X)
具体例(LogisticRegression、KMeans)で示すと以下のようになります。（事前にfitの実行が必要）
#ロジスティック回帰
from sklearn.linear_model import LogisticRegression
classifier = LogisticRegression()
#学習
classifier.fit(X_train, y_train)
#予測(教師データ）
y_train_predict = classifier.predict(X_train)
#予測(テストデータ）
y_test_predict = classifier.predict(X_test)#Kmeans(教師なし学習）
from sklearn.cluster import KMeans
#クラスタ数は10
km = KMeans(nclusters=10)
#学習
km.fit(X)
#予測（クラスター分類）
cluster_pred = km.predict(X)Pipelineの仕組みと使用パターン
Pipelineを使用すると複数のEstimatorの処理をまとめて実行する事ができます。以下２つの使用パターンについての仕組みと使用方法を見ていきましょう。
パターン１ TransformerとしてのPipeline
Pipelineを使用して複数のTransformerの処理をまとめて実行する例を記載します。
例として以下のようなPipelineを定義します。
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer
transform_pipeline = Pipeline([
#欠損値を中央値で補完する。
('imputer', SimpleImputer(strategy="median")),
#標準化する。
('std_scaler', StandardScaler())
])
このPipelineでは、最初のステップでSimpleImputerによって入力値に対する欠損値の補完を実行し、2番目のステップでStandardScalerによって標準化を実行する事ができます。
実際の使用するには、Pipelineの「fit_transform」を実行します。
#説明変数
X_train = np.array([[1,np.nan,2, 5],[2, 4, 1,np.nan],[ 3, 2, np.nan,0],[ 1, 1, 7,2]])
#変換
X_train_transformed = transform_pipeline.fit_transform(X_train)
変換前、変換後の値は以下のようになります。
X_train（変換前）の値
[[ 1. nan 2. 5.]
[ 2. 4. 1. nan]
[ 3. 2. nan 0.]
[ 1. 1. 7. 2.]]
X_train_transformed（変換後）の値
[[-0.90453403 -0.22941573 -0.42640143 1.54030809]
[ 0.30151134 1.60591014 -0.85280287 -0.14002801]
[ 1.50755672 -0.22941573 -0.42640143 -1.26025208]
[-0.90453403 -1.14707867 1.70560573 -0.14002801]]「fit_transform」を実行すると、Pipelineの各ステップで「fit」「transform」が以下のように順次実行されます。
- SimpleImputer#fit
→各列の中央値を求める - SimpleImputer#transform
→求めた中央値を使って欠損値を補完する - StandardScaler#fit
→各列の平均と標準偏差を求める - StandardScaler#transform
→求めた平均と標準偏差を使って各値を標準化する
テストデータを変換する場合は「fit_transform」ではなく「transform」を実行します。（事前に教師データを使って「fit」が実行されている必要があります。）
#説明変数
X_test = np.array([[2,0,2, 4],[1, 5, np.nan,3]])
#変換（教師データの統計値を利用)
X_test_transformed = transform_pipeline.transform(X_test)
変換前、変換後の値は以下のようになります。
X_test(変換前）の値
[[ 2. 0. 2. 4.]
[ 1. 5. nan 3.]]
X_test_transformed（変換後)の値
[[ 0.30151134 -2.0647416 -0.42640143 0.98019606]
[-0.90453403 2.52357307 -0.42640143 0.42008403]]
「transform」を実行すると、Pipelineの各ステップで「transform」のみが順次実行されます。
- SimpleImputer#transform
→教師データで求めた中央値を使って欠損値を補完する - StandardScaler#transform
→教師データで求めた平均と標準偏差を使って各値を標準化する
何故、テストデータで「fit_transform」を使用してはいけないのでしょうか？
SimpleImputer、StandardScalerの例では「fit」でテストデータを変換する場合の統計値（中央値、平均値、標準偏差など）を求めています。
これらの統計値は、教師データの値を使用する（学習する）必要があります。（教師データを使用してモデルを作成するので、教師データの分布がモデルのデータ分布となります。）
※そもそもテストデータ毎に統計値が変わってしまったらおかしいですよね？（極端な例で言えば、もしテストデータが１件しかなかったら、テストデータの平均値、標準偏差など意味がありません。）
教師データとテストデータを分割する前に全体のデータで平均値・標準偏差などを求めて標準化などをしてしまう事をついつい”やりがち”ですが、完全にNGです。
これは、教師データの情報に、テストデータの情報が混入している事になります。
あくまでも変換は教師データのデータ分布を使って処理する必要があります。
パターン２ Transformer + PredictorのPipeline
予測するための分類器が絞られてきてパラメータサーチをする場合、分析が終わって最終的なモデルが決定した場合は、データの変換から学習・予測までの全ての処理をPipelineでまとめる事ができます。
例として以下のようなPipelineを定義します。
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
predict_pipeline = Pipeline([
#欠損値を中央値で補完する。
('imputer', SimpleImputer(strategy="median")),
#標準化する。
('std_scaler', StandardScaler()),
#学習・予測する。
('log_reg', LogisticRegression(solver='liblinear')),
])
このPipelineでは、最初のステップでSimpleImputerによって入力値に対する欠損値の補完を実行し、2番目のステップでStandardScalerによって標準化を実行し、3番目のステップでLogisticRegressionによりロジスティック回帰による学習・予測を実行する事ができます。
このPipelineを使って「学習」するためには、Pipelineの「fit」を使用します。
#説明変数
X_train = np.array([[1,np.nan,2, 5],[2, 4, 1,np.nan],[ 3, 2, np.nan,0],[ 1, 1, 7,2]])
#目的変数（正解ラベル）
y_train = [1,0,0,1]
#変換＋学習
predict_pipeline.fit(X_train,y_train)
「fit」を実行すると、Pipelineの1番目、2番目のステップ(SimpleImputer,StandardScaler)では「fit」「transform」が実行され、最後のステップ(LogisticRegression)では「fit」のみが実行されます。
- SimpleImputer#fit
→各列の中央値を求める - SimpleImputer#transform
→求めた中央値を使って欠損値を補完する - StandardScaler#fit
→各列の平均と標準偏差を求める - StandardScaler#transform
→求めた平均と標準偏差を使って各値を標準化する - LogisticRegression#fit
→入力データ（変換されたデータと正解ラベル）を学習する
さらにこのPipelineを使用して「予測」をするためには「predict」を使用します。
教師データとテストデータでの予測を実行する例を以下に示します。
#教師データそのものを予測
y_train_predict = predict_pipeline.predict(X_train)
#テストデータを予測
#説明変数
X_test = np.array([[2,0,2, 4],[1, 5, np.nan,3],[0,1,0,0]])
#目的変数(正解ラベル)
y_test = [1,0,0]
#予測
y_test_predict = predict_pipeline.predict(X_test)
結果は以下のようになります。
教師データ（精度=1.0)
array([1, 0, 0, 1])
テストデータ（精度=0.6666・・・)
array([1, 0, 1])
「predict」を実行すると、Pipelineの1番目、2番目のステップ(SimpleImputer,StandardScaler)では「transform」が実行され、最後のステップ(LogisticRegression)では「predict」が実行されます。
- SimpleImputer#transform
→教師データで求めた中央値を使って欠損値を補完する - StandardScaler#transform
→教師データで求めた平均と標準偏差を使って各値を標準化する - LogisticRegression#predict
→教師データで学習したモデルを使って予測する
「fit」が呼ばれないので、教師データで学習した時(fit実行時）の状態を元に予測する事ができます。
パラメータの設定について
Pipelineの各ステップのEstimatorにはパラメータを設定する事ができます。ただしEstimatorに直接設定する場合と比べて注意が必要です。
上記で定義したpredict_pipline（以下、再掲）の場合、どのように設定するかを以下説明します。
predict_pipeline = Pipeline([
#欠損値を中央値で補完する。
('imputer', SimpleImputer(strategy="median")),
#標準化する。
('std_scaler', StandardScaler()),
#学習・予測する。
('log_reg', LogisticRegression(solver='liblinear')),
])
各ステップのEstimatorにパラメータを設定するためには「set_params」を使って「ステップのキー名__パラメータ名=パラメータ値」を設定します。
predict_pipeline.set_params(
imputer__strategy='mean',
log_reg__max_iter=50)
この例では、SimpleImputerに対してstrategy=’mean’を、LogisticRegressionに対してmax_iter=50を設定しています。
後からパラメータを設定するケースで最も多いのが、グリッドサーチ・ランダムサーチを使ってパラメータサーチをする場合でしょう。
Pipelineを使用しておらず、LogisticRegressionを直接使用する場合は以下のように記載します。
param_grid_cv = [
{'C': [0.1,1,10], 'penalty': ['l1', 'l2']}
]
gscv = GridSearchCV(LogisticRegression(solver='liblinear'),
param_grid_cv, cv=2)
gscv.fit(X_train_transformed,y_train)
一方、Pipelineを使用した場合は以下のようパラメータ名にステップのキー名を付与する必要があります。
param_grid_cv = [
{'log_reg__C': [0.1,1,10], 'log_reg__penalty': ['l1', 'l2']}
]
gscv = GridSearchCV(predict_pipeline, param_grid_cv, cv=2)
gscv.fit(X_train,y_train)
カスタムのEstimatorの作り方
分析をするにあたってカスタムのEstimatorを作成したいケースがあるかと思います。
Transformerを実装する場合、最低限必要な事は以下の４つです。
- BaseEstimatorを継承する。
- TransformerMixinを継承する。
- fitを実装する。
- transformを実装する。
BaseEstimatorを継承するのは「set_params」と「get_params」を使用できるようにするため、TransformerMixinを継承するのは「fit_transform」を使用できるようにするためです。
以下は、列を絞り込むためのTransformerの簡単な実装例です。
from sklearn.base import BaseEstimator, TransformerMixin
class ColSelecter(BaseEstimator, TransformerMixin):
def __init__(self,selector=None):
self.selector = selector
def fit(self, X, y=None):
return self
def transform(self, X):
if self.selector is None:
return X
else:
return X[:, self.selector]
単独でも使用できますが、以下のようにPipelineに組み込む事ができます。
transform_pipeline = Pipeline([
#列を絞り込む
('selector', ColSelecter(selector=[1,3])),
#欠損値を中央値で補完する。
('imputer', SimpleImputer(strategy="median")),
#標準化する。
('std_scaler', StandardScaler())
])
X_train = np.array([[1,np.nan,2, 5],[2, 4, 1,np.nan],[ 3, 2, np.nan,0],[ 1, 1, 7,2]])
X_train_transformed = transform_pipeline.fit_transform(X_train)
実行結果は以下のようになります。
X_train（変換前）の値
[[ 1. nan 2. 5.]
[ 2. 4. 1. nan]
[ 3. 2. nan 0.]
[ 1. 1. 7. 2.]]
X_train_transformed（変換後）の値
[[-0.22941573 1.54030809]
[ 1.60591014 -0.14002801]
[-0.22941573 -1.26025208]
[-1.14707867 -0.14002801]]
Predictorを実装する場合、最低限必要な事は以下の４つです。
- BaseEstimatorを継承する。
- TransformerMixinを継承する。
- fitを実装する。
- predictを実装する。
以下、お遊びですが「0」もしくは「1」をランダムで返すPredictorの実装例です。
class RandomeEstimator(BaseEstimator):
def __init__(self):
pass
def fit(self, X, y=None):
return self
def predict(self, X):
return np.round(np.random.rand(X.shape[0]))
Pipelineに組み込むと以下のようになります。
predict_pipeline = Pipeline([
#欠損値を中央値で補完する。
('imputer', SimpleImputer(strategy="median")),
#標準化する。
('std_scaler', StandardScaler()),
#学習・予測する。
('rnd', RandomeEstimator()),
])
実行結果については、省略します。（適当に0,1返すだけなので・・）
まとめ
以上、scikit-learnのAPIとPipelineの基本的な仕組みと使用方法、カスタムのEstimatorの作成の仕方について説明しました。
本記事で説明した基本的な概念を押さえておけば、使用する場合に「あれ、これってfit使うんだっけ？fit_transform使うんだっけ？」などと悩むことはないでしょう。
また、Pipelineを使用すれば、分析する上で最も泥臭くかつ重要な「前処理」をある程度、自動化する事ができます。
使用した事がない方は是非、試してみてください。
