Scal wiele RDD generowanych w pętli

Question

Wywołuję funkcję w scala, która daje RDD[(Long,Long,Double)] jako wyjście.

def helperfunction(): RDD[(Long, Long, Double)]

nazywam tę funkcję w pętli w innej części kodu i chcę, aby scalić wszystkie wygenerowane RDD. Pętla wywołując funkcję wygląda tak

for (i <- 1 to n){ 
    val tOp = helperfunction() 
    // merge the generated tOp 
} 

Co chcę zrobić coś podobnego do tego, co StringBuilder zrobiłby dla ciebie w Javie, kiedy chciał się połączyć za sznurki. Mam spojrzał na technikach łączących RDD, które w większości wskazują na użyciu funkcji związkowej jak ten

RDD1.union(RDD2) 

Wymaga to jednak oba RDD mają zostać wygenerowane przed podjęciem unii. Chciałem zainicjować var ​​RDD1, aby zebrać wyniki poza pętlą for, ale nie jestem pewien jak zainicjować puste RDD typu [(Long,Long,Double)]. Również zaczynam od iskry, więc nie jestem nawet pewien, czy jest to najbardziej elegancka metoda rozwiązania tego problemu.

Answer 1

Zamiast Vars, można użyć funkcjonalnych paradygmat programowania, aby osiągnąć to, co chcesz:

val rdd = (1 to n).map(x => helperFunction()).reduce(_ union _) 

Ponadto, jeśli nadal trzeba utworzyć pusty RDD, można zrobić to za pomocą:

val empty = sc.emptyRDD[(long, long, String)] 

Answer 2

Masz rację, że nie jest to optymalny sposób, aby to zrobić, ale potrzebowalibyśmy więcej informacji na temat tego, co próbujesz osiągnąć, generując nowe RDD z każdym wywołaniem funkcji pomocnika.

Możesz zdefiniować 1 RDD przed pętlą i przypisać mu var, a następnie uruchomić go w pętli. Oto przykład:

val rdd = sc.parallelize(1 to 100) 
val rdd_tuple = rdd.map(x => (x.toLong, (x*10).toLong, x.toDouble)) 
var new_rdd = rdd_tuple 
println("Initial RDD count: " + new_rdd.count()) 
for (i <- 2 to 4) { 
    new_rdd = new_rdd.union(rdd_tuple) 
} 
println("New count after loop: " + new_rdd.count())