如何从 spark RDD 读取连续索引的值

Question

我对 Spark Scala 从系列键获取第一个值有疑问，我创建了一个新的 RDD，如下所示：

[(a,1),(a,2),(a,3),(a,4),(b,1),(b,2),(a,3),(a,4),(a,5),(b,8),(b,9)]

我想获取这样的结果：

[(a,1),(b,1),(a,3),(b,8)]

如何使用 RDD 中的 scala 来做到这一点

Answer 1

如 cmets 中所述，为了能够使用 RDD 中元素的顺序，您必须以某种方式在数据本身中表示此顺序。正是为了这个目的，创建了zipWithIndex——索引被添加到数据中；然后，通过一些操作（join 在带有修改索引的 RDD 上）我们可以得到你需要的：

// add index to RDD:
val withIndex = rdd.zipWithIndex().map(_.swap)

// create another RDD with indices increased by one, to later join each element with the previous one
val previous = withIndex.map { case (index, v) => (index + 1, v) }

// join RDDs, filter out those where previous "key" is identical
val result = withIndex.leftOuterJoin(previous).collect {
  case (i, (left, None)) => (i, left) // keep first element in RDD
  case (i, (left, Some((key, _)))) if left._1 != key => (i, left) // keep only elements where the previous key is different
}.sortByKey().values // if you want to preserve the original order...

result.collect().foreach(println)
// (a,1)
// (b,1)
// (a,3)
// (b,8)