如何将 (String, Array[String]) 的 RDD 拆分为数组中每个项目的 (String, String) 的 RDD？答案

【问题标题】：How to split RDD of (String, Array[String]) into RDD of (String, String) for each item in array?如何将 (String, Array[String]) 的 RDD 拆分为数组中每个项目的 (String, String) 的 RDD？
【发布时间】：2015-09-04 18:14:35
【问题描述】：

我有一个 RDD[(String, Array[String])] 形式的 PairRDD。我想展平这些值，以便我有一个RDD[(String, String)]，其中第一个 RDD 的 Array[String] 中的每个元素都成为第二个 RDD 中的专用元素。

例如，我的第一个 RDD 有以下元素：

("a", Array("x", "y"))
("b", Array("y", "z"))

我想要的结果是这样的：

("a", "x")
("a", "y")
("b", "y")
("b", "z")

我该怎么做？ flatMapValues(f: Array[String] => TraverableOnce[String]) 在这里似乎是正确的选择，但我需要使用什么作为参数 f？

【问题讨论】：

随便rdd.flatMapValues(x => x)
@kaktusito 好的，谢谢；我已经更新了这个问题，因为我实际上是在寻找传递给 flatMapValues() 的参数。你已经把它弄干净了。
@Carsten 我会使用identity 而不是x => x。 scala 编译器可能足够聪明，可以意识到这是identity，但可能不是，然后您创建一个新对象。
改用这个有什么区别：rdd.flatMap{ case (a,b) => b.map(a->_) }？ flatMapValues 有什么不同吗？
@tuxdna 我相信这是性能原因。 flatMap 不能保证保留原始 rdd 的分区器（因为无法检查密钥是否保持不变），而 flatMapValues 会。这在执行需要改组的操作时很重要，例如连接。

标签： scala apache-spark scala-collections

【解决方案1】：

要达到预期的效果，请执行以下操作：

val rdd1: RDD[(Any, Array[Any])] = ...
val rddFlat: RDD[(Any, Any)] = rdd1.flatMapValues(identity[Array[Any]])

结果看起来像问题中要求的结果。

【讨论】：

protip：它应该是一个 Wiki 答案，因为您只是收集了 cmets。