如何在Spark的RDD中获取元素位置？

Question

我是Apache Spark的新手,我知道核心数据结构是RDD.现在我正在编写一些需要元素位置信息的应用程序.例如,在将ArrayList转换为(Java)RDD之后,对于RDD中的每个整数,我需要知道它的(全局)数组下标.有可能吗？

据我所知,RDD 有一个take(int)函数,所以我相信位置信息仍然保留在RDD中.

Answer 1

我相信在大多数情况下,zipWithIndex()会做到这一点,它会保留顺序.再次阅读评论.我的理解是,它确实意味着保持RDD中的顺序.

scala> val r1 = sc.parallelize(List("a", "b", "c", "d", "e", "f", "g"), 3)
scala> val r2 = r1.zipWithIndex
scala> r2.foreach(println)
(c,2)
(d,3)
(e,4)
(f,5)
(g,6)
(a,0)
(b,1)

以上示例确认了它.红色有3个分区,a有索引0,b有索引1,等等.

Answer 2

本质上,RDD的zipWithIndex()方法似乎这样做,但它不会保留RDD创建的数据的原始顺序.至少你会获得稳定的订购.

val orig: RDD[String] = ...
val indexed: RDD[(String, Long)] = orig.zipWithIndex()

您不太可能找到保留原始数据中的顺序的原因隐藏在zipWithIndex()的API文档中:

"将RDD与其元素索引一起使用.排序首先基于分区索引,然后是每个分区中项目的顺序.所以第一个分区中的第一个项目得到索引0,最后一个分区中的最后一个项目接收到这与Scala的zipWithIndex相似,但它使用Long而不是Int作为索引类型.当这个RDD包含多个分区时,此方法需要触发一个spark作业."

所以看起来原始订单被丢弃了.如果保留原始顺序对您很重要,则在创建RDD 之前,您需要添加索引.