从 RDD[ListBuffer] 中的 ListBuffer 中提取对象并在 RDD 中创建新条目的 Spark 转换答案

【问题标题】：Spark transformation that withdraws objects from a ListBuffer in a RDD[ListBuffer] and creates new entries in the RDD从 RDD[ListBuffer] 中的 ListBuffer 中提取对象并在 RDD 中创建新条目的 Spark 转换
【发布时间】：2017-12-07 14:45:52
【问题描述】：

所以我有一个 RDD，它是 RDD[ListBuffer[(Array[String], Long)]]。为简单起见，我们可以将其称为RDD[X]，其中X 是一些变量的列表。 X 是 obj 对象的列表，X[obj]。

我的想法是我想要一个函数，它将RDD[X] 作为输入并输出一个新的RDD[X]，一个转换。此转换将通过从一个 X 中取出 obj 并创建一个新列表来创建新的 X 列表，就像将其“附加”到 RDD 中一样。

我在 Spark 中没有找到任何直接支持这一点的东西。现在我能想到的唯一解决方案是通过执行collect() 并在驱动程序中管理大部分内容，但这显然不是很好。有什么想法吗？

基本上是这样的：

val data = RDD[ListBuffer[(Array[String], Long)]]
// some transformation that calls some function
// what will happen is some (Array[String], Long) will be moved into an entirely new ListBuffer in outData while some may be completely removed
val outData = RDD[ListBuffer[(Array[String], Long)]]

假设我们有一个包含 7 个元素的 ListBuffer 的起始 RDD：

Element1（在 ListBuffer1 中）

Element2（在 ListBuffer1 中）

Element3（在 ListBuffer1 中）

Element4（在 ListBuffer1 中）

Element5（在 ListBuffer1 中）

Element6（在 ListBuffer1 中）

Element7（在 ListBuffer1 中）

转换后的RDD会有如下内容：

Element1（在 ListBuffer1 中）

Element2（在 ListBuffer1 中）

Element4（在 ListBuffer2 中）

Element5（在 ListBuffer2 中）

Element6（在 ListBuffer2 中）

一些元素已被移至 RDD 中的新 ListBuffer，而两个元素已被完全删除。

我使用的是 Spark 1.6.0。

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

您可以将每个ListBuffer 转换为ListBuffer 的集合，比如ListBuffer 中的List，然后在RDD 上执行flatMap。

下面是虚拟 POC。

val rdd = spark.sparkContext.parallelize(Seq(List(1,2,3,4), List(11,22,76,44)))
val flattenRdd = rdd.map(s => List(s.filter(_%2 == 1), s.filter(_%2 == 0)))
    .flatMap(s => s)
flattenRdd.collect().foreach(s => println(s.mkString(",")))

1,3
2,4
11
22,76,44

【讨论】：

如果我不知道 Seq 中有多少个 List 怎么办？就我而言，这可能在 1 到数千之间。
假设我运行 rdd.map(s => func(s)).flatMap(s => s) 并且函数将一个 s 拆分为多个 s。那会奏效吗？例如，如果 func 将一个 Seq 作为输入并返回一个包含 3 个 Seq 的列表。（一个Seq被分成3个）
这太棒了，非常感谢，让它与我的真实代码一起工作！