如何将 Scala RDD 转换为 Map答案

【问题标题】：How to convert Scala RDD to Map如何将 Scala RDD 转换为 Map
【发布时间】：2014-10-14 01:36:08
【问题描述】：

我有一个 RDD（字符串数组）org.apache.spark.rdd.RDD[String] = MappedRDD[18] 并将其转换为具有唯一 ID 的地图。我做了'val vertexMAp = vertices.zipWithUniqueId' 但这给了我另一个 'org.apache.spark.rdd.RDD[(String, Long)]' 类型的 RDD，但我想要一个 'Map[String, Long]' 。如何转换我的“org.apache.spark.rdd.RDD[(String, Long)] to Map[String, Long]”？

谢谢

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

PairRDDFunctions 中有一个内置的 collectAsMap 函数，可以为您提供 RDD 中的对值映射。

val vertexMAp = vertices.zipWithUniqueId.collectAsMap

请务必记住，RDD 是一种分布式数据结构。您可以将其可视化为分布在集群中的数据的“片段”。当您collect 时，您会强制所有这些部分交给驱动程序，并且为了能够做到这一点，它们需要适合驱动程序的内存。

从 cmets 看来，在您的情况下，您需要处理一个大型数据集。用它制作地图是行不通的，因为它不适合驱动程序的内存；如果您尝试，则会导致 OOM 异常。

您可能需要将数据集保留为 RDD。如果您正在创建 Map 以查找元素，则可以在 PairRDD 上使用 lookup，如下所示：

import org.apache.spark.SparkContext._  // import implicits conversions to support PairRDDFunctions

val vertexMap = vertices.zipWithUniqueId
val vertixYId = vertexMap.lookup("vertexY")

【讨论】：

如果你的左元组中已经有唯一值，你需要使用 zipWithUniqueId 吗？
@maasg 是否可以跨 RDD 中的工作节点进行查找？
@santhosh 是的。 lookup 适用于完整的分布式 RDD
嗨@alex9311，您找到问题的答案了吗？

【解决方案2】：

收集到“本地”机器，然后将 Array[(String, Long)] 转换为 Map

val rdd: RDD[String] = ???

val map: Map[String, Long] = rdd.zipWithUniqueId().collect().toMap

【讨论】：

我的 RDD 有 19123380 条记录，当我运行 val map: Map[String, Long] = rdd.zipWithUniqueId().collect().toMap 时，我得到一个 lang.OutOfMemoryError。有更好的方法吗？
没有。您可以使用 NoSql 存储（例如 Cassandra）来加载您的 RDD 并通过类似 Map 的界面访问它。
嗨，Eugene，您能否详细说明您的评论？
您可以使用 github.com/datastax/spark-cassandra-connector 将您的 RDD[(String, Long)] 保存为 cassandra 表。稍后将其用作快速按键查找

【解决方案3】：

您不需要转换。 PairRDDFunctions 的隐式检测基于二元组的 RDD 并自动应用 PairRDDFunctions 方法。

【讨论】：