स्पार्क के कैशिंग को समझना

Question

मैं समझने की कोशिश कर रहा हूं कि स्पार्क का कैश कैसे काम करता है।

यहाँ मेरी भोली समझ है, मुझे पता है कि अगर मैं कुछ याद कर रहा हूँ तो कृपया:

val rdd1 = sc.textFile("some data") 
rdd1.cache() //marks rdd1 as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

ऊपर, rdd1 में केवल एक बार डिस्क (जैसे HDFS) से लोड किया जाएगा। (जब rdd2 बचाया जाता है तो मुझे लगता है) और फिर कैश से (मान लीजिए कि पर्याप्त रैम है) जब rdd3 बचाया जाता है)

अब मेरा प्रश्न है। मान लीजिए कि मैं rdd2 और rdd3 को कैश करना चाहता हूं क्योंकि दोनों का बाद में उपयोग किया जाएगा, लेकिन मुझे बनाने के बाद मुझे rdd1 की आवश्यकता नहीं है।

असल में नकल है, है ना? चूंकि एक बार rdd2 और rdd3 की गणना की जाती है, मुझे अब rdd1 की आवश्यकता नहीं है, मुझे शायद इसे अनपेक्षित करना चाहिए, है ना? सवाल कब है?

क्या यह काम करेगा? (विकल्प A)

val rdd1 = sc.textFile("some data") 
rdd1.cache() // marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.cache() 
rdd3.cache() 
rdd1.unpersist() 

चिंगारी DAG को unpersist कॉल होता है? या यह तुरंत किया जाता है? अगर यह तुरंत किया जाता है, तो जब मैं rdd2 और rdd3 से पढ़ता हूं, तो मूल रूप से rdd1 गैर कैश किया जाएगा, है ना?

क्या मुझे इसके बजाय ऐसा करना चाहिए (विकल्प बी)?

val rdd1 = sc.textFile("some data") 
rdd1.cache() // marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 

rdd2.cache() 
rdd3.cache() 

rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

rdd1.unpersist() 

तो सवाल यह है: विकल्प पर्याप्त एक अच्छा है? यानी rdd1 अभी भी एक बार फ़ाइल लोड करेगा? या मुझे विकल्प बी के साथ जाने की ज़रूरत है?

Answer 1

ऐसा लगता है कि विकल्प बी आवश्यक है। कारण स्पार्क द्वारा कैसे जारी/कैश और unpersist निष्पादित किया जाता है से संबंधित है। चूंकि आरडीडी परिवर्तन केवल निष्पादन के बिना डीएजी विवरण बनाते हैं, जब आप एस्पर्सिस्ट को कॉल करते समय विकल्प ए में, तब भी आपके पास केवल नौकरी के विवरण होते हैं और चल रहे निष्पादन नहीं होते हैं।

यह प्रासंगिक है क्योंकि cache या persist कॉल केवल आरडीडी को आरडीडी के मानचित्र में जोड़ता है जो खुद को नौकरी निष्पादन के दौरान जारी रखने के लिए चिह्नित करता है। हालांकि, unpersist सीधे ब्लॉक प्रबंधक को भंडारण से आरडीडी को बेदखल करने के लिए कहता है और लगातार आरडीडी के मानचित्र में संदर्भ को हटा देता है।

persist function

unpersist function

तो तुम स्पार्क वास्तव में मार डाला और ब्लॉक प्रबंधक के साथ RDD संग्रहीत के बाद unpersist कॉल करने के लिए की आवश्यकता होगी।

इस दिशा में RDD.persist विधि संकेत के लिए टिप्पणियाँ: rdd.persist

Answer 2

विकल्प एक में, आप नहीं दिखाया जब आप कार्रवाई बुला रहे हैं (बचाने के लिए कॉल)

val rdd1 = sc.textFile("some data") 
rdd.cache() //marks rdd as cached 
val rdd2 = rdd1.filter(...) 
val rdd3 = rdd1.map(...) 
rdd2.cache() 
rdd3.cache() 
rdd1.unpersist() 
rdd2.saveAsTextFile("...") 
rdd3.saveAsTextFile("...") 

अनुक्रम है उपर्युक्त के रूप में, विकल्प ए को rdd2 और rdd 3

Answer 3

विकल्प बी की गणना करने के लिए rdd1 के कैश किए गए संस्करण का उपयोग करना चाहिए विकल्प बी छोटे tweak-in के साथ एक इष्टतम दृष्टिकोण है। कम महंगी कार्रवाई विधियों का उपयोग करें। आपके कोड द्वारा उल्लिखित दृष्टिकोण में, saveAsTextFile एक महंगी ऑपरेशन है, इसे गिनती विधि द्वारा प्रतिस्थापित करें।

आइडिया यहाँ DAG से बड़ा rdd1 दूर करने के लिए, अगर यह प्रासंगिक नहीं है आगे गणना के लिए (के बाद rdd2 और rdd3 बनाई गई हैं)

कोड से

दृष्टिकोण अपडेट किया गया

val rdd1 = sc.textFile("some data").cache() 
val rdd2 = rdd1.filter(...).cache() 
val rdd3 = rdd1.map(...).cache() 

rdd2.count 
rdd3.count 

rdd1.unpersist()