scala/spark 中的平面映射

flatMapping in scala/spark

正在寻求有关如何使用 spark 在 scala 中执行某些操作的问题的帮助。

我有：

type DistanceMap = HashMap[(VertexId,String), Int]

这构成了我的数据的一部分，形式为 RDD：

org.apache.spark.rdd.RDD[(DistanceMap, String)]

简而言之，我的数据集如下所示：

 ({(101,S)=3},piece_of_data_1)
 ({(101,S)=3},piece_of_data_2)
 ({(101,S)=1, (100,9)=2},piece_of_data_3)

我想做什么我们平面映射我的距离图（我可以做）但同时对于每个平面映射 DistanceMap 想保留与之相关的字符串。所以我的结果数据将如下所示：

({(101,S)=3},piece_of_data_1))<br>
({(101,S)=3},piece_of_data_2))<br>
({(101,S)=1},piece_of_data_3))<br>
({(109,S)=2},piece_of_data_3))<br>

如前所述，我可以使用以下方法对第一部分进行平面映射：

x.flatMap(x=>x._1).collect.foreach(println))

但我对如何保留原始数据第二部分的字符串感到困惑。

更新

我有一个替代方案——翻转键值并使用flatMapValues转换，然后再次翻转键值：见伪代码：

x.map(x=>x._2, x._1).flatMapValues(x=>x).map(x=>x._2, x._1)

以前的版本

我建议添加一个预处理步骤（抱歉，明天我面前没有装有 scala 解释器的计算机来编写工作代码）。

  1. 将 (DistanceMap, String) 中的 rdd 对转换为具有 Tuple4 列表的 rdd：List((VertexId,String, Int, String), ... ())
  2. 在结果上应用 flatMap

伪代码：

rdd.map( (DistanceMap, String) =>  List((VertexId,String, Int, String), ... ()))
    .flatMap(x=>x)

这可能对你有用：

x.flatMap(x => x._1.map(y => (y,x._2)))

想法是从 (Seq(a,b,c),Value) 转换为 Seq( (a,Value), (b, Value), (c, Value))。

这在 Scala 中是一样的，所以这是一个独立的简化 Scala 示例，您可以将其粘贴到 Scala REPL 中：

Seq((Seq("a","b","c"), 34), (Seq("r","t"), 2)).flatMap( x => x._1.map(y => (y,x._2)))

这导致：

res0: Seq[(String, Int)] = List((a,34), (b,34), (c,34), (r,2), (t,2))