แชร์ผ่าน

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

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

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

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

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

การตั้งค่า

นําเข้าไลบรารี 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)

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

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

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

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

  1. ประมวลผลคุณลักษณะ
    • โทเค็นคอลัมน์ข้อความ
    • แฮชคอลัมน์โทเค็นลงในเวกเตอร์โดยใช้การแฮช
    • ผสานคุณลักษณะตัวเลขกับเวกเตอร์
  2. ในการประมวลผลคอลัมน์ป้ายชื่อ ให้แคสต์คอลัมน์นั้นให้เป็นประเภทที่เหมาะสม
  3. ฝึกอัลกอริทึม LogisticRegression หลายรายการบน train ชุดข้อมูลด้วย hyperparameters ที่แตกต่างกัน
  4. คํานวณพื้นที่ภายใต้เส้นโค้ง ROC สําหรับแต่ละแบบจําลองที่ได้รับการฝึกและเลือกแบบจําลองที่มีเมตริกสูงสุดตามที่คํานวณบน test ชุดข้อมูล
  5. ประเมินแบบจําลองที่ดีที่สุดบน 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))

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

ตัวเลือก synapseml เกี่ยวข้องกับขั้นตอนที่ง่ายกว่า:

  1. ตัวTrainClassifierประมาณภายในประกอบด้วยข้อมูล ตราบใดที่คอลัมน์ที่เลือกใน traintestvalidation ชุดข้อมูลแสดงถึงคุณลักษณะ

  2. ตัว FindBestModel ประมาณการค้นหาแบบจําลองที่ดีที่สุดจากกลุ่มแบบจําลองที่ได้รับการฝึก เมื่อต้องการทําเช่นนี้ จะค้นหาแบบจําลองที่ทํางานได้ดีที่สุดบน test ชุดข้อมูลที่กําหนดเมตริกที่ระบุ

  3. ComputeModelStatisticsตัวแปลงจะคํานวณเมตริกที่แตกต่างกันบนชุดข้อมูลที่ให้คะแนน (ในกรณีvalidationของเรา ชุดข้อมูล) ในเวลาเดียวกัน

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

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