加入 RDD 以产生交集的补码答案

【问题标题】：Join RDD's to result in complement of intersection加入 RDD 以产生交集的补码
【发布时间】：2017-08-09 18:33:31
【问题描述】：

我有两个 RDD：第一个（User ID、Mov ID、Rating、Timestamp）

data_wo_header: RDD[String]
scala> data_wo_header.take(5).foreach(println)
1,2,3.5,1112486027
1,29,3.5,1112484676
1,32,3.5,1112484819
1,47,3.5,1112484727
1,50,3.5,1112484580

和RDD2（用户ID，Mov ID）

data_test_wo_header: RDD[String]
scala> data_test_wo_header.take(5).foreach(println)
1,2
1,367
1,1009
1,1525
1,1750

我需要加入两个 RDD，这样加入将删除 RDD1 中常见的条目（UserID、Mov ID）。有人可以指导两个 RDD 的 scala-spark 连接。此外，我还需要一个连接，其中从 RDD1 派生的新 RDD 只有公共项目。

【问题讨论】：

标签： scala join apache-spark rdd

【解决方案1】：

首先将您的 RDD 转换为 DataFrame，因为 DataFrame 具有常见的 sql 类 API，例如 join、select 等。

要将您的 RDD 转换为 DataFrame，您需要一个 RDD[Row] 而不是 RDD[String]。

Import sqlContext.implicits._

case class cs1(UserID: Int, MovID: Int, Rating: String, Timestamp: String)

case class cs2(UserID: Int, MovID: Int)

val df1 = data_wo_header.map(row => {
   val splits = row.split(",")

   cs1(splits(0).toInt, splits(1).toInt, splits(2),splits(3))
}).toDF("UserID", "MovID", "Rating", "Timestamp")

 val df2 = data_test_wo_header.map(row => {
   val splits = row.split(",")

   cs2(splits(0).toInt, splits(1).toInt)
}).toDF("UserID", "MovID")

现在，向 df2 添加一个新列，

val df2Prime = df2.withColumn("isPresent", lit(1))

然后将 df2Prime 与 df1 左连接，并过滤掉 isPresent 为 1 的行，得到相交的结果。另外，删除临时 isPresent 标志。

val temp = df1.join(df2Prime, usingColumns = Seq("UserID", "MovID"), "left")

temp.filter(temp("isPresent") =!= "1").drop("isPresent")

【讨论】：

【解决方案2】：

一个超级简单的方法是使用按键减法。以下对我有用：

val data_wo_header=dropheader(data).map(_.split(",")).map(x=>((x(0),x(1)),(x(2),x(3))))
val data_test_wo_header=dropheader(data_test).map(_.split(",")).map(x=>((x(0),x(1)),1))
val ratings_train=data_wo_header.subtractByKey(data_test_wo_header)
val ratings_test=data_wo_header.subtractByKey(ratings_train)

【讨论】：