وكيف يمكن أن يساعدك PySpark في التعامل مع مجموعات البيانات الضخمة مثل المحترفين

تعتبر أطر التعلم الآلي مثل PyTorch و TensorFlow رائعة لبناء النماذج. ولكن الحقيقة هي أنه عندما يتعلق الأمر بمشاريع العالم الحقيقي - حيث تتعامل مع مجموعات بيانات ضخمة - فإنك تحتاج إلى أكثر من مجرد نموذج جيد. أنت بحاجة إلى طريقة لمعالجة وإدارة كل تلك البيانات بكفاءة. وهنا يأتي دور الحوسبة الموزعة، مثل PySpark، لإنقاذ الموقف.

دعنا نوضح لماذا يعني التعامل مع البيانات الضخمة في التعلم الآلي في العالم الحقيقي تجاوز PyTorch و TensorFlow، وكيف تساعدك PySpark على الوصول إلى هناك.
المشكلة الحقيقية: البيانات الضخمة
تستخدم معظم أمثلة تعلم الآلة التي تراها عبر الإنترنت مجموعات بيانات صغيرة يمكن التحكم فيها. يمكنك حفظ كل شيء في ذاكرتك، والتلاعب به، وتدريب نموذج في دقائق. ولكن في سيناريوهات العالم الحقيقي - مثل الكشف عن الاحتيال في بطاقات الائتمان، أو أنظمة التوصيات، أو التوقعات المالية - فإنك تتعامل مع ملايين أو حتى مليارات الصفوف. وفجأة، لا يستطيع الكمبيوتر المحمول أو الخادم التعامل معه.

إذا حاولت تحميل كل تلك البيانات إلى PyTorch أو TensorFlow مرة واحدة، فسوف تتعطل الأمور. تم تصميم هذه الأطر للتدريب النموذجي، وليس للتعامل بكفاءة مع مجموعات البيانات الضخمة. وهنا تصبح الحوسبة الموزعة أمرًا بالغ الأهمية.
لماذا PyTorch و TensorFlow ليسا كافيين
يعد كل من PyTorch وTensorFlow رائعين لبناء النماذج وتحسينها، إلا أنهما يعجزان عن التعامل مع مهام البيانات واسعة النطاق. مشكلتان رئيسيتان:

• التحميل الزائد للذاكرة: يقومون بتحميل مجموعة البيانات بأكملها في الذاكرة قبل التدريب. ينجح هذا مع مجموعات البيانات الصغيرة، ولكن عندما يكون لديك تيرابايت من البيانات، تنتهي اللعبة.
• لا توجد معالجة للبيانات الموزعة: لم يتم تصميم PyTorch و TensorFlow للتعامل مع معالجة البيانات الموزعة. إذا كان لديك كميات هائلة من البيانات المنتشرة عبر أجهزة متعددة، فلن تجدي نفعًا حقًا.

هذا هو المكان الذي يتألق فيه PySpark. لقد تم تصميمه للعمل مع البيانات الموزعة ومعالجتها بكفاءة عبر أجهزة متعددة أثناء التعامل مع مجموعات البيانات الضخمة دون تعطل نظامك.

مثال من العالم الحقيقي: اكتشاف الاحتيال على بطاقة الائتمان باستخدام PySpark
دعونا نتعمق في مثال. لنفترض أنك تعمل على نظام للكشف عن الاحتيال باستخدام بيانات معاملات بطاقة الائتمان. في هذه الحالة، سنستخدم مجموعة بيانات شائعة من Kaggle. يحتوي على أكثر من 284.000 معاملة، وأقل من 1% منها احتيالية.

الخطوة 1: إعداد PySpark في Google Colab
سنستخدم Google Colab لهذا الغرض لأنه يتيح لنا تشغيل PySpark بأقل قدر من الإعداد.

!pip install pyspark

بعد ذلك، قم باستيراد المكتبات الضرورية وابدأ جلسة Spark.

import os
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, udf
from pyspark.ml.feature import VectorAssembler, StringIndexer, MinMaxScaler
from pyspark.ml.classification import RandomForestClassifier, GBTClassifier
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator
from pyspark.ml.linalg import Vectors
import numpy as np
from pyspark.sql.types import FloatType

بدء جلسة pyspark

spark = SparkSession.builder \
    .appName("FraudDetectionImproved") \
    .master("local[*]") \
    .config("spark.executorEnv.PYTHONHASHSEED", "0") \
    .getOrCreate()

الخطوة 2: تحميل البيانات وإعدادها

data = spark.read.csv('creditcard.csv', header=True, inferSchema=True)
data = data.orderBy("Time")  # Ensure data is sorted by time
data.show(5)
data.describe().show()

# Check for missing values in each column
data.select([sum(col(c).isNull().cast("int")).alias(c) for c in data.columns]).show()

# Prepare the feature columns
feature_columns = data.columns
feature_columns.remove("Class")  # Removing "Class" column as it is our label

# Assemble features into a single vector
assembler = VectorAssembler(inputCols=feature_columns, outputCol="features")
data = assembler.transform(data)
data.select("features", "Class").show(5)

# Split data into train (60%), test (20%), and unseen (20%)
train_data, temp_data = data.randomSplit([0.6, 0.4], seed=42)
test_data, unseen_data = temp_data.randomSplit([0.5, 0.5], seed=42)

# Print class distribution in each dataset
print("Train Data:")
train_data.groupBy("Class").count().show()

print("Test and parameter optimisation Data:")
test_data.groupBy("Class").count().show()

print("Unseen Data:")
unseen_data.groupBy("Class").count().show()

الخطوة 3: تهيئة النموذج

# Initialize RandomForestClassifier
rf = RandomForestClassifier(labelCol="Class", featuresCol="features", probabilityCol="probability")

# Create ParamGrid for Cross Validation
paramGrid = ParamGridBuilder() \
    .addGrid(rf.numTrees, [10, 20 ]) \
    .addGrid(rf.maxDepth, [5, 10]) \
    .build()

# Create 5-fold CrossValidator
crossval = CrossValidator(estimator=rf,
                          estimatorParamMaps=paramGrid,
                          evaluator=BinaryClassificationEvaluator(labelCol="Class", metricName="areaUnderROC"),
                          numFolds=5)

الخطوة 4: الملاءمة، وتشغيل التحقق المتبادل، واختيار أفضل مجموعة من المعلمات

# Run cross-validation, and choose the best set of parameters
rf_model = crossval.fit(train_data)

# Make predictions on test data
predictions_rf = rf_model.transform(test_data)

# Evaluate Random Forest Model
binary_evaluator = BinaryClassificationEvaluator(labelCol="Class", rawPredictionCol="rawPrediction", metricName="areaUnderROC")
pr_evaluator = BinaryClassificationEvaluator(labelCol="Class", rawPredictionCol="rawPrediction", metricName="areaUnderPR")

auc_rf = binary_evaluator.evaluate(predictions_rf)
auprc_rf = pr_evaluator.evaluate(predictions_rf)
print(f"Random Forest - AUC: {auc_rf:.4f}, AUPRC: {auprc_rf:.4f}")

# UDF to extract positive probability from probability vector
extract_prob = udf(lambda prob: float(prob[1]), FloatType())
predictions_rf = predictions_rf.withColumn("positive_probability", extract_prob(col("probability")))

الخطوة 5 وظيفة لحساب الدقة والاستدعاء ودرجة F1

# Function to calculate precision, recall, and F1-score
def calculate_metrics(predictions):
    tp = predictions.filter((col("Class") == 1) & (col("prediction") == 1)).count()
    fp = predictions.filter((col("Class") == 0) & (col("prediction") == 1)).count()
    fn = predictions.filter((col("Class") == 1) & (col("prediction") == 0)).count()

    precision = tp / (tp   fp) if (tp   fp) != 0 else 0
    recall = tp / (tp   fn) if (tp   fn) != 0 else 0
    f1_score = (2 * precision * recall) / (precision   recall) if (precision   recall) != 0 else 0

    return precision, recall, f1_score

الخطوة 6: ابحث عن أفضل عتبة للنموذج

# Find the best threshold for the model
best_threshold = 0.5
best_f1 = 0
for threshold in np.arange(0.1, 0.9, 0.1):
    thresholded_predictions = predictions_rf.withColumn("prediction", (col("positive_probability") > threshold).cast("double"))
    precision, recall, f1 = calculate_metrics(thresholded_predictions)

    if f1 > best_f1:
        best_f1 = f1
        best_threshold = threshold

print(f"Best threshold: {best_threshold}, Best F1-score: {best_f1:.4f}")

الخطوة 7: التقييم على البيانات غير المرئية

# Evaluate on unseen data
predictions_unseen = rf_model.transform(unseen_data)
auc_unseen = binary_evaluator.evaluate(predictions_unseen)
print(f"Unseen Data - AUC: {auc_unseen:.4f}")

precision, recall, f1 = calculate_metrics(predictions_unseen)
print(f"Unseen Data - Precision: {precision:.4f}, Recall: {recall:.4f}, F1-score: {f1:.4f}")

area_under_roc = binary_evaluator.evaluate(predictions_unseen)
area_under_pr = pr_evaluator.evaluate(predictions_unseen)
print(f"Unseen Data - AUC: {area_under_roc:.4f}, AUPRC: {area_under_pr:.4f}")

نتائج

Best threshold: 0.30000000000000004, Best F1-score: 0.9062
Unseen Data - AUC: 0.9384
Unseen Data - Precision: 0.9655, Recall: 0.7568, F1-score: 0.8485
Unseen Data - AUC: 0.9423, AUPRC: 0.8618

يمكنك بعد ذلك حفظ هذا النموذج (بضعة كيلو بايت) واستخدامه في أي مكان في خط أنابيب pyspark

rf_model.save()

هذا هو السبب الذي يجعل PySpark يُحدث فرقًا كبيرًا عند التعامل مع مجموعات البيانات الكبيرة في مهام التعلم الآلي في العالم الحقيقي:

يمكن توسيع نطاقه بسهولة: يمكن لـ PySpark توزيع المهام عبر المجموعات، مما يسمح لك بمعالجة تيرابايت من البيانات دون نفاد الذاكرة.
معالجة البيانات أثناء التنقل: لا يحتاج PySpark إلى تحميل مجموعة البيانات بأكملها في الذاكرة. فهو يعالج البيانات حسب الحاجة، مما يجعلها أكثر كفاءة.
تدريب أسرع للنماذج: باستخدام الحوسبة الموزعة، يمكنك تدريب النماذج بشكل أسرع من خلال توزيع عبء العمل الحسابي عبر أجهزة متعددة.
الأفكار النهائية
تعد PyTorch وTensorFlow أدوات رائعة لبناء نماذج التعلم الآلي، ولكن بالنسبة للمهام الواقعية واسعة النطاق، فأنت بحاجة إلى المزيد. تتيح لك الحوسبة الموزعة باستخدام PySpark التعامل مع مجموعات البيانات الضخمة بكفاءة، ومعالجة البيانات في الوقت الفعلي، وتوسيع نطاق مسارات التعلم الآلي لديك.

لذا، في المرة القادمة التي تتعامل فيها مع بيانات ضخمة - سواء كان ذلك كشف الاحتيال، أو أنظمة التوصية، أو التحليل المالي - فكر في استخدام PySpark للارتقاء بمشروعك إلى المستوى التالي.

للحصول على الكود الكامل والنتائج، راجع هذا الدفتر. :
https://colab.research.google.com/drive/1W9naxNZirirLRodSEnHAUWevYd5LH8D4?authuser=5#scrollTo=odmodmqKcY23

__

أنا Swapnil، لا تتردد في ترك تعليقاتك النتائج والأفكار، أو الاتصال بي - [email protected] للبيانات ووظائف مطوري البرامج والوظائف