और कैसे PySpark आपको एक पेशेवर की तरह विशाल डेटासेट को संभालने में मदद कर सकता है

PyTorch और TensorFlow जैसे मशीन लर्निंग फ्रेमवर्क मॉडल बनाने के लिए अद्भुत हैं। लेकिन वास्तविकता यह है कि, जब वास्तविक दुनिया की परियोजनाओं की बात आती है - जहां आप विशाल डेटासेट के साथ काम कर रहे हैं - तो आपको सिर्फ एक अच्छे मॉडल से कहीं अधिक की आवश्यकता होती है। आपको उस सभी डेटा को कुशलतापूर्वक संसाधित करने और प्रबंधित करने का एक तरीका चाहिए। यहीं पर पाइस्पार्क जैसी वितरित कंप्यूटिंग, दिन बचाने के लिए आती है।

आइए देखें कि वास्तविक दुनिया की मशीन लर्निंग में बड़े डेटा को संभालने का मतलब PyTorch और TensorFlow से आगे जाना क्यों है, और PySpark आपको वहां पहुंचने में कैसे मदद करता है।
वास्तविक समस्या: बड़ा डेटा
अधिकांश एमएल उदाहरण जो आप ऑनलाइन देखते हैं वे छोटे, प्रबंधनीय डेटासेट का उपयोग करते हैं। आप पूरी चीज़ को मेमोरी में फिट कर सकते हैं, उसके साथ खेल सकते हैं और मिनटों में एक मॉडल को प्रशिक्षित कर सकते हैं। लेकिन वास्तविक दुनिया के परिदृश्यों में - जैसे क्रेडिट कार्ड धोखाधड़ी का पता लगाना, अनुशंसा प्रणाली, या वित्तीय पूर्वानुमान - आप लाखों या अरबों पंक्तियों से निपट रहे हैं। अचानक, आपका लैपटॉप या सर्वर इसे संभाल नहीं सकता।

यदि आप उस सभी डेटा को एक ही बार में PyTorch या TensorFlow में लोड करने का प्रयास करते हैं, तो चीजें खराब हो जाएंगी। ये ढाँचे मॉडल प्रशिक्षण के लिए डिज़ाइन किए गए हैं, न कि विशाल डेटासेट को कुशलतापूर्वक संभालने के लिए। यहीं पर वितरित कंप्यूटिंग महत्वपूर्ण हो जाती है।
PyTorch और TensorFlow पर्याप्त क्यों नहीं हैं
PyTorch और TensorFlow मॉडल बनाने और अनुकूलित करने के लिए बहुत अच्छे हैं, लेकिन बड़े पैमाने पर डेटा कार्यों से निपटने के दौरान वे कम पड़ जाते हैं। दो प्रमुख समस्याएँ:

• मेमोरी ओवरलोड: वे प्रशिक्षण से पहले संपूर्ण डेटासेट को मेमोरी में लोड करते हैं। यह छोटे डेटासेट के लिए काम करता है, लेकिन जब आपके पास टेराबाइट डेटा हो जाता है, तो खेल खत्म हो जाता है।
• कोई वितरित डेटा प्रोसेसिंग नहीं: PyTorch और TensorFlow वितरित डेटा प्रोसेसिंग को संभालने के लिए नहीं बनाए गए हैं। यदि आपके पास कई मशीनों में भारी मात्रा में डेटा फैला हुआ है, तो वे वास्तव में मदद नहीं करते हैं।

यह वह जगह है जहां पायस्पार्क चमकता है। इसे वितरित डेटा के साथ काम करने, आपके सिस्टम को क्रैश किए बिना बड़े पैमाने पर डेटासेट को संभालने के दौरान इसे कई मशीनों पर कुशलतापूर्वक संसाधित करने के लिए डिज़ाइन किया गया है।

वास्तविक दुनिया का उदाहरण: पाइस्पार्क के साथ क्रेडिट कार्ड धोखाधड़ी का पता लगाना
आइए एक उदाहरण पर गौर करें। मान लीजिए कि आप क्रेडिट कार्ड लेनदेन डेटा का उपयोग करके धोखाधड़ी का पता लगाने वाली प्रणाली पर काम कर रहे हैं। इस मामले में, हम कागल के एक लोकप्रिय डेटासेट का उपयोग करेंगे। इसमें 284,000 से अधिक लेनदेन शामिल हैं, और उनमें से 1% से भी कम धोखाधड़ी वाले हैं।

चरण 1: Google Colab में PySpark सेट करें
हम इसके लिए Google Colab का उपयोग करेंगे क्योंकि यह हमें न्यूनतम सेटअप के साथ PySpark चलाने की सुविधा देता है।

!pip install pyspark

इसके बाद, आवश्यक लाइब्रेरी आयात करें और एक स्पार्क सत्र शुरू करें।

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

पाइस्पार्क सत्र शुरू करें

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 सटीकता, रिकॉल और एफ1-स्कोर की गणना करने का कार्य

# 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

फिर आप इस मॉडल (कुछ केबी) को सहेज सकते हैं और इसे पाइस्पार्क पाइपलाइन में कहीं भी उपयोग कर सकते हैं

rf_model.save()

यही कारण है कि वास्तविक दुनिया के मशीन सीखने के कार्यों में बड़े डेटासेट से निपटने के दौरान पाइस्पार्क एक बड़ा अंतर बनाता है:

यह आसानी से स्केल करता है: PySpark समूहों में कार्यों को वितरित कर सकता है, जिससे आप मेमोरी से बाहर हुए बिना टेराबाइट डेटा संसाधित कर सकते हैं।
ऑन-द-फ्लाई डेटा प्रोसेसिंग: पायस्पार्क को संपूर्ण डेटासेट को मेमोरी में लोड करने की आवश्यकता नहीं है। यह आवश्यकतानुसार डेटा को संसाधित करता है, जो इसे और अधिक कुशल बनाता है।
तेज़ मॉडल प्रशिक्षण: वितरित कंप्यूटिंग के साथ, आप कई मशीनों में कम्प्यूटेशनल कार्यभार वितरित करके मॉडल को तेज़ी से प्रशिक्षित कर सकते हैं।
अंतिम विचार
मशीन लर्निंग मॉडल बनाने के लिए PyTorch और TensorFlow शानदार उपकरण हैं, लेकिन वास्तविक दुनिया, बड़े पैमाने के कार्यों के लिए, आपको और अधिक की आवश्यकता है। PySpark के साथ वितरित कंप्यूटिंग आपको विशाल डेटासेट को कुशलतापूर्वक संभालने, वास्तविक समय में डेटा संसाधित करने और अपनी मशीन लर्निंग पाइपलाइनों को स्केल करने की अनुमति देता है।

तो, अगली बार जब आप बड़े पैमाने पर डेटा के साथ काम कर रहे हों - चाहे वह धोखाधड़ी का पता लगाना हो, सिफारिश प्रणाली हो, या वित्तीय विश्लेषण हो - अपने प्रोजेक्ट को अगले स्तर पर ले जाने के लिए PySpark का उपयोग करने पर विचार करें।

पूर्ण कोड और परिणामों के लिए, इस नोटबुक को देखें। :
https://colab.research.google.com/drive/1W9naxNZirirLRodSEnHAUWevYd5LH8D4?authuser=5#scrollTo=odmodmqKcY23

__

मैं स्वप्निल हूं, बेझिझक अपनी टिप्पणियाँ परिणाम और विचार छोड़ें, या डेटा, सॉफ्टवेयर डेव गिग्स और नौकरियों के लिए मुझे पिंग करें - [email protected]