डेटाफ्रेम का उपयोग क्यों करेगा। DataFrame.rdd.map (या इसके विपरीत) पर चयन करें?

Question

वहाँ एक DataFrame पर select का उपयोग कर हमारे लिए आवश्यक जानकारी लेने के लिए और एक ही उद्देश्य के लिए अंतर्निहित RDD की प्रत्येक पंक्ति के मानचित्रण के बीच किसी भी "यांत्रिक" अंतर है?

तक "यांत्रिक" मैं तंत्र कि संचालन करता है की चर्चा करते हुए कर रहा हूँ। कार्यान्वयन विवरण, दूसरे शब्दों में।

प्रस्तावित दो में से कौन सा बेहतर/अधिक प्रदर्शनकारी है?

df = # create dataframe ... 
df.select("col1", "col2", ...) 

या

df = # create dataframe ... 
df.rdd.map(lambda row: (row[0], row[1], ...)) 

मैं निष्पादन परीक्षण के बीच में हूं, तो मैं पता लगाने के लिए जो तेजी से होता है जा रहा हूँ, लेकिन मैं पता है कि कार्यान्वयन मतभेद और पेशेवरों/विपक्ष हैं चाहते हैं।

Answer 1

इस व्यापक उदाहरण में DataFrame.select और DataFrame.rdd.map के साथ तुलना में तेज़ होगा, मुझे लगता है कि अंतर लगभग नगण्य हो सकता है।

आखिरकार आपने अपना डेटा सेट पहले ही लोड कर लिया है और केवल प्रक्षेपण किया है। अंततः दोनों को कार्रवाई के परिणाम की गणना करने के लिए स्पार्क के InternalRow स्तंभ प्रारूप से डेटा को deserialize करना होगा।

आप DataFrame.select के साथ explain(extended = true) पर क्या हो सकता है, जहां आप भौतिक योजनाओं (और एक भौतिक योजना भी) के बारे में जानेंगे।

scala> spark.version 
res4: String = 2.1.0-SNAPSHOT 

scala> spark.range(5).select('id).explain(extended = true) 
== Parsed Logical Plan == 
'Project [unresolvedalias('id, None)] 
+- Range (0, 5, step=1, splits=Some(4)) 

== Analyzed Logical Plan == 
id: bigint 
Project [id#17L] 
+- Range (0, 5, step=1, splits=Some(4)) 

== Optimized Logical Plan == 
Range (0, 5, step=1, splits=Some(4)) 

== Physical Plan == 
*Range (0, 5, step=1, splits=Some(4)) 

आप rdd.map साथ क्या कर रहे हैं (toDebugString द्वारा) करने के लिए भौतिक योजना (अर्थात SparkPlan) की तुलना करें और आपको पता चल जाएगा कि क्या "बेहतर" हो सकता है।

scala> spark.range(5).rdd.toDebugString 
res5: String = 
(4) MapPartitionsRDD[8] at rdd at <console>:24 [] 
| MapPartitionsRDD[7] at rdd at <console>:24 [] 
| MapPartitionsRDD[6] at rdd at <console>:24 [] 
| MapPartitionsRDD[5] at rdd at <console>:24 [] 
| ParallelCollectionRDD[4] at rdd at <console>:24 [] 

(फिर से इस संक्रमित उदाहरण में मुझे लगता है कि कोई विजेता नहीं है - दोनों जितना संभव हो उतना कुशल हैं)।

कृपया ध्यान दें कि DataFrame वास्तव में एक Dataset[Row] जो RowEncoder का उपयोग करता है एक InternalRow स्तंभ द्विपदीय प्रारूप में डेटा एन्कोड करने के लिए (अर्थात serialize) है। यदि आप पाइपलाइन में अधिक ऑपरेटरों को निष्पादित करना चाहते थे, तो से पर चिपकने के साथ आप बेहतर प्रदर्शन कर सकते हैं क्योंकि निम्न स्तर के पीछे-दृश्य-दृश्य लॉजिकल क्वेरी प्लान ऑप्टिमाइज़ेशन और कॉलमर बाइनरी प्रारूप।

बहुत सारे अनुकूलन हैं और उन्हें हरा करने की कोशिश करने से अक्सर आपके समय का अपशिष्ट हो सकता है। बेहतर प्रदर्शन पाने के लिए आपको स्पार्क इंटर्नल्स को दिल से जानना होगा (और कीमत निश्चित रूप से पठनीयता होगी)।

इसके लिए बहुत कुछ है और मैं दृढ़ता से सभी अनुकूलन को जानने और सराहना करने के लिए हरमन वैन होवेल द्वारा A Deep Dive into the Catalyst Optimizer पर बात करने की सलाह देता हूं।

मेरा ले लेना है ... "आरडीडी से दूर रहें जबतक कि आप नहीं जानते कि आप क्या कर रहे हैं"।

Answer 2

आरडीडी केवल परिवर्तन और क्रिया का एक ग्राफ वंश है।

एक DataFrame एक तार्किक योजना है कि है आंतरिक रूप से कार्रवाई के निष्पादन से पहले उत्प्रेरक तार्किक क्वेरी अनुकूलक द्वारा अनुकूलित है।

आपके मामले में इसका क्या अर्थ है?

यदि आपके पास डेटाफ्रेम है तो आपको select का उपयोग करना चाहिए - फ़िल्टरिंग, जॉइनिंग इत्यादि जैसे किसी अतिरिक्त काम को अनुकूलित किया जाएगा। अनुकूलित डेटाफ्रेम सादे आरडीडी की तुलना में 10 गुना तेज हो सकता है। दूसरे शब्दों में, select निष्पादित करने से पहले स्पार्क क्वेरी को तेज़ी से बनाने का प्रयास करेगा। जब dataFrame.rdd.map का उपयोग कर यह नहीं किया जाएगा()

एक और: rdd मूल्य करके lazily गणना की जाती है:

lazy val rdd: RDD[T] = { 
    val objectType = exprEnc.deserializer.dataType 
    val deserialized = CatalystSerde.deserialize[T](logicalPlan) 
    sparkSession.sessionState.executePlan(deserialized).toRdd.mapPartitions { rows => 
     rows.map(_.get(0, objectType).asInstanceOf[T]) 
    } 
    } 

तो स्पार्क, यह RDD है का उपयोग करेगा नक्शा और सामग्री डाली। प्रश्नों की तरह क्वेरी में दोनों संस्करणों का डीएजी लगभग समान होगा, इसलिए प्रदर्शन समान होगा। हालांकि डेटासेट का उपयोग करने के अधिक उन्नत मामलों में लाभ बहुत ही दिखाई देंगे, क्योंकि स्पार्क पीएमसी ने डाटाबेसिक्स ब्लॉग पर लिखा था, डेटासेट

द्वारा ऑप्टिमाइज़ेशन के बाद डेटासेट 100 गुना तेज हो सकता है, ध्यान रखें, डेटाफ्रेम = डेटासेट [पंक्ति] और यह पृष्ठभूमि में आरडीडी का उपयोग करता है - लेकिन आरडीडी का ग्राफ ऑप्टिमाइज़ेशन

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