使用共享可变状态向 RDD 添加索引

Question

以这个简单的RDD为例说明问题：

val testRDD=sc.parallelize(List((1, 2), (3, 4), (3, 6)))

我有这个函数来帮助我实现索引：

 var sum = 0; 

 def inc(l: Int): Int = {
    sum += l
    sum 
 }

现在我想为每个元组创建 id：

val indexedRDD= testRDD.map(x=>(x._1,x._2,inc(1)));

输出RDD应该是((1,2,1), (3,4,2), (3,6,3))

但事实证明所有的值都是一样的。所有元组都取 1：

((1,2,1), (3,4,1), (3,6,1))

我哪里出错了？有没有其他方法可以达到同样的效果。

Answer 1

您正在寻找：

def zipWithIndex(): RDD[(T, Long)]

但是，请注意文档：

请注意，某些 RDD，例如由 groupBy() 返回的 RDD，不 保证分区中元素的顺序。分配给每个的索引 因此，不保证元素，如果 RDD 为 重新评估。如果需要固定订购以保证相同 索引分配，您应该使用 sortByKey() 对 RDD 进行排序或保存 到一个文件。