合并循环中生成的多个RDD

Question

我在scala中调用一个函数,它给出了一个RDD[(Long,Long,Double)]输出.

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

我在代码的另一部分循环调用此函数,我想合并所有生成的RDD.调用该函数的循环看起来像这样

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

我想要做的是类似于StringBuilder在你想要合并字符串时为你做的事情.我已经研究了合并RDD的技术,这些技术主要指向使用这样的union函数

RDD1.union(RDD2)

但是这需要在结合之前生成两个RDD.我虽然初始化一个var RDD1来累积for循环之外的结果,但我不知道如何初始化类型的空白RDD [(Long,Long,Double)].另外我开始使用spark,所以我甚至不确定这是否是解决此问题的最优雅的方法.

Answer 1

您可以使用函数式编程范例来实现您想要的目的，而不是使用变量：

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

另外，如果您仍然需要创建一个空的 RDD，您可以使用：

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