将 RDD 与缺少的一些值相结合答案

【问题标题】：Combining RDD's with some values missing将 RDD 与缺少的一些值相结合
【发布时间】：2017-07-31 18:35:01
【问题描述】：

您好，我有两个 RDD，我想合并为 1。第一个 RDD 的格式为

//((UserID,MovID),Rating)
val predictions =
model.predict(user_mov).map { case Rating(user, mov, rate) =>
  ((user, mov), rate)
}

我还有一个 RDD

//((UserID,MovID),"NA")
val user_mov_rat=user_mov.map(x=>(x,"N/A"))

所以第二个 RDD 中的键更多的是 no。但与 RDD1 重叠。我需要组合RDD，以便只有第二个RDD的那些键附加到RDD1中，而RDD1中不存在这些键。

【问题讨论】：

标签： scala apache-spark rdd

【解决方案1】：

你可以这样做 -

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.col

// Setting up the rdds as described in the question
case class UserRating(user: String, mov: String, rate: Int = -1)

val list1 = List(UserRating("U1", "M1", 1),UserRating("U2", "M2", 3),UserRating("U3", "M1", 3),UserRating("U3", "M2", 1),UserRating("U4", "M2", 2))

val list2 = List(UserRating("U1", "M1"),UserRating("U5", "M4", 3),UserRating("U6", "M6"),UserRating("U3", "M2"), UserRating("U4", "M2"), UserRating("U4", "M3", 5))

val rdd1 = sc.parallelize(list1)
val rdd2 = sc.parallelize(list2)

// Convert to Dataframe so it is easier to handle    
val df1 = rdd1.toDF
val df2 = rdd2.toDF

// What we got:
df1.show
+----+---+----+
|user|mov|rate|
+----+---+----+
|  U1| M1|   1|
|  U2| M2|   3|
|  U3| M1|   3|
|  U3| M2|   1|
|  U4| M2|   2|
+----+---+----+

df2.show
+----+---+----+
|user|mov|rate|
+----+---+----+
|  U1| M1|  -1|
|  U5| M4|   3|
|  U6| M6|  -1|
|  U3| M2|  -1|
|  U4| M2|  -1|
|  U4| M3|   5|
+----+---+----+

// Figure out the extra reviews in second dataframe that do not match (user, mov) in first    
val xtraReviews = df2.join(df1.withColumnRenamed("rate", "rate1"), Seq("user", "mov"), "left_outer").where("rate1 is null")

// Union them. Be careful because of this: http://stackoverflow.com/questions/32705056/what-is-going-wrong-with-unionall-of-spark-dataframe

def unionByName(a: DataFrame, b: DataFrame): DataFrame = {
    val columns = a.columns.toSet.intersect(b.columns.toSet).map(col).toSeq
    a.select(columns: _*).union(b.select(columns: _*))
}

// Final result of combining only unique values in df2    
unionByName(df1, xtraReviews).show

+----+---+----+
|user|mov|rate|
+----+---+----+
|  U1| M1|   1|
|  U2| M2|   3|
|  U3| M1|   3|
|  U3| M2|   1|
|  U4| M2|   2|
|  U5| M4|   3|
|  U4| M3|   5|
|  U6| M6|  -1|
+----+---+----+

【讨论】：

【解决方案2】：

也可以这样：

RDD 真的很慢，因此请读取数据或将数据转换为数据帧。
在 df.dropDuplicates(['Key1', 'Key2']) 等两个数据帧上使用 spark dropDuplicates() 以获取两个数据帧中键的不同值，然后
只需像 df1.union(df2) 一样将它们联合起来。

好处是您以 Spark 方式进行操作，因此您拥有所有的并行性和速度。

【讨论】：