Apache Spark MLlib で機械学習モデルを構築する

この記事では、Apache Spark MLlib を使用して、開いているデータセットの予測分析を処理する機械学習アプリケーションAzure作成する方法について説明します。 Spark には、組み込みの機械学習ライブラリが用意されています。 この例では、ロジスティック回帰による分類を使用しています。

このチュートリアルでは、次の手順について説明します。

コア SparkML と MLlib Spark ライブラリには、機械学習タスクに役立つ多数のユーティリティが用意されています。 これらのユーティリティは、次の用途に適しています。

• 分類
• クラスタリング
• 仮説テストとサンプル統計の計算
• リグレッションテスト
• 特異値分解 (SVD) と主成分分析 (PCA)
• トピックのモデリング

Prerequisites

• Microsoft Fabric サブスクリプションを取得します。 または、無料の Microsoft Fabric 試用版にサインアップします。

• Microsoft Fabric にサインインします。

• ホーム ページの左下にあるエクスペリエンス スイッチャーを使用して Fabric に切り替えます。

  

• 必要に応じて、Microsoft Fabric でレイクハウスを作成するに記載されている手順に従って、Microsoft Fabric のレイクハウスを作成します。
• [+] を選択し、ノートブックを使用して、ワークスペースに新しいノートブックを作成します。 詳細については、「 ノートブックの作成」を参照してください。

分類およびロジスティック回帰について

一般的な Machine Learning タスクである分類は、入力データをカテゴリに分類します。 分類アルゴリズムは、指定された入力データに ラベル を割り当てる方法を示します。 たとえば、機械学習アルゴリズムでは、株式情報を入力として受け入れ、株式を 2 つのカテゴリ (販売する必要がある株式と保持する必要がある株式) に分割できます。

ロジスティック回帰アルゴリズムは、分類に役立ちます。 Spark のロジスティック回帰 API は、入力データを 2 つのグループのいずれかに分類する二項分類に適しています。 ロジスティック回帰の詳細については、Wikipedia を参照してください。

ロジスティック回帰では、入力ベクトルが 1 つのグループまたは他のグループに属する確率を予測する ロジスティック関数 が生成されます。

NYC タクシー データの予測分析の例

データは Azure Open Datasets リソースから入手できます。 このデータセットのサブセットには、出発と到着の時刻および場所、料金、その他の属性などを含むイエロー タクシー乗車に関する情報がホストされています。

このチュートリアルでは、Apache Spark を使用して NYC タクシー乗車チップ データの分析を実行し、特定の乗車にチップが含まれているかどうかを予測するモデルを開発します。

Apache Spark 機械学習モデルを作成する

1. PySpark ノートブックを作成します。 詳細については、「 ノートブックの作成」を参照してください。

   ノートブックを作成したら、左側のパネルで [レイクハウスの追加] を選択して 、ノートブックをレイクハウス にアタッチします。

2. このノートブックに必要な型をインポートします。 次のコードを最初のセルに貼り付けて実行します。

   import matplotlib.pyplot as plt
   from pyspark.sql.functions import unix_timestamp, date_format, col, when
   from pyspark.ml import Pipeline
   from pyspark.ml.feature import RFormula
   from pyspark.ml.feature import OneHotEncoder, StringIndexer
   from pyspark.ml.classification import LogisticRegression
   from pyspark.ml.evaluation import BinaryClassificationEvaluator
   

   確認: セルは ImportErrorなしで完了します。 エラーが表示された場合は、ノートブックで PySpark ランタイムが使用されているかどうかを確認します。

3. MLflow を使用して、機械学習の実験と対応する実行を追跡します。 Microsoft Fabric の自動ログ記録が有効になっている場合、対応するメトリックとパラメーターが自動的にキャプチャされます。

   import mlflow
   

   確認: セルはエラーなしで完了します。 print(mlflow.__version__)を実行して、MLflow が使用可能であることを確認します。

入力 DataFrame を作成する

この例では、Azure Open Datasets ストレージから Apache Spark DataFrame にデータを読み込みます。 次に、Spark 操作を適用してデータセットのクリーンアップとフィルター処理を行います。

1. 次のコードを新しいセルに貼り付け、それを実行して Spark DataFrame を作成します。 この手順では、2018 年 5 月にフィルター処理された NYC 黄色のタクシー データを取得します。

   blob_account_name = "azureopendatastorage"
   blob_container_name = "nyctlc"
   blob_relative_path = "yellow"
   wasbs_path = f"wasbs://{blob_container_name}@{blob_account_name}.blob.core.windows.net/{blob_relative_path}"
   
   nyc_tlc_df = spark.read.parquet(wasbs_path) \
       .filter((col("tpepPickupDateTime") >= "2018-05-01") & (col("tpepPickupDateTime") < "2018-06-01")) \
       .repartition(20)
   

   確認: 次のセルを実行して、データの読み込みが正常に完了したことを確認します。

   print(f"Loaded {nyc_tlc_df.count()} rows")
   # Expected output: Loaded approximately 9,000,000+ rows
   

2. データセットをサンプリングして、開発とトレーニングを高速化します。

   # Sample without replacement to avoid duplicates
   sampled_taxi_df = nyc_tlc_df.sample(False, 0.001, seed=1234)
   

   確認: サンプル サイズが管理可能であることを確認します。

   print(f"Sampled {sampled_taxi_df.count()} rows")
   # Expected output: Sampled approximately 9,000-10,000 rows
   

3. 組み込みの display() コマンドを使用してデータを表示し、データ サンプルを探索します。

   display(sampled_taxi_df.limit(10))
   

   確認: 10 行のテーブルが表示され、 tpepPickupDateTime、 fareAmount、 tipAmount、 tripDistanceなどの列が表示されます。

データを準備する

データの準備は、機械学習プロセスの重要なステップです。 これには、生データのクリーニング、変換、整理が含まれており、分析とモデリングに適しています。 このセクションでは、いくつかのデータ準備手順を実行します。

• データセットをフィルター処理して、外れ値と正しくない値を削除します。
• モデルの学習に不要な列を削除します。
• 生データから新しい列を作成します。
• 特定のタクシー乗車にチップが含まれるかどうかを判断するラベルを生成します。

次のコードを実行して、関連する列を選択し、派生特徴を計算し、外れ値をフィルター処理します。

taxi_df = sampled_taxi_df.select('totalAmount', 'fareAmount', 'tipAmount', 'paymentType', 'rateCodeId', 'passengerCount',
                    'tripDistance', 'tpepPickupDateTime', 'tpepDropoffDateTime',
                    date_format('tpepPickupDateTime', 'HH').cast('integer').alias('pickupHour'),
                    date_format('tpepPickupDateTime', 'EEEE').alias('weekdayString'),
                    (unix_timestamp(col('tpepDropoffDateTime')) - unix_timestamp(col('tpepPickupDateTime'))).alias('tripTimeSecs'),
                    (when(col('tipAmount') > 0, 1).otherwise(0)).alias('tipped')
                    ) \
            .filter((sampled_taxi_df.passengerCount > 0) & (sampled_taxi_df.passengerCount < 8)
                    & (sampled_taxi_df.tipAmount >= 0) & (sampled_taxi_df.tipAmount <= 25)
                    & (sampled_taxi_df.fareAmount >= 1) & (sampled_taxi_df.fareAmount <= 250)
                    & (sampled_taxi_df.tipAmount < sampled_taxi_df.fareAmount)
                    & (sampled_taxi_df.tripDistance > 0) & (sampled_taxi_df.tripDistance <= 100)
                    & (sampled_taxi_df.rateCodeId <= 5)
                    & (sampled_taxi_df.paymentType.isin({"1", "2"}))
                    )

次に、時間帯に基づいて交通時刻ビン機能を追加します。

taxi_featurised_df = taxi_df.select('totalAmount', 'fareAmount', 'tipAmount', 'paymentType', 'passengerCount',
                                    'tripDistance', 'weekdayString', 'pickupHour', 'tripTimeSecs', 'tipped',
                                    when((col('pickupHour') <= 6) | (col('pickupHour') >= 20), "Night")
                                    .when((col('pickupHour') >= 7) & (col('pickupHour') <= 10), "AMRush")
                                    .when((col('pickupHour') >= 11) & (col('pickupHour') <= 15), "Afternoon")
                                    .when((col('pickupHour') >= 16) & (col('pickupHour') <= 19), "PMRush")
                                    .otherwise("Other").alias('trafficTimeBins')
                                    ) \
                            .filter((taxi_df.tripTimeSecs >= 30) & (taxi_df.tripTimeSecs <= 7200))

確認: トラフィックの時間ビンが正しく配分されていることを確認してください。

taxi_featurised_df.groupBy('trafficTimeBins').count().show()
# Expected output: Shows counts for Night, AMRush, Afternoon, PMRush categories

ロジスティック回帰モデルを作成する

最後のタスクでは、ラベル付けされたデータをロジスティック回帰を処理できる形式に変換します。 ロジスティック回帰アルゴリズムへの入力は、"ラベルと特徴ベクトルのペア" 構造である必要があります。ここで "特徴ベクトル" とは、入力ポイントを表す数のベクトルです。

trafficTimeBinsアプローチを使用して、カテゴリ列weekdayStringとOneHotEncoderを整数表現に変換します。

# Convert categorical features into numeric representations
sI1 = StringIndexer(inputCol="trafficTimeBins", outputCol="trafficTimeBinsIndex")
en1 = OneHotEncoder(inputCol="trafficTimeBinsIndex", outputCol="trafficTimeBinsVec")
sI2 = StringIndexer(inputCol="weekdayString", outputCol="weekdayIndex")
en2 = OneHotEncoder(inputCol="weekdayIndex", outputCol="weekdayVec")

# Apply the encodings to create a new DataFrame
encoded_final_df = Pipeline(stages=[sI1, en1, sI2, en2]).fit(taxi_featurised_df).transform(taxi_featurised_df)

確認: エンコードされた DataFrame に予期される新しい列があることを確認します。

print("Columns:", encoded_final_df.columns)
print(f"Row count: {encoded_final_df.count()}")
# Expected output: Columns list includes 'trafficTimeBinsVec' and 'weekdayVec'

ロジスティック回帰モデルのトレーニング

データセットをトレーニング セット (70%) とテスト セット (30%) に分割します。

# Split the DataFrame into training and test sets
trainingFraction = 0.7
testingFraction = (1 - trainingFraction)
seed = 1234

train_data_df, test_data_df = encoded_final_df.randomSplit([trainingFraction, testingFraction], seed=seed)

検証: 分割によって適切なサイズが生成されたことを確認します。

print(f"Training rows: {train_data_df.count()}, Test rows: {test_data_df.count()}")
# Expected output: Approximately 70%/30% split of the encoded data

モデル式を作成し、ロジスティック回帰モデルをトレーニングし、ROC (受信側動作特性) 曲線の下の面積を使用して評価します。

# Create a logistic regression model
logReg = LogisticRegression(maxIter=10, regParam=0.3, labelCol='label')

# Define the formula: 'tipped' is the response variable, right-hand side are predictors
classFormula = RFormula(formula="tipped ~ pickupHour + weekdayVec + passengerCount + tripTimeSecs + tripDistance + fareAmount + paymentType + trafficTimeBinsVec")

# Train the model using a pipeline
lrModel = Pipeline(stages=[classFormula, logReg]).fit(train_data_df)

# Generate predictions on the test dataset
predictions = lrModel.transform(test_data_df)

# Evaluate using Area Under ROC
evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", metricName="areaUnderROC")
auc = evaluator.evaluate(predictions)
print(f"Area under ROC = {auc}")

確認: 出力には AUC 値が表示されます。 パフォーマンスの高いモデルでは、1.0 に近い値が生成されます。

Area under ROC = 0.97 (approximately)

予測を視覚化する

モデルの結果を解釈するための最終的な視覚化を作成します。 ROC 曲線は、真陽性率と偽陽性率のトレードオフを示します。

# Plot the ROC curve from the model training summary
modelSummary = lrModel.stages[-1].summary

# Extract FPR and TPR values as plain lists
roc_data = modelSummary.roc.select('FPR', 'TPR').toPandas()

plt.figure(figsize=(8, 6))
plt.plot([0, 1], [0, 1], 'r--', label='Random classifier')
plt.plot(roc_data['FPR'], roc_data['TPR'], label=f'Logistic Regression (AUC = {auc:.4f})')
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
plt.title('ROC Curve - NYC Taxi Tip Prediction')
plt.legend(loc='lower right')
plt.show()

確認: 赤い破線の斜めの線の上に ROC 曲線が表示されたプロットが表示されます。 曲線は左上隅に向かって弓なりに曲がり、高い分類性能を示すはずです。

リソースをクリーンアップする

このチュートリアルを完了したら、ノートブックと lakehouse を削除してワークスペースの容量を解放します。

1. ワークスペースでノートブックを右クリックし、[削除] を選択 します。
2. このチュートリアル専用のレイクハウスを作成した場合は、それを右クリックして [削除] を選択 します。

トレーニング済みのモデルを将来使用するために保持するには、クリーンアップの前に次のコードを追加します。

# Save the model to the lakehouse
model_path = "abfss://<your-workspace>@onelake.dfs.fabric.microsoft.com/<your-lakehouse>.Lakehouse/Files/models/taxi_tip_model"
lrModel.write().overwrite().save(model_path)
print(f"Model saved to: {model_path}")

Troubleshooting

関連するコンテンツ

• AI サンプルを使用して機械学習モデルを構築する: AI サンプルを使用する
• 実験を使用して機械学習の実行を追跡する: 機械学習の実験