สร้างแบบจําลองการเรียนรู้ของเครื่องด้วย Apache Spark MLlib

ในบทความนี้ คุณจะได้เรียนรู้วิธีใช้ Apache Spark MLlib เพื่อสร้างแอปพลิเคชันแมชชีนเลิร์นนิงที่จัดการการวิเคราะห์เชิงคาดการณ์บนชุดข้อมูลที่เปิดอยู่ Azure Spark มีไลบรารีการเรียนรู้ของเครื่องอยู่แล้วภายใน ตัวอย่างนี้ใช้ การจัดประเภท ผ่านการถดถอยโลจิสติกส์

บทช่วยสอนนี้ครอบคลุมขั้นตอนเหล่านี้:

ไลบรารี SparkML และ MLlib Spark หลักมีโปรแกรมอรรถประโยชน์มากมายที่มีประโยชน์สําหรับงานการเรียนรู้ของเครื่อง ยูทิลิตี้เหล่านี้เหมาะสําหรับ:

• การจัดประเภท
• การจัดกลุ่มคลัสเตอร์
• การทดสอบสมมติฐานและการคํานวณสถิติตัวอย่าง
• การถดถอย
• การวิเคราะห์การแยกย่อยค่าเอกพจน์ (SVD) และการวิเคราะห์คอมโพเนนต์หลัก (PCA)
• การสร้างแบบจําลองหัวข้อ

Prerequisites

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

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

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

  

• หากจําเป็น ให้สร้างเลคเฮาส์ Microsoft Fabric ตามที่อธิบายไว้ใน สร้างเลคเฮาส์ใน Microsoft Fabric
• สร้างสมุดบันทึกใหม่ในพื้นที่ทํางานของคุณโดยการเลือก + จากนั้น สมุดบันทึก สําหรับข้อมูลเพิ่มเติม ให้ดูที่ การสร้างสมุดบันทึก

ทําความเข้าใจเกี่ยวกับการจัดประเภทและการถดถอยโลจิสติกส์

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

อัลกอริธึม การถดถอยโลจิสติกส์ มีประโยชน์สําหรับการจําแนกประเภท API การถดถอยโลจิสติกของ Spark มีประโยชน์สําหรับ การจัดประเภท ข้อมูลอินพุตแบบไบนารีลงในหนึ่งในสองกลุ่ม สําหรับข้อมูลเพิ่มเติมเกี่ยวกับการกระจายโลจิสติกส์ โปรดดูที่วิกิพีเดีย

การถดถอยโลจิสติกส์สร้าง ฟังก์ชันโลจิสติกส์ ที่คาดการณ์ความน่าจะเป็นที่เวกเตอร์อินพุตเป็นของกลุ่มใดกลุ่มหนึ่ง

ตัวอย่างการวิเคราะห์เชิงทํานายของข้อมูลแท็กซี่ของ NYC

ข้อมูลจะพร้อมใช้งานผ่านทรัพยากร Azure Open Datasets ชุดข้อมูลนี้โฮสต์ข้อมูลเกี่ยวกับการเดินทางด้วยรถแท็กซี่สีเหลือง รวมถึงเวลาเริ่มต้น เวลาสิ้นสุด ตําแหน่งเริ่มต้น ตําแหน่งสิ้นสุด ค่าใช้จ่ายการเดินทาง และแอตทริบิวต์อื่น ๆ

บทช่วยสอนนี้ใช้ Apache Spark เพื่อทําการวิเคราะห์ข้อมูลทิปทริปแท็กซี่ในนิวยอร์ก และพัฒนาแบบจําลองเพื่อคาดการณ์ว่าการเดินทางนั้นมีทิปหรือไม่

สร้างแบบจําลองการเรียนรู้ของเครื่อง Apache Spark

1. สร้างสมุดบันทึก PySpark สําหรับข้อมูลเพิ่มเติม ให้ดูที่ การสร้างสมุดบันทึก

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

2. นําเข้าชนิดที่จําเป็นสําหรับสมุดบันทึกนี้ วางรหัสต่อไปนี้ลงในเซลล์แรกแล้วเรียกใช้

   import matplotlib.pyplot as plt
   from pyspark.sql.functions import unix_timestamp, date_format, col, when
   from pyspark.ml import Pipeline
   from pyspark.ml.feature import RFormula
   from pyspark.ml.feature import OneHotEncoder, StringIndexer
   from pyspark.ml.classification import LogisticRegression
   from pyspark.ml.evaluation import BinaryClassificationEvaluator
   

   ยืนยัน: เซลล์เสร็จสมบูรณ์โดยไม่มี ImportError. หากคุณเห็นข้อผิดพลาด ให้ยืนยันว่าสมุดบันทึกของคุณใช้รันไทม์ PySpark

3. ใช้ MLflow เพื่อติดตามการทดลองแมชชีนเลิร์นนิงและการเรียกใช้ที่เกี่ยวข้อง ถ้าเปิดใช้งานการล็อกอัตโนมัติของ Microsoft Fabric เมตริกและพารามิเตอร์ที่สอดคล้องกันจะถูกบันทึกโดยอัตโนมัติ

   import mlflow
   

   ยืนยัน: เซลล์เสร็จสมบูรณ์โดยไม่มีข้อผิดพลาด เรียกใช้ print(mlflow.__version__) เพื่อยืนยันว่า MLflow พร้อมใช้งาน

สร้าง DataFrame อินพุต

ตัวอย่างนี้จะโหลดข้อมูลจากที่เก็บข้อมูล Azure Open Datasets ลงใน Apache Spark DataFrame จากนั้น คุณใช้การดําเนินการ Spark เพื่อล้างและกรองชุดข้อมูล

1. วางโค้ดต่อไปนี้ลงในเซลล์ใหม่และเรียกใช้เพื่อสร้าง Spark DataFrame ขั้นตอนนี้จะดึงข้อมูลแท็กซี่สีเหลืองของนิวยอร์กที่กรองเป็นเดือนพฤษภาคม 2018

   blob_account_name = "azureopendatastorage"
   blob_container_name = "nyctlc"
   blob_relative_path = "yellow"
   wasbs_path = f"wasbs://{blob_container_name}@{blob_account_name}.blob.core.windows.net/{blob_relative_path}"
   
   nyc_tlc_df = spark.read.parquet(wasbs_path) \
       .filter((col("tpepPickupDateTime") >= "2018-05-01") & (col("tpepPickupDateTime") < "2018-06-01")) \
       .repartition(20)
   

   ยืนยัน: เรียกใช้เซลล์ต่อไปนี้เพื่อยืนยันการโหลดข้อมูลสําเร็จ

   print(f"Loaded {nyc_tlc_df.count()} rows")
   # Expected output: Loaded approximately 9,000,000+ rows
   

2. สุ่มตัวอย่างชุดข้อมูลเพื่อเร่งการพัฒนาและการฝึกอบรม

   # Sample without replacement to avoid duplicates
   sampled_taxi_df = nyc_tlc_df.sample(False, 0.001, seed=1234)
   

   ยืนยัน: ยืนยันว่าขนาดตัวอย่างสามารถจัดการได้

   print(f"Sampled {sampled_taxi_df.count()} rows")
   # Expected output: Sampled approximately 9,000-10,000 rows
   

3. ดูข้อมูลโดยใช้คําสั่งที่มีอยู่แล้วภายใน display() เพื่อสํารวจตัวอย่างข้อมูล

   display(sampled_taxi_df.limit(10))
   

   ตรวจสอบ: ตารางที่มี 10 แถวจะปรากฏขึ้นโดยแสดงคอลัมน์เช่น tpepPickupDateTime, fareAmount, และ tipAmounttripDistance

เตรียมข้อมูล

การเตรียมข้อมูลเป็นขั้นตอนสําคัญในกระบวนการเรียนรู้ของเครื่อง มันเกี่ยวข้องกับการทําความสะอาด แปลง และจัดระเบียบข้อมูลดิบเพื่อให้เหมาะสําหรับการวิเคราะห์และการสร้างแบบจําลอง ในส่วนนี้ ให้ทําตามขั้นตอนการเตรียมข้อมูลหลายขั้นตอน:

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

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

taxi_df = sampled_taxi_df.select('totalAmount', 'fareAmount', 'tipAmount', 'paymentType', 'rateCodeId', 'passengerCount',
                    'tripDistance', 'tpepPickupDateTime', 'tpepDropoffDateTime',
                    date_format('tpepPickupDateTime', 'HH').cast('integer').alias('pickupHour'),
                    date_format('tpepPickupDateTime', 'EEEE').alias('weekdayString'),
                    (unix_timestamp(col('tpepDropoffDateTime')) - unix_timestamp(col('tpepPickupDateTime'))).alias('tripTimeSecs'),
                    (when(col('tipAmount') > 0, 1).otherwise(0)).alias('tipped')
                    ) \
            .filter((sampled_taxi_df.passengerCount > 0) & (sampled_taxi_df.passengerCount < 8)
                    & (sampled_taxi_df.tipAmount >= 0) & (sampled_taxi_df.tipAmount <= 25)
                    & (sampled_taxi_df.fareAmount >= 1) & (sampled_taxi_df.fareAmount <= 250)
                    & (sampled_taxi_df.tipAmount < sampled_taxi_df.fareAmount)
                    & (sampled_taxi_df.tripDistance > 0) & (sampled_taxi_df.tripDistance <= 100)
                    & (sampled_taxi_df.rateCodeId <= 5)
                    & (sampled_taxi_df.paymentType.isin({"1", "2"}))
                    )

จากนั้น ให้เพิ่มฟีเจอร์ถังขยะเวลาการจราจรตามชั่วโมงของวัน:

taxi_featurised_df = taxi_df.select('totalAmount', 'fareAmount', 'tipAmount', 'paymentType', 'passengerCount',
                                    'tripDistance', 'weekdayString', 'pickupHour', 'tripTimeSecs', 'tipped',
                                    when((col('pickupHour') <= 6) | (col('pickupHour') >= 20), "Night")
                                    .when((col('pickupHour') >= 7) & (col('pickupHour') <= 10), "AMRush")
                                    .when((col('pickupHour') >= 11) & (col('pickupHour') <= 15), "Afternoon")
                                    .when((col('pickupHour') >= 16) & (col('pickupHour') <= 19), "PMRush")
                                    .otherwise("Other").alias('trafficTimeBins')
                                    ) \
                            .filter((taxi_df.tripTimeSecs >= 30) & (taxi_df.tripTimeSecs <= 7200))

ยืนยัน: ยืนยันว่ามีการกระจายช่องเวลาการรับส่งข้อมูลอย่างถูกต้อง

taxi_featurised_df.groupBy('trafficTimeBins').count().show()
# Expected output: Shows counts for Night, AMRush, Afternoon, PMRush categories

สร้างแบบจําลองการถดถอยโลจิสติกส์

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

แปลงคอลัมน์ trafficTimeBins ตามหมวดหมู่และ weekdayString เป็นการแสดงจํานวนเต็มโดยใช้ OneHotEncoder วิธีการ:

# Convert categorical features into numeric representations
sI1 = StringIndexer(inputCol="trafficTimeBins", outputCol="trafficTimeBinsIndex")
en1 = OneHotEncoder(inputCol="trafficTimeBinsIndex", outputCol="trafficTimeBinsVec")
sI2 = StringIndexer(inputCol="weekdayString", outputCol="weekdayIndex")
en2 = OneHotEncoder(inputCol="weekdayIndex", outputCol="weekdayVec")

# Apply the encodings to create a new DataFrame
encoded_final_df = Pipeline(stages=[sI1, en1, sI2, en2]).fit(taxi_featurised_df).transform(taxi_featurised_df)

ยืนยัน: ยืนยันว่า DataFrame ที่เข้ารหัสมีคอลัมน์ใหม่ที่คาดไว้

print("Columns:", encoded_final_df.columns)
print(f"Row count: {encoded_final_df.count()}")
# Expected output: Columns list includes 'trafficTimeBinsVec' and 'weekdayVec'

ฝึกแบบจําลองการถดถอยโลจิสติกส์

แบ่งชุดข้อมูลออกเป็นชุดการฝึกอบรม (70%) และชุดทดสอบ (30%):

# Split the DataFrame into training and test sets
trainingFraction = 0.7
testingFraction = (1 - trainingFraction)
seed = 1234

train_data_df, test_data_df = encoded_final_df.randomSplit([trainingFraction, testingFraction], seed=seed)

ยืนยัน: ยืนยันว่าการแยกมีขนาดที่เหมาะสม

print(f"Training rows: {train_data_df.count()}, Test rows: {test_data_df.count()}")
# Expected output: Approximately 70%/30% split of the encoded data

สร้างสูตรแบบจําลอง ฝึกแบบจําลองการถดถอยโลจิสติกส์ และประเมินโดยใช้ Area Under the ROC (Receiver Operating Characteristic) Curve:

# Create a logistic regression model
logReg = LogisticRegression(maxIter=10, regParam=0.3, labelCol='label')

# Define the formula: 'tipped' is the response variable, right-hand side are predictors
classFormula = RFormula(formula="tipped ~ pickupHour + weekdayVec + passengerCount + tripTimeSecs + tripDistance + fareAmount + paymentType + trafficTimeBinsVec")

# Train the model using a pipeline
lrModel = Pipeline(stages=[classFormula, logReg]).fit(train_data_df)

# Generate predictions on the test dataset
predictions = lrModel.transform(test_data_df)

# Evaluate using Area Under ROC
evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", metricName="areaUnderROC")
auc = evaluator.evaluate(predictions)
print(f"Area under ROC = {auc}")

ตรวจสอบ: เอาต์พุตแสดงค่า AUC โมเดลที่มีประสิทธิภาพดีจะสร้างค่าที่ใกล้เคียงกับ 1.0

Area under ROC = 0.97 (approximately)

สร้างการแสดงภาพของการคาดการณ์

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

# Plot the ROC curve from the model training summary
modelSummary = lrModel.stages[-1].summary

# Extract FPR and TPR values as plain lists
roc_data = modelSummary.roc.select('FPR', 'TPR').toPandas()

plt.figure(figsize=(8, 6))
plt.plot([0, 1], [0, 1], 'r--', label='Random classifier')
plt.plot(roc_data['FPR'], roc_data['TPR'], label=f'Logistic Regression (AUC = {auc:.4f})')
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
plt.title('ROC Curve - NYC Taxi Tip Prediction')
plt.legend(loc='lower right')
plt.show()

ยืนยัน: พล็อตปรากฏขึ้นโดยแสดงเส้นโค้ง ROC เหนือเส้นทแยงมุมประสีแดง เส้นโค้งควรโค้งไปทางมุมซ้ายบน ซึ่งบ่งบอกถึงประสิทธิภาพการจําแนกประเภทที่แข็งแกร่ง

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

หลังจากที่คุณทําบทช่วยสอนนี้เสร็จแล้ว ให้ลบสมุดบันทึกและเลคเฮาส์เพื่อเพิ่มความจุพื้นที่ทํางาน:

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

เมื่อต้องการรักษาโมเดลที่ผ่านการฝึกอบรมไว้สําหรับการใช้งานในอนาคต ให้เพิ่มโค้ดต่อไปนี้ก่อนล้างข้อมูล:

# Save the model to the lakehouse
model_path = "abfss://<your-workspace>@onelake.dfs.fabric.microsoft.com/<your-lakehouse>.Lakehouse/Files/models/taxi_tip_model"
lrModel.write().overwrite().save(model_path)
print(f"Model saved to: {model_path}")

แก้ไข ปัญหา

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

• ใช้ตัวอย่าง AI เพื่อสร้างแบบจําลองการเรียนรู้ของเครื่อง: ใช้ตัวอย่าง AI
• ติดตามการทํางานของการเรียนรู้ของเครื่องโดยใช้การทดลอง: การทดลองการเรียนรู้ของเครื่อง