Klassifiseringsoppgaver ved hjelp av SynapseML

Denne artikkelen viser hvordan man kan utføre en tekstklassifiseringsoppgave med to metoder. Den ene metoden bruker vanlig pyspark, og den andre bruker biblioteket synapseml . Begge metodene gir samme ytelse, men fremhever hvordan SynapseML reduserer kodekompleksiteten sammenlignet med pyspark.

Oppgaven forutsier om en kundeanmeldelse av en bok solgt på Amazon er god (vurdering > 3) eller dårlig, basert på anmeldelsesteksten. Du trener LogisticRegression-lærere med ulike hyperparametere, og velger deretter den beste modellen.

Forutsetninger

• Få et Microsoft Fabric-abonnement. Eller registrer deg for en gratis prøveversjon av Microsoft Fabric.

• Logg på Microsoft Fabric.

• Bytt til Fabric ved å bruke erfaringsbryteren nederst til venstre på hjemmesiden din.

  

• Opprett en notatblokk.
• Legg notatblokken til et lakehouse. I notatboken velger du Legg til i venstre panel for å knytte til et eksisterende innsjøhus eller opprette et nytt.

Last inn og utforsk dataene

I Fabric notatbøker er en Spark-økt allerede tilgjengelig som variabelen spark. Last inn Amazon-bokanmeldelsesdatasett fra et offentlig Azure Blob Storage-sted:

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

Sjekk at datasettet lastet riktig:

print(f"Row count: {rawData.count()}")
print(f"Columns: {rawData.columns}")
assert rawData.count() == 10000, "Expected 10,000 rows"
assert set(rawData.columns) == {"text", "rating"}, "Expected columns: text, rating"
print("Data loaded successfully")

Trekke ut funksjoner og behandle data

Virkelige data har ofte egenskaper av flere typer, for eksempel tekst, numerisk og kategorisk. For å demonstrere arbeid med blandede funksjonstyper, legg til to numeriske funksjoner i datasettet: ordtellingen i anmeldelsen og gjennomsnittlig ordlengde.

Definer brukerdefinerte funksjoner (UDF-er)

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType, DoubleType
import numpy as np


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


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


wordLengthUDF = udf(calc_word_length, DoubleType())
wordCountUDF = udf(calc_word_count, IntegerType())

Bruk UDF-er med SynapseML UDFTransformer

Bruk fra UDFTransformer SynapseML til å pakke UDF-ene inn i rørledningskompatible transformatorer:

from synapse.ml.stages import UDFTransformer

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

Kjør feature-pipelinen

Bruk begge transformatorene og lag en binær etikettkolonne fra vurderingen:

from pyspark.ml import Pipeline

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

Verifiser funksjonsuttrekkingen:

data.show(5)
print(f"Columns: {data.columns}")
assert "wordLength" in data.columns, "wordLength column missing"
assert "wordCount" in data.columns, "wordCount column missing"
assert "label" in data.columns, "label column missing"
assert "rating" not in data.columns, "rating column should be dropped"
print("Feature extraction successful")

Klassifisere ved hjelp av pyspark

For å velge den beste LogisticRegression-klassifisatoren ved hjelp av biblioteket pyspark , må du eksplisitt utføre disse trinnene:

1. Behandle funksjonene:
   • Tokeniser tekstkolonnen.
   • Hash den tokeniserte kolonnen til en vektor ved å bruke hashing.
   • Slå sammen de numeriske egenskapene med vektoren.
2. Kast etikettkolonnen fra boolsk til heltallstype.
3. Tren flere LogisticRegresjonsalgoritmer på datasettet train med ulike hyperparametere.
4. Beregn arealet under ROC-kurven (AUC) for hver trent modell og velg modellen med høyest metrikk på test datasettet.
5. Vurder den beste modellen på settet validation .

Presenter og forbered dataene

from pyspark.ml.feature import Tokenizer, HashingTF, VectorAssembler
from pyspark.sql.types import IntegerType

# Tokenize the 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 into one feature column
featureColumnsArray = ["TextFeatures", "wordCount", "wordLength"]
assembler = VectorAssembler(inputCols=featureColumnsArray, outputCol="features")
assembledData = assembler.transform(featurizedData)

# Select only the label and features columns, cast label to integer
processedData = assembledData.select("label", "features").withColumn(
    "label", assembledData.label.cast(IntegerType())
)

Verifiser de presenterte dataene:

print(f"Feature vector size: {processedData.first()['features'].size}")
print(f"Label values: {sorted(processedData.select('label').distinct().rdd.flatMap(lambda x: x).collect())}")
assert processedData.first()["features"].size == 10002, "Expected 10000 text + 2 numeric features"
print("Featurization successful")

Tren og evaluer modeller

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

# Split the data into train, test, and validation sets
train, test, validation = processedData.randomSplit([0.60, 0.20, 0.20], seed=123)

# Train models with different regularization parameters
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 = []

# Train each model and evaluate on the test set
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)]

# Evaluate the best model on the validation dataset
scoredVal = bestModel.transform(validation)
validationAUC = evaluator.evaluate(scoredVal)
print(f"Best model's AUC on validation set = {validationAUC:.4f}")

Verifiser resultatene:

print(f"Number of models trained: {len(models)}")
print(f"Best regularization parameter: {lrHyperParams[metrics.index(bestMetric)]}")
print(f"Test AUC scores: {[f'{m:.4f}' for m in metrics]}")
assert 0.5 < validationAUC <= 1.0, f"AUC {validationAUC} is outside expected range (0.5, 1.0]"
print(f"pyspark classification complete - AUC: {validationAUC:.4f}")

Klassifisere ved hjelp av SynapseML

Tilnærmingen synapseml oppnår samme resultat med færre trinn. SynapseML håndterer featurisering internt, noe som reduserer koden du trenger for å skrive:

1. Estimatoren TrainClassifier fremhever dataene internt, så lenge kolonnene i train, test, og validation datasettene representerer egenskapene.
2. Estimatoren FindBestModel finner den beste modellen fra en pool av trente modeller ved å evaluere ytelsen på datasettet test med den angitte metrikken.
3. Transformatoren ComputeModelStatistics beregner flere måleparametere på et scoret datasett (i dette tilfellet datasettet validation ) samtidig.

from synapse.ml.train import TrainClassifier, ComputeModelStatistics
from synapse.ml.automl import FindBestModel
from pyspark.ml.classification import LogisticRegression

# Split the raw feature data (SynapseML handles featurization internally)
train, test, validation = data.randomSplit([0.60, 0.20, 0.20], seed=123)

# Train models with different regularization parameters
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 based on AUC
bestModel = FindBestModel(evaluationMetric="AUC", models=lrmodels).fit(test)

# Compute metrics 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)
)

Verifiser SynapseML-resultatene:

auc_value = metrics.first()["AUC"]
print(f"Available metrics: {metrics.columns}")
assert 0.5 < auc_value <= 1.0, f"AUC {auc_value} is outside expected range (0.5, 1.0]"
print(f"SynapseML classification complete - AUC: {auc_value:.4f}")

Sammenlign de to tilnærmingene

Feilsøking

Rydd opp ressurser

Hvis du opprettet et nytt innsjøhus for denne artikkelen og ikke lenger trenger det:

1. I arbeidsområdet ditt, høyreklikk på navnet på innsjøhuset.
2. Velg Slett.
3. Bekreft slettingen.

Notatboken forblir i arbeidsområdet ditt med mindre du sletter den separat.

Relatert innhold

• Slik bruker du k-NN -modellen (K-Nearest-Neighbors) med SynapseML
• Slik bruker du ONNX med SynapseML – Deep Learning
• Slik bruker du Kernel SHAP til å forklare en tabellklassifiseringsmodell