修改 zipWithIndex 的功能

【问题标题】：Modify the functioning of zipWithIndex修改 zipWithIndex 的功能
【发布时间】：2016-06-27 13:11:30
【问题描述】：

我有一个具有以下值的 RDD：

v
v v
v
v
v v
v

我想索引上面的RDD。我想要索引它的方式是如果一行有两个值，下一行的索引应该增加1 所以索引的RDD看起来像：

v, 0
v v, 1
v, 3
v, 4
v v, 5
v, 7

那么这甚至可以使用火花吗？如果是，我该如何完成？

更新 1

另外，如果有人能展示如何扩展 zipWithIndex 函数并创建一个新函数，该函数根据 ex 的输入字符串递增每一行的索引，那就太好了。 (data, index+data.length)

【问题讨论】：

v 到底是什么？
@meucaa "v" 是任意数据。基本上它们是空格分隔的值。

标签： scala indexing apache-spark rdd

【解决方案1】：

这是一个可能很慢但相对简单的解决方案：想法是用一些 ID 标记这些记录，拆分它们，以便每个值都是单独的记录（然后可以使用 @ 轻松枚举987654321@)，然后使用我们创建的 ID 组合回原始结构：

// sample data:
val input = sc.parallelize(Seq(
  List("a"),
  List("b", "c"),
  List("d"),
  List("e", "f")
))

val result: RDD[(List[String], Long)] = input
  .zipWithIndex() // add key to each record, so we can construct them back later
  .flatMap { case (list, id) => list.map(i => (id, i)) } // "split" into individual records
  .sortByKey().zipWithIndex() // enumerate individual values
  .groupBy { case ((key, value), index) => key } // grouping by key
  .values.map { iter =>   // mapping back to the structure we want
     val list = iter.toList
     val index: Long = list.map(_._2).min
     val values: List[String] = list.map(_._1._2)
     (values, index)
   }

结果是：

(List(a),0)
(List(b, c),1)
(List(d),3)
(List(e, f),4)

【讨论】：