1

我有两个火花数据框，具有以下结构。如使用 sqlContext 之前所读。

 itens.columns (scala command) 
 Array[String] = Array(id_location,id_item, name, price)

 rdd1 
 [1,1,item A,10]
 [1,2,item b,12]
 [1,3,item c,12]

 rdd2
 [1,2,item b,50]
 [1,4,item c,12]
 [1,5,item c,12]

我想要基于复合键 (id_location,id_item) 的以下结果

 [1,1,item A,10]
 [1,2,item b,50]
 [1,3,item c,12]
 [1,4,item c,12]
 [1,5,item c,12]

所以，我想要一个具有不同itens的结果（关于复合键），但是当我在两个rdds中找到具有相同键的记录时，我只想保留来自rdd2的记录。

有人有这种要求吗？

我正在使用 spark 和 scala。

最好的问候拉斐尔。

4

3 回答 3

1

我对 Spark 很陌生，所以可能有更好的方法来做到这一点，但是您是否可以映射到一对 RDD（基于您的复合键），然后执行 fullOuterJoin，仅使用结果数据在哪里有“左侧”和“右侧”的数据？

粗略的伪代码：

val pairRdd1 = rdd1 map {
  line => 
    (line(0)+line(1), line)
}

val pairRdd2 = rdd2 map {
  line => 
    (line(0)+line(1), line)
}

val joined = pairRdd1.fullOuterJoin(pairRdd2)

joined map {
  (id, left, right) =>
    right.getOrElse(left.get)
}

如果我早上有时间，我会尝试拼凑一个工作示例。希望有帮助！

于 2015-10-21T23:42:36.593 回答
0

@Steven 有正确的想法。您需要将数据集映射到键值对，然后执行outerjoin

val rdd1 = sc.parallelize(List((1,1,"item A",10),(1,2,"item b",12),(1,3,"item c",12)))
val rdd2 = sc.parallelize(List((1,2,"item b",50),(1,4,"item c",12),(1,5,"item c",12)))

val rdd1KV = rdd1.map{case(id_location,id_item, name, price) => ((id_location, id_item), (name, price))}
val rdd2KV = rdd2.map{case(id_location,id_item, name, price) => ((id_location, id_item), (name, price))}

val joined = rdd1KV.fullOuterJoin(rdd2KV)

val res = joined.map{case((id_location, id_item),(leftOption, rightOption)) =>
    val values = rightOption.getOrElse(leftOption.get)
    (id_location, id_item, values._1, values._2)
}

这将为您提供您正在寻找的结果。

于 2015-10-22T06:51:11.957 回答
0

看起来@Steven 的答案在逻辑上很好，但如果您的数据没有很多相交元素（即完全外连接将产生巨大的数据集），则可能会遇到问题。您也在使用 DataFrames，因此转换为 RDDs 然后再转换回 DataFrames 对于可以使用 DataFrames API 完成的任务来说似乎是多余的。我将在下面描述如何做到这一点。

让我们从一些示例数据开始（取自您的示例）：

val rdd1 = sc.parallelize(Array((1,1,"item A",10), (1,2,"item b",12), (1,3,"item c",12)))
val rdd2 = sc.parallelize(Array((1,2,"item b",50), (1,4,"item c",12), (1,5,"item c",12)))

接下来，我们可以将它们转换为单独的列别名下的 DataFrame。df1我们在此处和此处使用不同的别名，df2因为当我们最终连接这两个 DataFrame 时，可以更轻松地编写后续的选择（如果有一种方法可以在连接后识别列的来源，这是不必要的）。请注意，两个 DataFrame 的并集包含您要过滤的行。

val df1 = rdd1.toDF("id_location", "id_item", "name", "price")
val df2 = rdd2.toDF("id_location_2", "id_item_2", "name_2", "price_2")

// df1.unionAll(df2).show()
// +-----------+-------+------+-----+
// |id_location|id_item|  name|price|
// +-----------+-------+------+-----+
// |          1|      1|item A|   10|
// |          1|      2|item b|   12|
// |          1|      3|item c|   12|
// |          1|      2|item b|   50|
// |          1|      4|item c|   12|
// |          1|      5|item c|   12|
// +-----------+-------+------+-----+

在这里，我们首先将两个 DataFrame 连接到作为 和 的前两个元素的键df1上df2。df1然后，我们通过选择存在df2具有相同连接键的行的行（基本上是 from ）来创建另一个 DataFrame 。之后，我们运行一个 except ondf1以从先前创建的 DataFrame 中删除所有行。这可以看作是一种补充，因为我们基本上所做的就是删除df1存在相同("id_location", "id_item")in的所有行df2。最后，我们将补码与 结合在一起df2以产生输出 DataFrame。

val df_joined = df1.join(df2, (df1("id_location") === df2("id_location_2")) && (df1("id_item") === df2("id_item_2")))
val df1_common_keyed = df_joined.select($"id_location", $"id_item", $"name", $"price")
val df1_complement = df1.except(df1_common_keyed)
val df_union = df1_complement.unionAll(df2)

// df_union.show()
// +-----------+-------+------+-----+
// |id_location|id_item|  name|price|
// +-----------+-------+------+-----+
// |          1|      3|item c|   12|
// |          1|      1|item A|   10|
// |          1|      2|item b|   50|
// |          1|      4|item c|   12|
// |          1|      5|item c|   12|
// +-----------+-------+------+-----+

同样，就像@Steven 建议的那样，您可以通过将 DataFrames 转换为 RDD 并使用它运行来使用 RDD API。如果这是您想要做的，以下是完成您想要使用的另一种方法subtractByKey()以及上面的输入 RDD：

val keyed1 = rdd1.keyBy { case (id_location, id_item, _, _) => (id_location, id_item) }
val keyed2 = rdd2.keyBy { case (id_location, id_item, _, _) => (id_location, id_item) }
val unionRDD = keyed1.subtractByKey(keyed2).values.union(rdd2)

// unionRDD.collect().foreach(println)
// (1,1,item A,10)
// (1,3,item c,12)
// (1,2,item b,50)
// (1,4,item c,12)
// (1,5,item c,12)
于 2015-10-22T06:59:29.557 回答