एक पुनरावृत्ति गणना के परिणाम एकत्र करने के लिए आरडीडी बनाना

Question

मैं एक पुनरावृत्ति गणना के परिणाम एकत्र करने के लिए एक आरडीडी बनाना चाहता हूं।

मैं एक पाश (या किसी भी वैकल्पिक) का उपयोग कर सकते कैसे निम्नलिखित कोड को बदलने के लिए:

import org.apache.spark.mllib.random.RandomRDDs._  

val n = 10 

val step1 = normalRDD(sc, n, seed = 1) 
val step2 = normalRDD(sc, n, seed = (step1.max).toLong) 
val result1 = step1.zip(step2) 
val step3 = normalRDD(sc, n, seed = (step2.max).toLong) 
val result2 = result1.zip(step3) 

... 

val step50 = normalRDD(sc, n, seed = (step49.max).toLong) 
val result49 = result48.zip(step50) 

(एन कदम RDDs बनाने और अंत में एक साथ तो ज़िप करने भी लंबे 50 RDDs के रूप में ठीक हो जाएगा बीज = सम्मान करने के लिए बनाई गई हैं iteratively (चरण (n-1) .max) हालत)

Answer 1

एक पुनरावर्ती समारोह काम करेगा:

/** 
* The return type is an Option to handle the case of a user specifying 
* a non positive number of steps. 
*/ 
def createZippedNormal(sc : SparkContext, 
         numPartitions : Int, 
         numSteps : Int) : Option[RDD[Double]] = { 

    @scala.annotation.tailrec 
    def accum(sc : SparkContext, 
      numPartitions : Int, 
      numSteps : Int, 
      currRDD : RDD[Double], 
      seed : Long) : RDD[Double] = { 
    if(numSteps <= 0) currRDD 
    else { 
     val newRDD = normalRDD(sc, numPartitions, seed) 
     accum(sc, numPartitions, numSteps - 1, currRDD.zip(newRDD), newRDD.max) 
    } 
    } 

    if(numSteps <= 0) None 
    else Some(accum(sc, numPartitions, numSteps, sc.emptyRDD[Double], 1L)) 
}