apache-spark
विभाजन
खोज…
टिप्पणियों
किसी एप्लिकेशन के प्रदर्शन और / या सफल समाप्ति के लिए विभाजन की संख्या महत्वपूर्ण है।
एक लचीला वितरित डेटासेट (RDD) स्पार्क का मुख्य अमूर्त है। एक RDD विभाजन में विभाजित है, इसका मतलब है कि एक विभाजन डेटासेट का एक हिस्सा है, इसका एक टुकड़ा है, या दूसरे शब्दों में, इसका एक हिस्सा है।
विभाजन की संख्या जितनी अधिक होती है, प्रत्येक विभाजन का आकार उतना ही छोटा होता है।
हालांकि, ध्यान दें कि बड़ी संख्या में विभाजन Hadoop Distributed File System (HDFS) पर बहुत अधिक दबाव डालता है, जिसके लिए मेटाडेटा का एक महत्वपूर्ण हिस्सा रखना पड़ता है।
विभाजन की संख्या मेमोरी उपयोग से संबंधित है, और एक मेमोरीऑवरहेड समस्या इस संख्या ( व्यक्तिगत अनुभव ) से संबंधित हो सकती है।
नए उपयोगकर्ताओं के लिए एक सामान्य नुकसान केवल एक विभाजन के साथ अपने RDD को RDD में बदलना है, जो आमतौर पर ऐसा दिखता है:
data = sc.textFile(file)
data = data.coalesce(1)
यह आमतौर पर एक बहुत बुरा विचार है, क्योंकि आप स्पार्क को बता रहे हैं कि सभी डेटा को बस एक विभाजन है! उसे याद रखो:
स्पार्क में एक चरण एक समय में एक विभाजन पर काम करेगा (और उस विभाजन में डेटा को मेमोरी में लोड करेगा)।
नतीजतन, आप स्पार्क को एक ही बार में सभी डेटा को संभालने के लिए कहते हैं, जिसके परिणामस्वरूप आमतौर पर मेमोरी संबंधित त्रुटियां (उदाहरण के लिए मेमोरी से बाहर), या यहां तक कि एक शून्य पॉइंटर अपवाद भी होता है।
इसलिए, जब तक आप नहीं जानते कि आप क्या कर रहे हैं, केवल एक विभाजन में अपने RDD को पुन: प्रस्तुत करने से बचें!
विभाजन परिचय
RDD का विभाजन कैसे होता है?
डिफ़ॉल्ट रूप से प्रत्येक HDFS विभाजन के लिए एक विभाजन बनाया जाता है, जो डिफ़ॉल्ट रूप से 64MB है। और पढ़ें यहाँ
विभाजन में अपने डेटा को कैसे संतुलित करें?
सबसे पहले, उन तीन तरीकों पर एक नज़र डालें, जिनसे कोई अपने डेटा को पुनः प्राप्त कर सकता है:
एक दूसरे पैरामीटर को पास करें, अपने RDD के लिए विभाजन की वांछित न्यूनतम संख्या को TextFile () में रखें , लेकिन सावधान रहें:
[१४] में: लाइनें = sc.textFile ("डेटा")
[१५] में: लाइनों .getNumPartitions () आउट [१५]: १०००
[१६] में: लाइनें = sc.textFile ("डेटा", ५००)
[१itions] में: लाइनों.getNumPartitions () आउट [१ 14]: १४३४
[१ile] में: लाइनें = sc.textFile ("डेटा", ५०००)
[१ ९]: लाइनों.गेटनमपार्टिशन () आउट [१ ९]: ५ ९ २६ में
जैसा कि आप देख सकते हैं, [16] वह नहीं करता है जिसकी कोई अपेक्षा करता है, क्योंकि RDD के विभाजन की संख्या, हमारे द्वारा अनुरोध किए गए विभाजन की न्यूनतम संख्या से पहले से अधिक है।
रिपर्टिशन () का उपयोग करें, जैसे:
[२२] में: लाइनें = लाइनें।
[२३] में: लाइनों.getNumPartitions () आउट [२३]: १०
चेतावनी: यह एक फेरबदल करेगा और इसका उपयोग तब किया जाना चाहिए जब आप अपने RDD के विभाजन की संख्या बढ़ाना चाहते हैं।
डॉक्स से :
फेरबदल डेटा को फिर से वितरित करने के लिए स्पार्क का तंत्र है ताकि यह विभाजन में अलग-अलग समूहीकृत हो। इसमें आमतौर पर निष्पादकों और मशीनों में डेटा की प्रतिलिपि बनाना, फेरबदल को एक जटिल और महंगा ऑपरेशन बनाना है।
इस तरह () का उपयोग करें:
[२५]: लाइनों में = तर्ज पर (२)
[२६] में: लाइनों.getNumPartitions () आउट [२६]: २
यहां, स्पार्क जानता है कि आप आरडीडी को सिकोड़ लेंगे और इसका लाभ प्राप्त कर सकते हैं। पुनरावर्तन () बनाम कोलेससे () के बारे में और पढ़ें।
लेकिन क्या यह सब गारंटी देगा कि आपका डेटा आपके विभाजन में पूरी तरह से संतुलित होगा? वास्तव में नहीं, जैसा कि मैंने अनुभव किया कि विभाजन में अपने डेटा को कैसे संतुलित किया जाए?
एक RDD के विभाजन
जैसा कि "रिमार्क्स" में उल्लेख किया गया है, एक विभाजन RDD का एक हिस्सा / टुकड़ा / हिस्सा है। नीचे अपने RDD के लिए विभाजन की न्यूनतम संख्या का अनुरोध करने का एक न्यूनतम उदाहरण दिया गया है:
In [1]: mylistRDD = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 2)
In [2]: mylistRDD.getNumPartitions()
Out[2]: 2
सूचना [1] में हमने 2 को parallelize() दूसरे पैरामीटर के रूप में कैसे पारित किया parallelize() । वह पैरामीटर कहता है कि हम चाहते हैं कि हमारे RDD में कम से कम 2 विभाजन हों।
आरडीपीडी एक आरडीपी
कभी-कभी हम उदाहरण के लिए आरडीडी को फिर से भरना चाहते हैं, क्योंकि यह एक ऐसी फ़ाइल से आता है जो हमारे द्वारा नहीं बनाई गई थी, और निर्माता से परिभाषित विभाजन की संख्या वह नहीं है जो हम चाहते हैं।
इसे प्राप्त करने के लिए दो सबसे प्रसिद्ध कार्य हैं:
repartition(numPartitions)
तथा:
coalesce(numPartitions, shuffle=False)
अंगूठे के एक नियम के रूप में, पहले का उपयोग करें जब आप अपने RDD को विभाजन की एक बड़ी संख्या में और दूसरे को अपने RDD को कम करने के लिए विभाजन की संख्या में कम करना चाहते हैं। चिंगारी - repartition () vs coalesce () ।
उदाहरण के लिए:
data = sc.textFile(file)
data = data.coalesce(100) // requested number of #partitions
RDD के विभाजन की संख्या को 'डेटा' नामक 100 तक घटा देगा, यह देखते हुए कि इस RDD में 100 से अधिक विभाजन हैं, जब यह textFile() द्वारा पढ़ा गया।
और इसी तरह, यदि आप अपने RDD के लिए विभाजन की वर्तमान संख्या से अधिक होना चाहते हैं, तो आप कर सकते हैं (उदाहरण के लिए आपका RDD 200 विभाजन में वितरित किया गया है):
data = sc.textFile(file)
data = data.repartition(300) // requested number of #partitions
विभाजन की संख्या के बारे में अंगूठे का नियम
अंगूठे के नियम के रूप में, कोई भी अपने आरडीडी को 3 विभाजन (या शायद 4) द्वारा प्रयुक्त कोर की संख्या से निष्पादनकर्ताओं की संख्या के रूप में कई विभाजन करना चाहेगा। बेशक, यह एक अनुमान है और यह वास्तव में आपके आवेदन, डेटासेट और क्लस्टर कॉन्फ़िगरेशन पर निर्भर करता है।
उदाहरण:
In [1]: data = sc.textFile(file)
In [2]: total_cores = int(sc._conf.get('spark.executor.instances')) * int(sc._conf.get('spark.executor.cores'))
In [3]: data = data.coalesce(total_cores * 3)
RDD सामग्री दिखाएं
RDD की सामग्री दिखाने के लिए, इसे प्रिंट करना होगा:
myRDD.foreach(println)
मुद्रित पंक्तियों की संख्या सीमित करने के लिए:
myRDD.take(num_of_rows).foreach(println)