scikit-learn と SparkML パイプラインの使用

多くの機械学習ライブラリ、たとえば scikit-learn や SparkML などでは、一連の変換をカプセル化する "Pipeline" の概念が提供されています。

foundry_ml のこれらのパッケージ向けのネイティブラッパーは、パイプラインオブジェクトの保存をサポートしていますが、パイプラインを個々の変換に展開し、これらを Stage オブジェクトにラッピングし、再度 Model として組み合わせることをお勧めします。これにより、モデル間でのステージの再利用が容易になり、他のパッケージからの実装と個々のステージを容易に交換できるようになり、モデルのプレビューを表示する際に追加の機能と透明性が得られます。

以下に、scikit-learn および SparkML に対してこの展開-ラッピング-再結合の手順を実行する方法の例を示します。

両方の例では、Code Workbook の構文と Modeling Objectives チュートリアル の住宅データを使用しています。これらは、訓練された ML パイプラインの構築、Model への変換、およびステージ間でデータを渡すための関連する入力行と出力行をキャプチャするパラメーターを示しています。

scikit-learn Pipeline を使用した例

この例では、scikit-learn Pipeline を保存する方法を示しています。output_column_name パラメーターの明示的な使用に注意してください。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def sklearn_pipeline_model(housing):

    from sklearn.linear_model import LinearRegression
    from sklearn.preprocessing import PolynomialFeatures
    from sklearn.preprocessing import StandardScaler
    from sklearn.compose import make_column_transformer
    from sklearn.pipeline import Pipeline

    from foundry_ml import Model, Stage

    X_train = housing[['median_income', "housing_median_age", "total_rooms"]]
    y_train = housing['median_house_value']

    # ベクトル化を作成
    column_transformer = make_column_transformer(
       ('passthrough', ['median_income', "housing_median_age", "total_rooms"])
    )
    # ベクトル化 -> 標準化 -> 多項式展開 -> 線形回帰
    pipeline = Pipeline([
            ("preprocessing", column_transformer),
            ("ss", StandardScaler()),
            ("pf", PolynomialFeatures(degree=2)),
            ("lr", LinearRegression())
       ])

    pipeline.fit(X_train, y_train)

    # パイプラインを展開
    # リスト内包表記で作成することもできます:  Model(*[Stage(s[1]) for s in pipeline.steps])
    model = Model( Stage(pipeline["preprocessing"]), 
                    Stage(pipeline["ss"], output_column_name="features"), 
                    Stage(pipeline["pf"], output_column_name="features"), 
                    Stage(pipeline["lr"]))
                    
    return model

SparkMLパイプラインを使用した例

この例では、SparkMLパイプラインを保存する方法を示します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def spark_pipeline(housing):
    from pyspark.ml.feature import PolynomialExpansion
    from pyspark.ml.feature import StandardScaler
    from pyspark.ml.feature import VectorAssembler
    from pyspark.ml.regression import LinearRegression
    from pyspark.ml import Pipeline
    from foundry_ml import Model, Stage

    # 変数をベクトル化するためのアセンブラーを設定します
    assembler = VectorAssembler(
        inputCols=['median_income', "housing_median_age", "total_rooms"], # 入力列
        outputCol="features" # 出力列
        )
    # 特徴量のスケーラーを設定します
    scaler = StandardScaler(inputCol="features", outputCol="scaledFeatures",
                        withStd=True, withMean=False)
                        
    # ポリノミアル拡張を設定します
    polyExpansion = PolynomialExpansion(degree=2, inputCol="scaledFeatures", outputCol="polyFeatures")

    # 線形回帰モデルを設定します
    lr = LinearRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8, featuresCol='polyFeatures', labelCol='median_house_value')

    # パイプラインを作成します
    pipeline = Pipeline(stages=[assembler, scaler, polyExpansion, lr])

    # パイプラインを訓練データに適合させます
    pipeline = pipeline.fit(housing)

    # パイプラインを展開します
    model = Model(*[Stage(s) for s in pipeline.stages])

    return model

このコードは、PySparkを使用してデータ処理と機械学習のパイプラインを作成しています。まず、特定の特徴量をベクトルとして組み合わせるVectorAssemblerを設定します。次に、StandardScalerを使用して特徴量をスケーリング（標準化）します。これにより、各特徴量のスケールが同じになり、機械学習アルゴリズムがより効率的に学習できます。PolynomialExpansionを使用して特徴量を2次元に拡張します。これは、特徴量間の関係が線形でない場合に役立ちます。最後に、LinearRegressionアルゴリズムを使用して回帰モデルをトレーニングします。これらのステップ全体がパイプラインとして組み合わせられ、トレーニングデータに適合します。パイプラインが適合した後、パイプラインの各ステージ（ステップ）が個別のステージとしてモデルに格納されます。