การจัดประเภทงานโดยใช้ SynapseML

บทความนี้แสดงวิธีการทํางานการจัดประเภทข้อความด้วยสองวิธี วิธีหนึ่งใช้pysparkธรรมดา และอีกวิธีหนึ่งใช้ไลบรารีsynapseml ทั้งสองวิธีให้ประสิทธิภาพเท่ากัน แต่เน้นว่า SynapseML ช่วยลดความซับซ้อนของโค้ดได้อย่างไรเมื่อเทียบกับpyspark

งานจะคาดการณ์ว่าบทวิจารณ์ของลูกค้าเกี่ยวกับหนังสือที่ขายใน Amazon นั้นดี (คะแนน > 3) หรือไม่ดี โดยพิจารณาจากข้อความรีวิว คุณฝึกอบรมผู้เรียน LogisticRegression ด้วยไฮเปอร์พารามิเตอร์ที่แตกต่างกัน แล้วเลือกโมเดลที่ดีที่สุด

ข้อกําหนดเบื้องต้น

• รับการสมัครใช้งาน Microsoft Fabric หรือลงทะเบียนเพื่อทดลองใช้งาน Microsoft Fabric ฟรี

• ลงชื่อเข้าใช้ Microsoft Fabric

• สลับไปยัง Fabric โดยใช้ตัวสลับประสบการณ์ที่ด้านซ้ายล่างของโฮมเพจของคุณ

  

• สร้างสมุดบันทึก
• แนบสมุดบันทึกของคุณเข้ากับเลคเฮ้าส์ ในสมุดบันทึก ให้เลือก เพิ่ม บนบานหน้าต่างด้านซ้ายเพื่อแนบเลคเฮาส์ที่มีอยู่หรือสร้างเลคเฮาส์ใหม่

โหลดและสํารวจข้อมูล

ในสมุดบันทึก Fabric เซสชัน Spark พร้อมใช้งานแล้วเป็นตัวแปร spark โหลดชุดข้อมูลบทวิจารณ์หนังสือของ Amazon จากตําแหน่ง Azure Blob Storage สาธารณะ:

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

ตรวจสอบว่าชุดข้อมูลโหลดอย่างถูกต้อง:

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

แยกคุณลักษณะและกระบวนการข้อมูล

ข้อมูลจริงมักมีคุณสมบัติหลายประเภท เช่น ข้อความ ตัวเลข และหมวดหมู่ เมื่อต้องการสาธิตการทํางานกับชนิดคุณลักษณะแบบผสม ให้เพิ่มคุณลักษณะตัวเลขสองคุณลักษณะลงในชุดข้อมูล: จํานวนคําของบทวิจารณ์และความยาวของคําเฉลี่ย

กําหนดฟังก์ชันที่ผู้ใช้กําหนดเอง (UDF)

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

ใช้ UDF กับ SynapseML UDFTransformer

ใช้ from UDFTransformer SynapseML เพื่อห่อ UDF ลงในหม้อแปลงที่เข้ากันได้กับไปป์ไลน์:

from synapse.ml.stages import UDFTransformer

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

จัดประเภทโดยใช้ pyspark

ในการเลือกตัวแยกประเภท LogisticRegression ที่ดีที่สุดโดยใช้ pyspark ไลบรารี คุณต้องทําตามขั้นตอนเหล่านี้อย่างชัดเจน:

1. ประมวลผลคุณสมบัติ:
   • โทเค็นคอลัมน์ข้อความ
   • แฮชคอลัมน์โทเค็นเป็นเวกเตอร์โดยใช้การแฮช
   • รวมคุณลักษณะตัวเลขกับเวกเตอร์
2. ส่งคอลัมน์ป้ายชื่อจากบูลีนเป็นประเภทจํานวนเต็ม
3. ฝึกอัลกอริธึม LogisticRegression หลายรายการบน train ชุดข้อมูลด้วยไฮเปอร์พารามิเตอร์ที่แตกต่างกัน
4. คํานวณพื้นที่ภายใต้เส้นโค้ง ROC (AUC) สําหรับแต่ละโมเดลที่ผ่านการฝึกอบรม และเลือกโมเดลที่มีเมตริกสูงสุดใน test ชุดข้อมูล
5. ประเมินโมเดล validation ที่ดีที่สุดในฉาก

นําเสนอและเตรียมข้อมูล

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

ตรวจสอบข้อมูลที่แสดง:

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

ฝึกอบรมและประเมินแบบจําลอง

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

ตรวจสอบผลลัพธ์:

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

จัดประเภทโดยใช้ SynapseML

วิธีการนี้ synapseml ให้ผลลัพธ์เดียวกันโดยมีขั้นตอนน้อยลง SynapseML จัดการคุณสมบัติภายใน ซึ่งจะช่วยลดโค้ดที่คุณต้องเขียน:

1. TrainClassifierตัวประมาณค่าจะจัดเตรียมข้อมูลภายใน ตราบใดที่คอลัมน์ใน train, testและvalidationชุดข้อมูลแสดงคุณลักษณะ
2. FindBestModelตัวประมาณจะค้นหาแบบจําลองที่ดีที่สุดจากกลุ่มของแบบจําลองที่ผ่านการฝึกอบรมโดยการประเมินประสิทธิภาพในtestชุดข้อมูลด้วยเมตริกที่ระบุ
3. ComputeModelStatisticsหม้อแปลงจะคํานวณเมตริกหลายตัวในชุดข้อมูลที่ให้คะแนน (ในกรณีนี้คือvalidationชุดข้อมูล) พร้อมกัน

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

ตรวจสอบผลลัพธ์ SynapseML:

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

เปรียบเทียบทั้งสองวิธี

แก้ไข ปัญหา

ล้างแหล่งข้อมูล

หากคุณสร้างเลคเฮาส์ใหม่สําหรับบทความนี้และไม่ต้องการอีกต่อไป:

1. ในพื้นที่ทํางานของคุณ ให้คลิกขวาที่ชื่อเลคเฮาส์
2. เลือก ลบ
3. ยืนยันการลบ

สมุดบันทึกจะยังคงอยู่ในพื้นที่ทํางานของคุณ เว้นแต่คุณจะลบแยกต่างหาก

เนื้อหาที่เกี่ยวข้อง

• วิธีใช้แบบจําลอง k-NN (K-Nearest-Neighbors) ด้วย SynapseML
• วิธีใช้ ONNX กับ SynapseML - การเรียนรู้เชิงลึก
• วิธีใช้ Kernel SHAP เพื่ออธิบายแบบจําลองการจัดประเภทแบบตาราง