Преобразование RDD внутри цикла

Question

Итак, у меня есть rdd: Array [String] с именем Adat, и я хочу преобразовать его в цикле и получить новое RDD, которое я могу использовать вне области цикла. Я пробовал это, но результат не то, что я хочу.

val sharedA = { 
    for { 
    i <- 0 to shareA.toInt - 1 
    j <- 0 to shareA.toInt - 1 
    } yield { 
    Adat.map(x => (x(1).toInt, i % shareA.toInt, j % shareA.toInt, x(2))) 
    } 
} 

Приведенный выше код преобразует SharedA рдд в IndexedSeq [РДУ [(Int, Int, Int, String)]], и когда я пытаюсь напечатать его результат является:

MapPartitionsRDD[12] at map at planet.scala:99 

MapPartitionsRDD[13] at map at planet.scala:99 and so on. 

Как преобразовать sharedA в RDD[(Int, Int, Int, String)]?

Если я делаю это так, у sharedA есть правильный тип данных, но я не могу использовать его вне области видимости.

 for { i <- 0 to shareA.toInt -1 
     j<-0 to shareA.toInt-1 } 
     yield { 
     val sharedA=Adat.map(x => (x(1).toInt,i % shareA.toInt ,j %  
     shareA.toInt,x(2)))  
     } 

Answer 1

Я не совсем понимаю ваше описание, но flatMap должен сделать трюк:

val rdd = sc.parallelize(Seq(Array("", "0", "foo"), Array("", "1", "bar"))) 
val n = 2 

val result = rdd.flatMap(xs => for { 
    i <- 0 to n 
    j <- 0 to n 
} yield (xs(1).toInt, i, j, xs(2))) 

result.take(5) 
// Array[(Int, Int, Int, String)] = 
// Array((0,0,0,foo), (0,0,1,foo), (0,0,2,foo), (0,1,0,foo), (0,1,1,foo)) 

Реже подход будет вызывать SparkContext.union по результатам:

val resultViaUnion = sc.union(for { 
    i <- 0 to n 
    j <- 0 to n 
} yield rdd.map(xs => (xs(1).toInt, i, j, xs(2)))) 

resultViaUnion.take(5) 
// Array[(Int, Int, Int, String)] = 
// Array((0,0,0,foo), (1,0,0,bar), (0,0,1,foo), (1,0,1,bar), (0,0,2,foo))