क्या आरडीडी में कोई कार्रवाई ऑर्डर रखती है?

Question

मुझे RDD में reduce जैसे प्रदर्शन की आवश्यकता है, लेकिन ऑपरेटर को कम्यूटिव होने की आवश्यकता नहीं है। यानी मैं चाहता हूं कि result अनुवर्ती में हमेशा "123456789" होगा।

scala> val rdd = sc.parallelize(1 to 9 map (_.toString)) 
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[24] at parallelize at <console>:24 

scala> val result = rdd.someAction{ _+_ } 

सबसे पहले, मुझे fold मिला। RDD#fold के दस्तावेज़ का कहना है:

डीईएफ़ गुना (zeroValue: टी) (सेशन: (टी, टी) ⇒ टी): टी सकल सभी विभाजनों के लिए के तत्वों प्रत्येक विभाजन, और उसके बाद परिणाम, का उपयोग करते हुए एक दिया साहचर्य समारोह और एक तटस्थ "शून्य मान"

नोट है कि वहाँ कोई विनिमेय दस्तावेज़ में की जरूरत है। हालांकि, परिणाम की उम्मीद नहीं है के रूप में:

scala> rdd.fold(""){ _+_ } 
res10: String = 312456879 

संपादित के रूप में @ dk14 ने उल्लेख मैं कोशिश की है, कोई भाग्य के साथ:

scala> val rdd = sc.parallelize(1 to 9 map (_.toString)) 
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[48] at parallelize at <console>:24 

scala> rdd.zipWithIndex.map(x => (x._2, x._1)).sortByKey().map(_._2).fold(""){ _+_ } 
res22: String = 341276895 

scala> rdd.zipWithIndex.map(x => (x._2, x._1)).sortByKey().map(_._2).fold(""){ _+_ } 
res23: String = 914856273 

scala> rdd.zipWithIndex.map(x => (x._2, x._1)).sortByKey().map(_._2).fold(""){ _+_ } 
res24: String = 742539618 

scala> rdd.zipWithIndex.map(x => (x._2, x._1)).sortByKey().map(_._2).fold(""){ _+_ } 
res25: String = 271468359 

Answer 1

कोई अंतर्निहित कार्रवाई कि स्काला में इस मानदंड को पूरा करता को कम करने, लेकिन आप आसानी mapPartitions, collect और स्थानीय कटौती के संयोजन के द्वारा अपने स्वयं के लागू कर सकते हैं :

import scala.reflect.ClassTag 

def orderedFold[T : ClassTag](rdd: RDD[T])(zero: T)(f: (T, T) => T): T = { 
    rdd.mapPartitions(iter => Iterator(iter.foldLeft(zero)(f))).collect.reduce(f) 
} 

बजाय मर्ज fold द्वारा प्रयोग किया जाता अतुल्यकालिक और अव्यवस्थित विधि के लिए collect और reduce के संयोजन का उपयोग करना सुनिश्चित करता है कि वैश्विक आदेश संरक्षित है।

निश्चित रूप से

यह सहित कुछ अतिरिक्त लागत के साथ आता है: ड्राइवर पर

• थोड़ा अधिक स्मृति पदचिह्न।
• काफी अधिक विलंबता - हम स्थानीय कमी शुरू करने से पहले सभी कार्यों को समाप्त करने के लिए स्पष्ट रूप से प्रतीक्षा करते हैं।

Answer 2

रूप @YuvalItzchakov fold द्वारा बताया आदेश की रक्षा नहीं करता है परिणामों को मिलाते समय विभाजित RDD में। इस मूल RDD एक के लिए केवल विभाजन वालों पर विचार समझाने के लिए,

scala> val rdd = sc.parallelize(1 to 9 map (_.toString)).coalesce(1) 
rdd: org.apache.spark.rdd.RDD[String] = CoalescedRDD[27] at coalesce at <console>:27 

scala> rdd.zipWithIndex.map(x => (x._2, x._1)).sortByKey().map(_._2).fold(""){ _+_ } 
res4: String = 123456789 

scala> rdd.zipWithIndex.map(x => (x._2, x._1)).sortByKey().map(_._2).fold(""){ _+_ } 
res5: String = 123456789 

scala> rdd.zipWithIndex.map(x => (x._2, x._1)).sortByKey().map(_._2).fold(""){ _+_ } 
res6: String = 123456789