根据 spark 中的日期列比较 2 个数据帧并过滤结果答案

【问题标题】：Compare 2 dataframes and filter results based on date column in spark根据 spark 中的日期列比较 2 个数据帧并过滤结果
【发布时间】：2017-10-07 11:01:37
【问题描述】：

如下所述，我在 spark 中有 2 个数据帧。

val test = hivecontext.sql("select max(test_dt) as test_dt from abc");

test: org.apache.spark.sql.DataFrame = [test_dt: string]

val test1 = hivecontext.table("testing");

其中 test1 包含 id、name、age、audit_dt 等列

我想比较这两个数据帧并过滤来自 test1 的行，其中 audit_dt > test_dt。不知何故，我无法做到这一点。我可以使用 lit 函数将 audit_dt 与文字日期进行比较，但我无法将其与另一个数据框列进行比较。

我可以使用下面提到的 lit 函数比较文字日期

val output = test1.filter(to_date(test1("audit_date")).gt(lit("2017-03-23")))

测试数据框中的最大日期为 -> 2017-04-26

test1 Dataframe 中的数据 ->

Id,Name,Age,Audit_Dt
1,Rahul,23,2017-04-26
2,Ankit,25,2017-04-26
3,Pradeep,28,2017-04-27

我只需要 Id=3 的数据，因为只有行符合最大日期的大于标准。

我已经尝试过下面提到的选项，但它不起作用。

val test = hivecontext.sql("select max(test_dt) as test_dt from abc")
val MAX_AUDIT_DT = test.first().toString()
val output = test.filter(to_date(test("audit_date")).gt((lit(MAX_AUDIT_DT))))

任何人都可以建议将其与数据框测试列进行比较吗？

谢谢

【问题讨论】：

请分享示例数据和预期输出。
@mtoto 问题更新

标签： scala apache-spark apache-spark-sql

【解决方案1】：

如果"test_dt" 和"audit_date" 都属于date 类，则可以使用non-equi 连接。

/// cast to correct type
import org.apache.spark.sql.functions.to_date
val new_test = test.withColumn("test_dt",to_date($"test_dt"))
val new_test1 = test1.withColumn("Audit_Dt", to_date($"Audit_Dt"))

/// join
new_test1.join(new_test, $"Audit_Dt" > $"test_dt")
         .drop("test_dt").show()
+---+-------+---+----------+
| Id|   Name|Age|  Audit_Dt|
+---+-------+---+----------+
|  3|Pradeep| 28|2017-04-27|
+---+-------+---+----------+

数据

val test1 = sc.parallelize(Seq((1,"Rahul",23,"2017-04-26"),(2,"Ankit",25,"2017-04-26"),
             (3,"Pradeep",28,"2017-04-27"))).toDF("Id","Name", "Age", "Audit_Dt")
val test = sc.parallelize(Seq(("2017-04-26"))).toDF("test_dt")

【讨论】：

您的解决方案对我来说看起来不错。我会试试的。但我的两个日期列都是字符串类。我们可以在您提到的查询中使用 to_date 投射它们，然后进行比较。

【解决方案2】：

试试这个：

test1.filter(to_date(test1("audit_date")).gt(to_date(test("test_dt"))))

将值存储在变量中并在过滤器中使用。

val dtValue = test.select("test_dt")

或

val dtValue = test.first().getString(0)

现在应用过滤器

val output = test1.filter(to_date(test1("audit_date")).gt(lit(dtValue)))

【讨论】：

它也无法正常工作，因为 test_dt 是不同数据框的列。我也试过了。