根据不包括重复项的两列值合并两个 Spark 数据集

Question

我有两个数据集，

Dataset1
+---+------+------+
|cid|itemId|bought|
+---+------+------+
|abc|   123|  true|
|abc|   345|  true|
|abc|   567|  true|
|def|   123|  true|
|def|   345|  true|
|def|   567|  true|
+---+------+------+

Dataset2
+---+------+------+
|cid|itemId|bought|
+---+------+------+
|abc|   123| false|
|abc|   345| false|
|def|   789| false|
+---+------+------+

我的目标是根据以下标准合并这两个数据集。

CID 和 ItemId 形成一个组合（组键）。 Dataset2 将检查 dataset1 中是否存在组键，如果存在，则合并这些记录时将被删除，而与第三列中的值无关。

例如在上面的示例中，dataset2 的第 1 行和第 2 行将被删除，并且只有第 3 列将被合并。

所需的输出数据集

+---+------+------+
|cid|itemId|bought|
+---+------+------+
|abc|   123|  true|
|abc|   345|  true|
|abc|   567|  true|
|def|   123|  true|
|def|   345|  true|
|def|   567|  true|
|def|   789| false|
+---+------+------+

这是我正在尝试的逻辑，

根据 cid 和 itemid 值与 dataset1 上的交集过滤 dataset2，并将其存储为 tempDataset。但是我无法获得可以通过这种方式工作的过滤器。仍在寻找它。

对 tempDataset 和 dataset1 应用联合并创建 dataset3。

我想知道有没有更简单的方法？

还有如何过滤得到tempDataset？

Answer 1

一种方法是先执行left_anti join，然后执行union，如下所示：

import org.apache.spark.sql.functions._

val df1 = Seq(
  ("abc", 123, true),
  ("abc", 345, true),
  ("abc", 567, true),
  ("def", 123, true),
  ("def", 345, true),
  ("def", 567, true)
).toDF("cid", "itemId", "bought")

val df2 = Seq(
  ("abc", 123, false),
  ("abc", 345, false),
  ("def", 789, false)
).toDF("cid", "itemId", "bought")

df1.union( df2.join(df1, Seq("cid", "itemId"), "left_anti") ).
  show
// +---+------+------+
// |cid|itemId|bought|
// +---+------+------+
// |abc|   123|  true|
// |abc|   345|  true|
// |abc|   567|  true|
// |def|   123|  true|
// |def|   345|  true|
// |def|   567|  true|
// |def|   789| false|
// +---+------+------+