SynapseML を使用した分類タスク

[アーティクル]
12/28/2023

この記事では、2 つの異なる方法で同じ分類タスクを実行します。1 つはプレーン pyspark を使用し、もう 1 つは synapseml ライブラリを使用します。 2 つの方法では、同じパフォーマンスが得られますが、pyspark と比較して synapseml を使用することのシンプルさが強調されます。

このタスクは、Amazon で販売されている書籍の顧客のレビューが、レビューのテキストに基づいて良い (評価 > 3) か悪いかを予測することです。これを実現するには、さまざまなハイパーパラメーターを使用して LogisticRegression 学習者をトレーニングし、最適なモデルを選択します。

前提条件

ノートブックをレイクハウスにアタッチします。左側の [追加] を選択して、既存のレイクハウスを追加するか、レイクハウスを作成します。

段取り

必要な Python ライブラリをインポートし、Spark セッションを取得します。

from pyspark.sql import SparkSession

# Bootstrap Spark Session
spark = SparkSession.builder.getOrCreate()

データを読み取る

データをダウンロードして読み取ります。

rawData = spark.read.parquet(
    "wasbs://publicwasb@mmlspark.blob.core.windows.net/BookReviewsFromAmazon10K.parquet"
)
rawData.show(5)

特徴の抽出とデータの処理

実際のデータは、上記のデータセットよりも複雑です。データセットには、テキスト、数値、カテゴリなどの複数の種類の特徴が含まれるのが一般的です。これらのデータセットを処理するのがどれほど難しいかを示すために、データセットに 2 つの数値特徴 (レビューの単語数と単語の長さの平均) を追加します。

from pyspark.sql.functions import udf
from pyspark.sql.types import *


def wordCount(s):
    return len(s.split())


def wordLength(s):
    import numpy as np

    ss = [len(w) for w in s.split()]
    return round(float(np.mean(ss)), 2)


wordLengthUDF = udf(wordLength, DoubleType())
wordCountUDF = udf(wordCount, IntegerType())

from synapse.ml.stages import UDFTransformer

wordLength = "wordLength"
wordCount = "wordCount"
wordLengthTransformer = UDFTransformer(
    inputCol="text", outputCol=wordLength, udf=wordLengthUDF
)
wordCountTransformer = UDFTransformer(
    inputCol="text", outputCol=wordCount, udf=wordCountUDF
)

from pyspark.ml import Pipeline

data = (
    Pipeline(stages=[wordLengthTransformer, wordCountTransformer])
    .fit(rawData)
    .transform(rawData)
    .withColumn("label", rawData["rating"] > 3)
    .drop("rating")
)

data.show(5)

pyspark を使用して分類する

pyspark ライブラリを使用して最適な LogisticRegression 分類子を選択するには、次の手順を "明示的に" 実行する必要があります。

この特徴を処理します:
- テキスト列をトークン化します
- ハッシュを使用してトークン化された列をベクターにハッシュします
- 数値特徴をベクトルと結合します
ラベル列を処理します。適切な型にキャストします。
異なるハイパーパラメーターを使用して train データセットに対して複数の LogisticRegression アルゴリズムをトレーニングします
トレーニング済みの各モデルの ROC 曲線の下の領域を計算し、test データセットで計算されたメトリックが最も高いモデルを選択します
validation セットに最適なモデルを評価します

from pyspark.ml.feature import Tokenizer, HashingTF
from pyspark.ml.feature import VectorAssembler

# Featurize text column
tokenizer = Tokenizer(inputCol="text", outputCol="tokenizedText")
numFeatures = 10000
hashingScheme = HashingTF(
    inputCol="tokenizedText", outputCol="TextFeatures", numFeatures=numFeatures
)
tokenizedData = tokenizer.transform(data)
featurizedData = hashingScheme.transform(tokenizedData)

# Merge text and numeric features in one feature column
featureColumnsArray = ["TextFeatures", "wordCount", "wordLength"]
assembler = VectorAssembler(inputCols=featureColumnsArray, outputCol="features")
assembledData = assembler.transform(featurizedData)

# Select only columns of interest
# Convert rating column from boolean to int
processedData = assembledData.select("label", "features").withColumn(
    "label", assembledData.label.cast(IntegerType())
)

from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.classification import LogisticRegression

# Prepare data for learning
train, test, validation = processedData.randomSplit([0.60, 0.20, 0.20], seed=123)

# Train the models on the 'train' data
lrHyperParams = [0.05, 0.1, 0.2, 0.4]
logisticRegressions = [
    LogisticRegression(regParam=hyperParam) for hyperParam in lrHyperParams
]
evaluator = BinaryClassificationEvaluator(
    rawPredictionCol="rawPrediction", metricName="areaUnderROC"
)
metrics = []
models = []

# Select the best model
for learner in logisticRegressions:
    model = learner.fit(train)
    models.append(model)
    scoredData = model.transform(test)
    metrics.append(evaluator.evaluate(scoredData))
bestMetric = max(metrics)
bestModel = models[metrics.index(bestMetric)]

# Get AUC on the validation dataset
scoredVal = bestModel.transform(validation)
print(evaluator.evaluate(scoredVal))