java - 如何使用 JAVA API 在 Spark 中使用另一个具有相同架构的数据集 的记录来更新数据集 ？

Question

假设您有一个包含以下记录的数据集 A:

Dataset A:
    {key1, val1}
    {key2, val2}
    {key3, val3}

Dataset B:
    {key4, val4}
    {key1, valBB}
    {key5, valN}
    {key2, NNNNN}

“更新”发生后，最终数据集应如下所示:

Dataset Final:
    {key1, valBB}
    {key2, NNNNN}
    {key3, val3}
    {key4, val4}
    {key5, valN}

到目前为止我采取的方法是将两个数据集转换为JavaRDD，然后转换JavaRDD -> JavaPairRDD，然后firstPairRDD.subtractByKey(secondPairRDD)。这为我提供了数据集 A 中存在但数据集 B 中不存在的记录。然后我将其重新转换回数据集。下一步是我与 DatasetB 进行联合以提供更新的数据集。对我来说，这并没有给我预期的结果。我采取了错误的方法吗？任何帮助将不胜感激。

Answer 1

我最终找到了一个更有效的解决方案:

    Dataset<Row> existsInAButNotB = A.join(B, A.col("key").equalTo(B.col("key") "left_anti");
    Dataset<Row> Final = existsInAButNotB.union(B); 

如果您有多个列用作 key ，那么您的解决方案应如下所示:

Dataset<Row> existsInAButNotB = A.join(B, A.col("key1").equalTo(B.col("key1").and(A.col("key2").equalTo(B.col("key2")) "left_anti");

这一行可以避免用户进入低效的 RDD 世界，并避免添加额外的代码。

看看这个:

Left Anti join in Spark?

有关左反连接的更多信息请参见此处:

what is the difference between an anti-join and an anti semi join?

数据集连接API: https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Dataset.html#join(org.apache.spark.sql.Dataset,%20org.apache.spark.sql.Column,%20java.lang.String)