パイプライン

Machine Learning with PySpark

Andrew Collier

Data Scientist, Fathom Data

リーケージ？

fit() メソッド

学習データのみ。

transform() メソッド

学習・テスト両方に使用。

リーキーなモデル

テストデータを学習に使用しているモデル

水漏れのないモデル

学習データのみで学習しているモデル

パイプライン

パイプラインは一連の処理から成ります。

複数ステージのパイプライン

各処理を個別に適用してもよいですが、パイプラインなら一括で適用できます。

車モデル：手順

indexer = StringIndexer(inputCol='type', outputCol='type_idx')

onehot = OneHotEncoder(inputCols=['type_idx'], outputCols=['type_dummy'])

assemble = VectorAssembler(
  inputCols=['mass', 'cyl', 'type_dummy'],
  outputCol='features'
)

regression = LinearRegression(labelCol='consumption')

車モデル：手順の適用

学習データ

indexer = indexer.fit(cars_train)
cars_train = indexer.transform(cars_train)

onehot = onehot.fit(cars_train)
cars_train = onehot.transform(cars_train)

cars_train = assemble.transform(cars_train)

# 学習データでモデルを学習
regression = regression.fit(cars_train)

テストデータ

cars_test  = indexer.transform(cars_test)

cars_test  = onehot.transform(cars_test)

cars_test  = assemble.transform(cars_test)

# テストデータで予測
predictions = regression.transform(cars_test)

車モデル：パイプライン

手順をパイプラインに統合します。

from pyspark.ml import Pipeline

pipeline = Pipeline(stages=[indexer, onehot, assemble, regression])

学習データ

pipeline = pipeline.fit(cars_train)

テストデータ

predictions = pipeline.transform(cars_test)

車モデル：ステージ

.stages 属性で各ステージにアクセスします。

# 線形回帰オブジェクト（第4ステージ → インデックス3）
pipeline.stages[3]


print(pipeline.stages[3].intercept)

4.19433571782916

print(pipeline.stages[3].coefficients)

DenseVector([0.0028, 0.2705, -1.1813, -1.3696, -1.1751, -1.1553, -1.8894])

パイプラインで作業効率化！

Machine Learning with PySpark