比较 Spark 中的 DataFrame

Question

我有 2 个数据框

df1

+----------+----------------+--------------------+--------------+-------------+
|      WEEK|DIM1            |DIM2                |T1            |  T2         |
+----------+----------------+--------------------+--------------+-------------+
|2016-04-02|              14|                NULL|          9874|   880       |
|2016-04-30|              14|FR                  |          9875|    13       |
|2017-06-10|              15|                 PQR|          9867| 57721       |
+----------+----------------+--------------------+--------------+-------------+

df2

+----------+----------------+--------------------+--------------+-------------+
|      WEEK|DIM1            |DIM2                |T1            |  T2         |
+----------+----------------+--------------------+--------------+-------------+
|2016-04-02|              14|                NULL|          9879|   820       |
|2016-04-30|              14|FR                  |          9785|    9        |
|2017-06-10|              15|                 XYZ|          9967| 57771       |
+----------+----------------+--------------------+--------------+-------------+

我想在 spark 中编写一个比较器，在 WEEK 时比较两个数据帧中的 T1、T2、DIM1、DIM2 与 T1、df1 中的 T2 应该比 T1、T2 大 3。我想返回所有不将上述标准与数据帧之间的 T1、T2 之间的差异相匹配。对于以下组合 WEEK、DIM1、DIM2，我还希望 df1 中存在的行不存在于 df2 中，反之亦然。

输出应该是这样的

+----------+----------------+--------------------+--------------+-------------+------------------+-----------------+
|      WEEK|DIM1            |DIM2                |T1_dIFF       |  T2_dIFF    | Presenent_In_DF1 | Presenent_In_DF2|
+----------+----------------+--------------------+--------------+-------------+------------------+-----------------+
|2016-04-30|              14|FR                  |            90|    4        | Y                | Y               |
|2017-06-10|              15|PQR                 |          9867|    57721    | Y                | N               |
|2017-06-10|              15|XYZ                 |          9967|    57771    | N                | Y               |
+----------+----------------+--------------------+--------------+-------------+------------------+-----------------+

解决这个问题的最佳方法是什么？

我已经实现了以下，但不知道在此之后如何进行 -

val df1 = Seq(
  ("2016-04-02", "14", "NULL", 9874, 880), ("2016-04-30", "14", "FR", 9875, 13), ("2017-06-10", "15", "PQR", 9867, 57721)
).toDF("WEEK", "DIM1", "DIM2","T1","T2")

val df2 = Seq(
  ("2016-04-02", "14", "NULL", 9879, 820), ("2016-04-30", "14", "FR", 9785, 9), ("2017-06-10", "15", "XYZ", 9967, 57771)
).toDF("WEEK", "DIM1", "DIM2","T1","T2")

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

val joined = df1.as("l").join(df2.as("r"), Seq("WEEK", "DIM1", "DIM2"), "fullouter")

加入后的样子是这样的 -

+----------+----+----+----+-----+----+-----+
|      WEEK|DIM1|DIM2|  T1|   T2|  T1|   T2|
+----------+----+----+----+-----+----+-----+
|2016-04-02|  14|NULL|9874|  880|9879|  820|
|2017-06-10|  15| PQR|9867|57721|null| null|
|2017-06-10|  15| XYZ|null| null|9967|57771|
|2016-04-30|  14|  FR|9875|   13|9785|    9|
+----------+----+----+----+-----+----+-----+

我不知道如何在这之后以一种好的方式进行，对 scala 来说相对较新。

Answer 1

一个简单的解决方案可能是将 df1 和 df2 与 WEEK 作为唯一键连接。在连接的数据中，您需要保留 df1 和 df2 中的所有列。

然后您可以对数据框执行映射操作以生成其余列。

类似

df1.createOrReplaceTempTable("df1")
df2.createOrReplaceTempTable("df2")
val df = spark.sql("select df1.*, df2.DIM1 as df2_DIM1, df2.DIM2 as df2_DIM2, df2.T1 as df2_T1, df2.T2 as df2_T2 from df1 join df2 on df1.WEEK = df2.WEEK")
// Now map on the dataframe to produce the diff dataframe
// Or you can use the SQL to do that.