Tugas klasifikasi menggunakan SynapseML

2025-04-30

Artikel ini memperlihatkan cara melakukan tugas klasifikasi tertentu dengan dua metode. Satu cara menggunakan pyspark biasa, dan satu cara menggunakan pustaka synapseml. Meskipun metode menghasilkan performa yang sama, metode ini menyoroti kesederhanaan synapseml dibandingkan dengan pyspark.

Tugas yang dijelaskan dalam artikel ini memprediksi apakah ulasan pelanggan tertentu tentang buku yang dijual di Amazon baik (peringkat > 3) atau buruk, berdasarkan teks ulasan. Untuk membangun tugas, Anda melatih pelajar LogisticRegression dengan hiperparameter yang berbeda, lalu memilih model terbaik.

Prasyarat

Lampirkan buku catatan Anda ke rumah tepi danau. Di sisi kiri, Anda dapat memilih Tambahkan untuk menambahkan lakehouse yang ada, atau Anda dapat membuat lakehouse baru.

Pengaturan

Impor pustaka Python yang diperlukan, dan dapatkan sesi Spark:

from pyspark.sql import SparkSession

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

Bacalah data

Unduh, dan baca data tersebut.

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

Mengekstrak fitur dan memproses data

Data nyata memiliki lebih banyak kompleksitas, dibandingkan dengan himpunan data yang kami unduh sebelumnya. Himpunan data sering memiliki fitur dari beberapa jenis - misalnya, teks, numerik, dan kategoris. Untuk memperlihatkan kesulitan bekerja dengan himpunan data ini, tambahkan dua fitur numerik ke himpunan data: jumlah kata tinjauan dan panjang kata rata-rata:

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)

Mengklasifikasikan menggunakan pyspark

Untuk memilih pengklasifikasi LogisticRegression terbaik menggunakan pyspark pustaka, Anda harus secara eksplisit melakukan langkah-langkah berikut:

Memproses fitur
- Tokenisasi kolom teks
- Hash kolom yang ditokenisasi ke dalam vektor menggunakan hashing
- Menggabungkan fitur numerik dengan vektor
Untuk memproses kolom label, ubah kolom tersebut ke dalam jenis yang tepat
Melatih beberapa algoritma LogisticRegression pada himpunan train data, dengan hiperparameter yang berbeda
Hitung area di bawah kurva ROC untuk setiap model terlatih, dan pilih model dengan metrik tertinggi yang dihitung pada test himpunan data
Mengevaluasi model terbaik pada validation set

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))

Mengklasifikasikan menggunakan SynapseML

Opsi ini synapseml melibatkan langkah-langkah yang lebih sederhana:

Estimator TrainClassifier secara internal mengolah data menjadi fitur, selama kolom yang dipilih di train, test, validation himpunan data merupakan fitur.
Estimator FindBestModel menemukan model terbaik dari kumpulan model terlatih. Untuk melakukan ini, ia menemukan model yang berkinerja terbaik pada himpunan test data yang diberikan metrik yang ditentukan
ComputeModelStatistics Transformer menghitung metrik yang berbeda pada himpunan data yang dinilai (dalam kasus kami, himpunan validation data) secara bersamaan

from synapse.ml.train import TrainClassifier, ComputeModelStatistics
from synapse.ml.automl import FindBestModel

# Prepare data for learning
train, test, validation = data.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
]
lrmodels = [
    TrainClassifier(model=lrm, labelCol="label", numFeatures=10000).fit(train)
    for lrm in logisticRegressions
]

# Select the best model
bestModel = FindBestModel(evaluationMetric="AUC", models=lrmodels).fit(test)


# Get AUC on the validation dataset
predictions = bestModel.transform(validation)
metrics = ComputeModelStatistics().transform(predictions)
print(
    "Best model's AUC on validation set = "
    + "{0:.2f}%".format(metrics.first()["AUC"] * 100)
)

Bagikan melalui