加入两个 RDD[String] -Spark Scala答案

【问题标题】：Joining two RDD[String] -Spark Scala加入两个 RDD[String] -Spark Scala
【发布时间】：2016-09-07 05:22:22
【问题描述】：

我有两个 RDDS：

rdd1 [String,String,String]: Name, Address, Zipcode
rdd2 [String,String,String]: Name, Address, Landmark

我正在尝试使用以下功能加入这两个 RDD：rdd1.join(rdd2)
但我收到一个错误：
error: value fullOuterJoin is not a member of org.apache.spark.rdd.RDD[String]

join 应该加入 RDD[String] 并且输出 RDD 应该是这样的：

rddOutput : Name,Address,Zipcode,Landmark

最后我想将这些文件保存为 JSON 文件。

有人可以帮助我吗？

【问题讨论】：

join 是在对 RDD 上定义的，因此您的 rdd1 不是 RDD[(String, T)] 类型。您应该像这样 rdd1.map(v => (v, 1)) 映射它（或映射到另一个元组，这取决于您的任务）。如果您更详细地解释您的目标（您希望从加入中获得什么），您可能会获得更多帮助。
@VitaliyKotlyarenko：很抱歉没有早点澄清。我刚刚编辑了这个问题。你能帮帮我吗？
您的编辑没有多大帮助。您没有RDD[String]，而是两个RDD[String, String, String]。您想加入哪些领域？ Name 和 Address，还是其中之一？您需要将 RDD 更改为具有元组的条目，其中第一个是键，其余的是 value，然后加入将起作用。
我想加入姓名和地址。

标签： string scala join apache-spark rdd

【解决方案1】：

正如 cmets 中所说，您必须在加入之前将您的 RDD 转换为 PairRDD，这意味着每个 RDD 必须是 RDD[(key, value)] 类型。只有这样，您才能通过密钥执行连接。在您的情况下，密钥由（名称，地址）组成，因此您必须执行以下操作：

// First, we create the first PairRDD, with (name, address) as key and zipcode as value:
val pairRDD1 = rdd1.map { case (name, address, zipcode) => ((name, address), zipcode) }
// Then, we create the second PairRDD, with (name, address) as key and landmark as value:
val pairRDD2 = rdd2.map { case (name, address, landmark) => ((name, address), landmark) }

// Now we can join them. 
// The result will be an RDD of ((name, address), (zipcode, landmark)), so we can map to the desired format:
val joined = pairRDD1.fullOuterJoin(pairRDD2).map { 
  case ((name, address), (zipcode, landmark)) => (name, address, zipcode, landmark) 
}

有关 Spark 的 Scala API documentation 中的 PairRDD 函数的更多信息

【讨论】：

您好@Daniel：我之前正在使用此函数处理RDD，以输出RDD1。有什么办法可以调整这个吗？ : i.map(r => { r._1 + "|" + r._2 + "|" + r._3+ "|" + sys.extractDocument(r._3) });
是的，将您的地图结果设为(key, value) 对。这已经说了好几遍了，也许你能注意到它？
@user2122466，对不起，我不明白你的评论。也许你可以改进你的问题？
@DanieldePaula .map 不会导致 PairRDD 并且是 RDD，您是在说 .mapToPair 吗？
@deFreitas 在 Scala 中，对的 RDD 会隐式转换为 PairRDD，因此无需调用 mapToPair，只需普通的 map 返回一对即可。如果您使用 Java API，则需要 mapToPair。