تصویری خصوصیت نکالنے کے لیے اپاچی اسپارک کے UDFs کے استعمال کے ساتھ SparkContext کے مسائل کو حل کرنا

اپاچی اسپارک کے UDFs میں SparkContext کی غلطیوں کے پیچھے اسرار سے پردہ اٹھانا

کے ساتھ کام کرنا اپاچی اسپارک اور PySpark میں اکثر بڑے پیمانے پر ڈیٹا کے کاموں کو سنبھالنے کے لیے تقسیم شدہ کمپیوٹنگ کا استعمال شامل ہوتا ہے۔ لیکن بعض اوقات، چیزیں منصوبہ بندی کے مطابق نہیں ہوتیں۔ ایک عام خرابی جس کا سامنا بہت سے ڈیٹا سائنسدانوں کو ہوتا ہے، خاص طور پر کال کرتے وقت یوزر ڈیفائنڈ فنکشنز (UDFs), بدنام زمانہ "SparkContext صرف ڈرائیور پر استعمال کیا جا سکتا ہے" غلطی ہے۔

یہ خرابی خاص طور پر مایوس کن ہوسکتی ہے جب امیج پروسیسنگ جیسے پیچیدہ کام انجام دیتے ہیں، جہاں کام متعدد کارکنوں میں تقسیم ہوتے ہیں۔ تصویر کی خصوصیت نکالنے جیسے منظرناموں میں، یہ سمجھنا کہ SparkContext اس طرح کیوں برتاؤ کرتا ہے۔ 💻

اس مضمون میں، میں آپ کو PyTorch میں ResNet ماڈل کو شامل کرنے والی ایک مثال کے ذریعے لے جاؤں گا۔ ہم اس بات کی کھوج کریں گے کہ UDF کے اندر آپریشنز کو سیریلائز کرنے کی کوشش کرتے وقت SparkContext کیوں مسائل پیدا کرتا ہے، جس سے رن ٹائم خرابی ہوتی ہے۔ اس کے ذریعے، میں اسپارک کے ساتھ ہموار ڈیٹا پروسیسنگ کو فعال کرنے کے لیے خامی کو دور کرنے کے لیے حکمت عملیوں کا اشتراک بھی کروں گا۔

اگر آپ کو اسپارک میں ایم ایل پائپ لائن بناتے وقت اس مسئلے کا سامنا کرنا پڑا ہے، تو آپ اکیلے نہیں ہیں! میرے ساتھ رہیں کیونکہ ہم اس غلطی سے بچنے اور تقسیم شدہ ماحول میں Spark UDFs کے ہموار آپریشن کو یقینی بنانے کے لیے عملی حل تلاش کر رہے ہیں۔ 🚀

ڈیپ لرننگ ماڈلز کے ساتھ اسپارک یو ڈی ایف سیریلائزیشن کا ٹربل شوٹنگ

کے ساتھ کام کرتے وقت اپاچی اسپارکتقسیم شدہ پروسیسنگ کا استعمال اکثر کاموں کو تیز کرنے کے لیے کیا جاتا ہے، خاص طور پر بڑے پیمانے پر امیج پروسیسنگ جیسے کاموں میں۔ تاہم، اسپارک کچھ پابندیاں عائد کرتا ہے، خاص طور پر اس پر SparkContext. اوپر دیے گئے اسکرپٹس میں، ResNet ڈیپ لرننگ ماڈل کو UDF کے اندر ڈیٹا فریم میں ہر قطار کے لیے امیجز سے فیچر نکالنے کے لیے استعمال کیا جاتا ہے۔ یہ نقطہ نظر SparkContext کی حد کو مارتا ہے: SparkContext صرف ڈرائیور نوڈ پر استعمال کیا جا سکتا ہے نہ کہ ورکر نوڈس پر چلنے والے کوڈ کے اندر، یہی وجہ ہے کہ کوڈ میں غلطی ہوتی ہے۔ ابتدائی حل میں اسپارک سیشن، امیج پری پروسیسنگ، اور فیچر نکالنے کے لیے امیج ویکٹرائزر کلاس بنانا شامل ہے۔ ان کاموں کو ایک کلاس میں سنٹرلائز کر کے، ہم کوڈ کو ماڈیولر اور قابل موافق رکھنے کے قابل ہو جاتے ہیں۔ 💻

پہلی اسکرپٹ میں، امیج ویکٹرائزر کلاس اسپارک سیشن کو شروع کرتی ہے اور ایک مشہور ڈیپ لرننگ لائبریری PyTorch سے پہلے سے تربیت یافتہ ResNet ماڈل لوڈ کرتی ہے۔ لاگو تبدیلیوں کے سیٹ کے ساتھ، جس میں سائز تبدیل کرنا اور معمول بنانا شامل ہے، ہر تصویر کو ماڈل کے لیے ایک ہم آہنگ فارمیٹ میں تبدیل کیا جا سکتا ہے۔ extract_features طریقہ اس بات کی وضاحت کرتا ہے کہ ہر تصویر پر کس طرح عمل کیا جاتا ہے: پہلے، تصویر کو پڑھا جاتا ہے، پہلے سے پروسیس کیا جاتا ہے، پھر اعلیٰ سطح کے فیچر ویکٹرز کو نکالنے کے لیے ResNet ماڈل سے گزر جاتا ہے۔ تاہم، یہ نقطہ نظر SparkContext سیریلائزیشن کے مسئلے کو متاثر کرتا ہے کیونکہ UDF کارکن کے کاموں کے اندر براہ راست اسپارک اجزاء تک رسائی حاصل کرنے کی کوشش کرتا ہے۔ چونکہ PySpark تقسیم شدہ نوڈس پر چلانے کے لیے ResNet ماڈل کو سیریلائز نہیں کر سکتا، اس سے رن ٹائم کا مسئلہ پیدا ہوتا ہے۔

اس کو حل کرنے کے لیے، دوسرا طریقہ Spark's کا استعمال کرتا ہے۔ نشر متغیرات، جو ہر کارکن کو صرف ایک بار ڈیٹا یا اشیاء تقسیم کرتے ہیں۔ ResNet ماڈل کو براڈکاسٹ کرنا ماڈل کو ہر ورکر نوڈ پر اسٹور کرنے کی اجازت دیتا ہے اور ہر UDF کال میں دوبارہ شروع ہونے سے روکتا ہے۔ اس کے بعد براڈکاسٹ ماڈل کا حوالہ تصویری فیچر نکالنے کے دوران دیا جاتا ہے، جس سے سیٹ اپ زیادہ موثر اور توسیع پذیر ہوتا ہے۔ یہ طریقہ وسائل کے استعمال کو نمایاں طور پر کم کرتا ہے اور اس بات کو یقینی بنا کر SparkContext کی غلطی سے بچتا ہے کہ Spark صرف ڈرائیور پر ضروری اجزاء تک رسائی حاصل کرتا ہے، کارکنوں پر نہیں۔ براڈکاسٹ متغیرات خاص طور پر اس وقت کارآمد ہوتے ہیں جب متوازی طور پر بڑے ڈیٹاسیٹس کو پروسیسنگ کرتے ہوئے، دوسری اسکرپٹ کو تقسیم شدہ امیج فیچر نکالنے کے لیے مثالی بناتے ہیں۔

براڈکاسٹ ماڈل استعمال کرنے کے لیے UDF فنکشن کو ایڈجسٹ کرنے کے بعد، ہم ایک UDF کی وضاحت کرتے ہیں جو ڈیٹا فریم کی ہر قطار پر تبدیلیوں کا اطلاق کرتا ہے۔ اس بات کی تصدیق کرنے کے لیے کہ اسکرپٹ مختلف ماحول میں کام کرتی ہیں، ایک تیسرا اسکرپٹ یونٹ ٹیسٹنگ کے لیے فراہم کیا جاتا ہے۔ پی ٹیسٹ. یہ اسکرپٹ بائنری امیج ڈیٹا کو ہینڈل کرنے، ٹرانسفارمیشن پائپ لائن چلانے، اور صحیح سائز کے فیچر ویکٹر کو آؤٹ پٹ کرنے کی فنکشن کی صلاحیت کی جانچ کرتا ہے۔ ٹیسٹنگ تعیناتی سے پہلے ہر جزو کے فنکشن کی تصدیق کرکے وشوسنییتا کی ایک اور پرت کا اضافہ کرتی ہے۔ 📊 یونٹ ٹیسٹ تقسیم شدہ ماحول میں خاص طور پر قیمتی ہوتے ہیں، کیونکہ وہ اس بات کو یقینی بناتے ہیں کہ کوڈ میں ترمیم سے تمام نوڈس میں غیر ارادی مسائل متعارف نہیں ہوتے ہیں۔

حقیقی دنیا کی ایپلی کیشنز میں، یہ طریقے Spark کی پیچیدہ تصویری ڈیٹا کو متوازی طور پر سنبھالنے کی صلاحیت کو بڑھاتے ہیں، جس سے مشین لرننگ اور AI پروجیکٹس میں وسیع امیج ڈیٹاسیٹس کے ساتھ کام کرنا ممکن ہو جاتا ہے۔ براڈکاسٹ ماڈل، UDFs، اور ٹیسٹنگ فریم ورک ان ورک فلو کو بہتر بنانے میں اہم کردار ادا کرتے ہیں۔ یہ حل بڑے پیمانے پر ڈیٹا پروسیسنگ میں لچک، اسکیل ایبلٹی، اور قابل اعتماد لاتے ہیں — تقسیم شدہ مشین لرننگ پائپ لائنز میں مسلسل، اعلیٰ معیار کے نتائج حاصل کرنے کے لیے اہم ہیں۔

# Import required libraries
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import udf
from pyspark.sql.types import ArrayType, FloatType
from torchvision import models, transforms
from PIL import Image
import torch
import numpy as np
from io import BytesIO
# Define the class to initialize Spark session and ResNet model
class ImageVectorizer:
    def __init__(self):
        # Initialize SparkSession
        self.spark = SparkSession.builder.getOrCreate()
        # Load pre-trained ResNet model
        self.resnet_model = models.resnet50(pretrained=True)
        self.resnet_model.eval()
        # Define image transformation pipeline
        self.transform = transforms.Compose([
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                                 std=[0.229, 0.224, 0.225])
        ])
    def extract_features(self, image_binary):
        # Convert image binary to tensor and extract features
        image = Image.open(BytesIO(image_binary))
        image = self.transform(image).unsqueeze(0)
        with torch.no_grad():
            features = self.resnet_model(image)
        return features.squeeze().numpy().tolist()
    def process_images(self, image_df):
        # Register a non-Spark UDF to call extract_features function
        extract_features_udf = udf(lambda x: self.extract_features(x), ArrayType(FloatType()))
        return image_df.withColumn("features", extract_features_udf(image_df["content"]))

# Import required libraries
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import ArrayType, FloatType
from torchvision import models, transforms
from PIL import Image
import torch
import numpy as np
from io import BytesIO
# Initialize Spark session and broadcast model
spark = SparkSession.builder.getOrCreate()
resnet_model = models.resnet50(pretrained=True)
resnet_model.eval()
bc_resnet_model = spark.sparkContext.broadcast(resnet_model)
# Define transformation pipeline separately
transform = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],
                     std=[0.229, 0.224, 0.225])
])
# Define feature extraction function using broadcast model
def extract_features(image_binary):
    image = Image.open(BytesIO(image_binary))
    image = transform(image).unsqueeze(0)
    with torch.no_grad():
        features = bc_resnet_model.value(image)
    return features.squeeze().numpy().tolist()
# Register UDF
extract_features_udf = udf(extract_features, ArrayType(FloatType()))

# Import pytest for unit testing
import pytest
import numpy as np
@pytest.fixture
def mock_image_binary():
    # Provide a sample image in binary format
    with open('test_image.jpg', 'rb') as f:
        return f.read()
def test_extract_features(mock_image_binary):
    # Initialize ImageVectorizer and call extract_features function
    vectorizer = ImageVectorizer()
    result = vectorizer.extract_features(mock_image_binary)
    assert isinstance(result, list)
    assert len(result) == 2048

امیج پروسیسنگ کے لیے Spark UDFs کے ساتھ سیریلائزیشن چیلنجز پر قابو پانا

استعمال کرنے میں اہم چیلنجوں میں سے ایک اپاچی اسپارک جیسے اعلی درجے کے کاموں کے لیے تصویر پروسیسنگ یوزر ڈیفائنڈ فنکشنز (UDFs) کے ساتھ کام کرتے وقت ہموار سیریلائزیشن کو یقینی بنا رہا ہے۔ چونکہ اسپارک کو فطری طور پر تقسیم کیا جاتا ہے، اس لیے اسپارک UDFs کے اندر کاموں کو پروسیسنگ کے لیے ورکر نوڈس کو بھیجا جاتا ہے، جو مسائل پیدا کر سکتے ہیں اگر غیر سیریلائز کرنے والی اشیاء جیسے پیچیدہ مشین لرننگ ماڈلز شامل ہوں۔ مثال کے طور پر، PyTorch کا ResNet ماڈل مقامی طور پر سیریلائز نہیں کیا جا سکتا ہے، یعنی اسے Spark کے اندر احتیاط سے ہینڈلنگ کی ضرورت ہے تاکہ "SparkContext صرف ڈرائیور پر استعمال کیا جا سکتا ہے" کی خرابی سے بچا جا سکے۔

سیریلائزیشن ایک رکاوٹ بن جاتی ہے کیونکہ اسپارک UDF میں حوالہ کردہ تمام عناصر بشمول SparkContext کو براہ راست ورکر نوڈس میں تقسیم کرنے کی کوشش کرتا ہے۔ یہ حد اسی وجہ سے ہے کہ ہم ہر بار دوبارہ شروع کیے بغیر ResNet ماڈل کو موثر طریقے سے نوڈس پر شیئر کرنے کے لیے براڈکاسٹ متغیر کا استعمال کرتے ہیں۔ ایسے معاملات میں، broadcast() طریقہ ہر کارکن کو صرف پڑھنے کے لیے ڈیٹا تقسیم کرنے میں مدد کرتا ہے، جہاں اسپارک کی سیریلائزیشن پابندیوں کو متحرک کیے بغیر مقامی طور پر اس کا حوالہ دیا جا سکتا ہے۔ ماڈل کو نشر کرنے سے، ResNet وزن تمام نوڈس پر ڈیٹا کو نقل کیے بغیر فیچر نکالنے کے لیے قابل رسائی ہیں، میموری کے استعمال اور کارکردگی دونوں کو بڑھاتے ہیں۔ 🌍

یہ تکنیک امیج پروسیسنگ سے آگے تقسیم شدہ ML پائپ لائنوں کے لیے وسیع پیمانے پر لاگو ہوتی ہے۔ مثال کے طور پر، اگر آپ سفارشی نظام کو نافذ کر رہے تھے، تو آپ Spark سیریلائزیشن کی غلطیوں سے بچنے کے لیے صارف کی ترجیحات یا پہلے سے تربیت یافتہ ماڈلز کے بڑے ڈیٹا سیٹس کو نشر کر سکتے ہیں۔ اسی طرح، دیگر پری پروسیسنگ کاموں (جیسے ٹیکسٹ ویکٹرائزیشن یا آڈیو پروسیسنگ) کے لیے UDFs کا استعمال غیر سیریلائز ایبل اشیاء کو نشر کرنے سے بھی فائدہ اٹھاتا ہے، جس سے Spark کو ڈیٹا ڈپلیکیشن اوور ہیڈز کے بغیر انتہائی متوازی کاموں کو ہینڈل کرنے کی اجازت ملتی ہے۔ یہ طرز عمل اسپارک کو نفیس ایم ایل ورک فلوز کو سنبھالنے کے لیے کافی مضبوط بناتا ہے، جس سے ساختی اور غیر ساختہ ڈیٹا دونوں کاموں میں بڑے ڈیٹا سیٹس کے لیے درکار اسکیل ایبلٹی فراہم ہوتی ہے۔ 🚀