Spark जॉब चलाने पर असल में क्या होता है?

परिचय

डेटासेट बढ़ने के साथ, CPU क्षमता, मेमोरी और स्टोरेज थ्रूपुट की सीमाओं के कारण एक ही मशीन पर उनका कुशलतापूर्वक प्रसंस्करण करना चुनौतीपूर्ण हो सकता है।

Apache Spark एक वितरित डेटा प्रसंस्करण इंजन उपलब्ध कराकर इन चुनौतियों का समाधान करता है। यह कार्यभार को विभाजनों में बाँट सकता है और CPU कोर तथा कई मशीनों पर कार्य समानांतर रूप से चला सकता है।

हालाँकि, Spark का मतलब सिर्फ़ और मशीनें जोड़ना नहीं है।

कुशल और भरोसेमंद डेटा पाइपलाइन तैयार करने के लिए यह समझना ज़रूरी है कि इसके घटक—ड्राइवर, एक्ज़ीक्यूटर, RDD, DataFrame, विभाजन, जॉब, स्टेज और टास्क—मिलकर कैसे काम करते हैं।

इस लेख में, मैं ई-कॉमर्स बिक्री के समेकन के एक व्यावहारिक उदाहरण के ज़रिए Spark की संरचना, निष्पादन की कार्यप्रणाली और प्रदर्शन से जुड़े प्रमुख पहलुओं पर चर्चा करूँगा।

1. Apache Spark क्यों—समानांतर डेटा प्रसंस्करण

Apache Spark एक वितरित कंप्यूटिंग इंजन है, जिसे बड़े पैमाने पर डेटा प्रसंस्करण और डेटा इंजीनियरिंग के लिए बनाया गया है।

Spark समानांतर डेटा प्रसंस्करण के ज़रिए बड़े डेटा को संसाधित करता है:

आकृति A. Apache Spark के वितरित क्लस्टर की उच्च-स्तरीय संरचना (क्लस्टर, ड्राइवर और वर्कर नोड), जिसमें एकल नोड पर निष्पादन की तुलना की गई है (Kuriko IWAI द्वारा निर्मित)

एकल नोड पर प्रसंस्करण (दाईं ओर, आकृति A) की तुलना में, Spark काम को टास्क में बाँटता है और उन्हें कई नोड पर उपलब्ध संसाधनों के ज़रिए मशीनों में निष्पादित करता है।

यहाँ, वितरित होने का यह अर्थ नहीं है कि कई मशीनें अनिवार्य हैं। Spark एक ही मशीन पर स्थानीय मोड में चल सकता है। वहाँ भी डेटा को विभाजित करके समानांतर रूप से संसाधित किया जा सकता है।

2. Apache Spark की मूल संरचना

Spark में कई प्रमुख घटक शामिल हैं:

• Spark ड्राइवर (ड्राइवर): Spark एप्लिकेशन के समन्वय की केंद्रीय प्रक्रिया।
• क्लस्टर मैनेजर: CPU कोर और मेमोरी जैसे हार्डवेयर संसाधन आवंटित करने के लिए ज़िम्मेदार एक बाहरी सेवा।
• वर्कर नोड (वर्कर): क्लस्टर में एक भौतिक/आभासी मशीन या कंप्यूट नोड।
• एक्ज़ीक्यूटर: वर्कर नोड पर शुरू की जाने वाली वितरित वर्कर प्रक्रियाएँ।
• Java Virtual Machine (JVM): Spark के बाइटकोड को चलाने के लिए ज़रूरी रनटाइम परिवेश (क्योंकि Spark, Scala/Java में लिखा गया है)।

सबसे पहले, ड्राइवर Spark एप्लिकेशन का समन्वय करता है और उसे भौतिक निष्पादन टास्क के निर्देशित अचक्रीय ग्राफ़ (DAG) में बदलता है। फिर वह इन्हें एक्ज़ीक्यूटर में शेड्यूल करता है।

इसके बाद, क्लस्टर मैनेजर परिनियोजन की संसाधन-प्रबंधन प्रणाली के तहत संसाधन आवंटित करता है।

वर्कर नोड एक्ज़ीक्यूटर को कंप्यूट संसाधन उपलब्ध कराते हैं। एक्ज़ीक्यूटर टास्क चलाते हैं और उनके निष्पादन के दौरान डेटा को अपने पास रखते हैं।

Databricks जैसे कुछ प्रबंधित परिवेशों में हर वर्कर के लिए एक एक्ज़ीक्यूटर होता है, लेकिन यह Spark का कोई सार्वभौमिक नियम नहीं है। क्लस्टर कॉन्फ़िगरेशन के अनुसार, एक वर्कर पर कई एक्ज़ीक्यूटर हो सकते हैं।

2-1. ई-कॉमर्स का उदाहरण—इनपुट डेटा की संरचना

एक छोटे ई-कॉमर्स व्यवसाय की कल्पना करें।

उसके पास बिक्री के 10,000 रिकॉर्ड हैं, जिनमें से हर रिकॉर्ड में स्टॉक कीपिंग यूनिट (SKU) और बेची गई मात्रा दर्ज है। मान लें कि तीन संभावित SKU हैं: A, B, और C.

तालिका 1. ई-कॉमर्स बिक्री रिकॉर्ड का विवरण और अपेक्षित समेकन का उदाहरण।

हमारा लक्ष्य है हर SKU के अनुसार बेची गई कुल मान्य इकाइयों की गणना करना:

1. ऐसे रिकॉर्ड फ़िल्टर करें जिनके qty <= 0, फिर
2. SKU के अनुसार नमूनों का समेकन करें।

पूरे डेटासेट के इस उदाहरण में, मान लें कि फ़िल्टर करने के बाद 10,000 में से 9,800 रिकॉर्ड बचते हैं। Spark से अपेक्षित अंतिम कुल इस प्रकार हैं:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Spark क्लस्टर

मैं इस उदाहरण के लिए एक सरलीकृत क्लस्टर तय करूँगा।

वर्कर नोड (वर्कर)

मैं 3 वर्कर नोड तय करूँगा, जिनमें से हर एक का इंस्टेंस प्रकार 8 vCPU और 32 GiB RAM वाला है:

आकृति B. वर्कर नोड में संसाधनों का आवंटन (Kuriko IWAI द्वारा निर्मित)

कुल मिलाकर, सिस्टम में 24 vCPU उपलब्ध हैं।

Spark एक्ज़ीक्यूटर और CPU/मेमोरी आवंटन

मान लें, मैं ये संसाधन तय करता हूँ:

• 7 एक्ज़ीक्यूटर।
• हर एक्ज़ीक्यूटर के लिए 2 vCPU कोर।
• हर एक्ज़ीक्यूटर के लिए 3GiB JVM हीप मेमोरी।

Spark एप्लिकेशन, क्लस्टर मैनेजर से इन 7 एक्ज़ीक्यूटर को संसाधन आवंटित करने का अनुरोध करता है।

इसके बाद, क्लस्टर मैनेजर हर वर्कर के संसाधन उपयोग के आधार पर CPU और मेमोरी आवंटन तय करता है।

उदाहरण के लिए, वह 14 टास्क स्लॉट बनाने का निर्णय लेता है—हर एक्ज़ीक्यूटर के लिए 2 vCPU कोर × 7 एक्ज़ीक्यूटर:

चित्र C-1. वर्कर A में संसाधनों के असंतुलन के साथ वर्करों में vCPU कोर, मेमोरी और एक्ज़ीक्यूटर का आवंटन

ध्यान दें कि एक्ज़ीक्यूटर को आवंटित 3 GiB मेमोरी उसकी कुल मेमोरी नहीं है, क्योंकि उसे वर्करों और अन्य प्रक्रियाओं के लिए अतिरिक्त मेमोरी भी सुरक्षित रखनी होती है।

यह मानते हुए कि एक टास्क के लिए 1 vCPU (spark.task.cpus = 1), हर वर्कर अपने vCPU इस तरह आवंटित करता है:

चित्र C-2. वर्करों में समानांतर vCPU आवंटन

इससे दो अहम बातें सामने आती हैं:

1. हर वर्कर की CPU उपयोग दर अलग होती है (चित्र C-2 में, वर्कर A का उपयोग 75% है, जबकि वर्कर B और C का उपयोग 50% पर बना रहता है)।
2. हालाँकि सिस्टम में 24 vCPU कोर हैं (प्रति वर्कर 8 vCPU × 3 वर्कर), लेकिन यह केवल 14 vCPU कोर का उपयोग कर सकता है; बाकी 10 इन टास्क के लिए अप्रयुक्त रहते हैं।

2-3. Spark के पीछे Java Virtual Machine (JVM)

Apache Spark, Java वर्चुअल मशीन (JVM) पर बनाया गया है।

Spark के बाइटकोड को चलाने के लिए JVM एक आवश्यक रनटाइम परिवेश है।

चूँकि Spark, Scala/Java में लिखा गया है:

• एप्लिकेशन में Scala या Java का उपयोग करने पर: कोड JVM के भीतर सीधे चलता है।
• PySpark (Python) का उपयोग करने पर: Python प्रक्रिया कोड चलाती है, लेकिन JVM-आधारित ड्राइवर से Py4J जैसे ब्रिज के ज़रिए संवाद करती है। वर्कर नोड पर मौजूद Spark एक्ज़ीक्यूटर JVM के भीतर ही चलते हैं, जो फिर Python कोड और Python उपयोगकर्ता-परिभाषित फ़ंक्शन (Python UDFs) चलाने के लिए Python वर्करों का प्रबंधन करते हैं।

दोनों ही स्थितियों में, JVM ड्राइवर और एक्ज़ीक्यूटर—दोनों के लिए निष्पादन थ्रेड, मेमोरी आवंटन और गार्बेज कलेक्शन का प्रबंधन करता है।

PySpark एप्लिकेशन में ड्राइवर और एक्ज़ीक्यूटर JVM

PySpark एप्लिकेशन में, Python ड्राइवर JVM में चल रहे Spark ड्राइवर से संवाद करता है;

• Python ड्राइवर: यहाँ एप्लिकेशन-स्तर का PySpark कोड लिखा और चलाया जाता है।
• ड्राइवर JVM: Spark के निष्पादन और शेड्यूलिंग का समन्वय करता है।
• एक्ज़ीक्यूटर JVM: Spark टास्क चलाता है और निष्पादन संसाधनों का प्रबंधन करता है। जब टास्क में Python निष्पादन की आवश्यकता होती है, तो Python वर्कर प्रक्रियाएँ भी शामिल होती हैं।
• क्लाइंट JVM (क्लाइंट प्रक्रिया या सबमिशन मशीन): स्थानीय मशीन या एज/गेटवे नोड पर चलने वाली JVM।

चित्र में दिखाए अनुसार, क्लाइंट JVM पहले ड्राइवर JVM से जुड़ता है, जो आगे एक्ज़ीक्यूटर JVM के साथ समन्वय करता है:

चित्र D. क्लाइंट, ड्राइवर और एक्ज़ीक्यूटर JVM के बीच समग्र संबंध

इसके बाद, ड्राइवर JVM एप्लिकेशन का प्रबंधन करने के लिए Spark ड्राइवर को होस्ट करता है, जबकि एक्ज़ीक्यूटर JVM अलग-अलग एक्ज़ीक्यूटर को होस्ट करते हैं, जो उन टास्क को चलाने के लिए ज़िम्मेदार होते हैं।

हर JVM ड्राइवर के पास डेटा प्रोसेसिंग, कैशिंग और उपयोगकर्ता ऑब्जेक्ट को कुशलता से संभालने के लिए JVM हीप होता है।

2-4. Resilient Distributed Datasets (RDDs) की गहन जानकारी

Resilient Distributed Dataset (RDD), तत्वों का एक अपरिवर्तनीय, विभाजित संग्रह है, जिसे Spark समानांतर रूप से संसाधित कर सकता है।

RDD, रूपांतरणों के बीच निर्भरताओं का हिसाब रखते हैं, जिससे विफलता की कई स्थितियों में Spark खोए हुए विभाजनों की फिर से गणना कर सकता है।

इस नाम में तीन विशेषताएँ समाहित हैं:

• लचीला: Spark वंशावली दर्ज करता है—कौन-सा RDD किस पिछले RDD पर निर्भर है—और हर मध्यवर्ती परिणाम की प्रतिकृतियाँ बनाए रखने के बजाय, अक्सर लापता विभाजनों को फिर से बना सकता है।
• वितरित: डेटासेट को विभाजनों के रूप में दर्शाया जाता है, जिन्हें Spark अलग-अलग एक्ज़ीक्यूटर पर संसाधित कर सकता है। कई भौतिक मशीनों की आवश्यकता नहीं होती; स्थानीय मोड में भी RDD समर्थित हैं।
• डेटासेट: RDD में कुंजी-मूल्य युग्म, पूर्णांक, स्ट्रिंग, ट्यूपल और अन्य समर्थित तत्व हो सकते हैं—केवल रिलेशनल पंक्तियाँ ही नहीं। RDD API में जैसे ऑपरेशन शामिल हैं map, filter, reduceByKey, join, और aggregateByKey.

RDD एक तार्किक संग्रह और गणना का अमूर्तन है, यह ऐसी घोषणा नहीं है कि कुछ खास विभाजनों को हमेशा के लिए कुछ खास वर्करों को सौंप दिया गया है।

विभाजन

विभाजन डेटा का एक तार्किक खंड होता है और आम तौर पर किसी स्टेज के एक टास्क के अनुरूप होता है।

मान लें कि हम 100 विभाजन बनाते हैं (शुरुआती 100 विभाजन शिक्षण के लिए लिया गया एक स्पष्ट विकल्प हैं)।

10,000 स्रोत तत्वों के साथ, हर विभाजन में लगभग 100 तत्व होंगे, यदि इनपुट मोटे तौर पर समान रूप से वितरित हो:

चित्र E. 100 शुरुआती डेटा खंडों में समानांतर विभाजन आवंटन

3. RDD API बनाम DataFrame API: आंतरिक Catalyst अनुकूलन

मूल RDD उदाहरण से वंशावली, विभाजन और शफ़ल को समझने में मदद मिलती है।

हालाँकि, सामान्य संरचित ETL में DataFrame API से शुरुआत करना बेहतर होता है, क्योंकि यह कॉलम, स्कीमा की जानकारी, SQL-जैसी अभिव्यक्तियाँ और ऑप्टिमाइज़र को दिखाई देने वाले ऑपरेशन उपलब्ध कराता है।

नीचे दिया गया आरेख इनके बीच का संबंध दिखाता है:

चित्र F. Spark की ग्राफ़ अवधारणाओं का आपसी संबंध; RDD वंशावली, DataFrame और Spark निष्पादन DAG

जिस तरह RDD में रूपांतरण को परिभाषित करने के लिए RDD वंशावली होती है, उसी तरह DataFrame की क्वेरी योजना उसकी संरचित क्वेरी को दर्शाती है।

Spark निर्भरताओं और भौतिक ऑपरेटरों के आधार पर निष्पादन का काम तैयार करता है।

DataFrame का निष्पादन आंतरिक रूप से RDD-आधारित तरीकों का उपयोग कर सकता है, लेकिन इसका यह मतलब नहीं है कि DataFrame कोड लिखते समय उपयोगकर्ताओं को RDD वंशावली खुद बनानी होगी।

3-1. DataFrame क्वेरी योजना

DataFrame क्वेरी योजना उन तार्किक और भौतिक चरणों का क्रम है जिन्हें निष्पादन इंजन तैयार करता है (चित्र L में बैंगनी बॉक्स)।

DataFrame क्वेरी कई निरूपणों से होकर गुजरती है:

1. तार्किक योजना: अनुरोधित ऑपरेशनों (स्कैन, फ़िल्टर, समूह बनाना, एग्रीगेशन) का विवरण।
2. विश्लेषित और अनुकूलित तार्किक योजना: Spark संदर्भों को हल करता है और लागू होने वाले तार्किक अनुकूलन करता है, जैसे कॉलम प्रूनिंग या प्रेडिकेट पुशडाउन।
3. भौतिक योजना: Spark निष्पादन योग्य ऑपरेटर चुनता है, जैसे स्कैन, आंशिक/अंतिम एग्रीगेशन, एक्सचेंज, जॉइन और सॉर्ट।
4. निष्पादन: भौतिक योजना को Spark के निष्पादन तंत्र के ज़रिए चलाया जाता है और अंततः इसमें एक्ज़ीक्यूटर पर टास्क शामिल होते हैं।

हमारे उपयोग के मामले में, भौतिक योजना में ये चरण शामिल हो सकते हैं:

• आंशिक हैश एग्रीगेशन,
• SKU के आधार पर एक्सचेंज, और
• अंतिम हैश एग्रीगेशन।

3-2. कार्रवाइयों की तुलना: RDD API बनाम DataFrame API

उदाहरण के तौर पर SKU एग्रीगेशन लेते हुए, आइए RDD API और DataFrame API की तुलना करें।

RDD API

तत्व (sku, qty) युग्म हैं। हमारा Python एग्रीगेशन तर्क स्पष्ट है और Spark RDD निर्भरताओं का अनुसरण करता है:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

DataFrame API

मान लें sales_df में ये कॉलम हैं sku (स्ट्रिंग) और qty (संख्यात्मक):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

यहाँ, दोनों उदाहरण एक ही व्यावसायिक समस्या हल करते हैं:

अमान्य बिक्री रिकॉर्ड को फ़िल्टर करना और SKU के आधार पर मान्य मात्राओं का एग्रीगेशन करना।

और दोनों ही मामलों में, collect() Spark का निष्पादन शुरू करता है और सभी परिणामी रिकॉर्ड Driver को लौटाता है।

यहाँ मुख्य अंतर परिणाम के निरूपण में है:

• RDD API: Python तत्व लौटाता है, जैसे (sku, total_qty) टपल।
• DataFrame API: संरचित Row ऑब्जेक्ट लौटाता है, जिनमें नाम वाले फ़ील्ड होते हैं, जैसे sku और total_qty.

यहाँ RDD और DataFrame API की समग्र तुलना दी गई है:

तालिका 2. RDD API बनाम DataFrame API—फ़ीचर और अमूर्तन की तुलना: (निर्माणकर्ता: Kuriko IWAI)

4. निष्कर्ष

Spark ट्यूनिंग का लक्ष्य पार्टिशनों की संख्या या CPU उपयोग को अलग-अलग अधिकतम करना नहीं है।

इसका लक्ष्य अस्थिरता पैदा किए बिना शुरू से अंत तक निष्पादन की लागत और विलंबता कम करना है।

इस लेख में, हम इसके सिस्टम के मूलभूत संबंधों को समझते हैं:

• संसाधन: वर्कर एक्ज़ीक्यूटर को होस्ट करते हैं; एक्ज़ीक्यूटर टास्क चलाते हैं; कॉन्फ़िगर किए गए CPU और मेमोरी समवर्ती निष्पादन को सीमित करते हैं।
• RDD प्रोग्रामिंग: ट्रांसफ़ॉर्मेशन से अपरिवर्तनीय RDD परिभाषाएँ और डेटा वंशावली बनती हैं; ऐक्शन गणना शुरू करते हैं।
• DataFrame प्रोग्रामिंग: ट्रांसफ़ॉर्मेशन से तार्किक क्वेरी ऑपरेशन बनते हैं; अनुकूलन प्रक्रिया निष्पादन योग्य भौतिक ऑपरेटर चुनती है; ऐक्शन गणना शुरू करते हैं।
• निष्पादन: Spark काम को जॉब, स्टेज और टास्क में व्यवस्थित करता है, और टास्क एक्ज़ीक्यूटर पर पार्टिशनों को प्रोसेस करते हैं। शफ़ल और अन्य निर्भरताएँ स्टेजों के बीच समन्वय की सीमाएँ बनाती हैं।

Spark वितरित कार्यभार को कैसे निष्पादित करता है, यह समझना केवल बुनियादी आधार है।

आप RDD API का उपयोग करके गणना की सात-चरणीय प्रक्रिया सीख सकते हैं।