过滤具有大于和小于日期列表的火花数据框

Question

我有一个dataframe 字段from_date 和to_date：

(2017-01-10     2017-01-14)
(2017-01-03     2017-01-13)

和日期列表

2017-01-05,
2017-01-12,
2017-01-13,
2017-01-15

这个想法是从表中检索该日期列表在 from_date 和 to_date 之间的所有行。

预期输出：

相同的数据框，但只有（from_date 和 to_date）在日期列表的值范围（=）内的行。 到目前为止，我尝试了 Nikk 的推荐：

Filter a spark dataframe with a greater than and a less than of list of dates

但我需要与整个日期列表进行比较，例如：

spark.sql("select * from dataframe_table where from_date  >= (select  date from date_list) AND  to_date  <= (select date from date_list)")

Answer 1

你的问题让我有点困惑，所以我提供了两个基于场景的代码。

1) 如果您想过滤介于您提供的列表范围之间的日期，例如从 2017 年 1 月 5 日到 2017 年 1 月 15 日，那么对于这种情况，下面的代码 sn-p 工作。

//Created dataframe view for both data
    Seq(("2017-01-10", "2017-01-14"),("2017-01-03","2017-01-13")).toDF("from_date","to_date").withColumn("from_date",'from_date.cast("date")).withColumn("to_date",'to_date.cast("date")).createOrReplaceTempView("date_table")


    List("2017-01-05","2017-01-12","2017-01-13","2017-01-15").toDF("list").createOrReplaceTempView("date_list")

   spark.sql("select * from date_table where (from_date BETWEEN (select min(cast(list as date)) from date_list) and (select max(cast(list as date)) from date_list)) and (to_date between (select min(cast(list as date)) from date_list) and (select max(cast(list as date)) from date_list))").show()
    +----------+----------+
    |from_date|  to_date|
    +----------+----------+
    |2017-01-10|2017-01-14|
    +----------+----------+

2) 或者，如果您想从 to_date 和 end_date 不在提供的日期列表中的数据框中过滤日期。因此，根据您提供的数据示例，列表之间将没有日期。对于这种情况，下面的代码将起作用。

//Created dataframe view for both data
       Seq(("2017-01-10", "2017-01-14"),("2017-01-03","2017-01-13")).toDF("from_date","to_date").withColumn("from_date",'from_date.cast("date")).withColumn("to_date",'to_date.cast("date")).createOrReplaceTempView("date_table")


        List("2017-01-05","2017-01-12","2017-01-13","2017-01-15").toDF("list").createOrReplaceTempView("date_list")

        spark.sql("select * from date_table where from_date in (select cast(list as date) from date_list) and to_date in (select cast(list as date) from date_list)").show() 

    +----------+----------+
    |from_date|  to_date|
    +----------+----------+
    |            |            |
    +----------+----------+

如果我遗漏了什么，请告诉我。

Answer 2

请检查：

//Creating DataFrame with Column from_date and to_date, you can ignore this step if you have dataframe

scala> val df =  Seq(("2017-01-10", "2017-01-14"),("2017-01-03","2017-01-13")).toDF("from_date","to_date").withColumn("from_date", col("from_date").cast("date")).withColumn("to_date",col("to_date").cast("date"))
df: org.apache.spark.sql.DataFrame = [from_date: date, to_date: date]

scala> df.show()
+----------+----------+
| from_date|   to_date|
+----------+----------+
|2017-01-10|2017-01-14|
|2017-01-03|2017-01-13|
+----------+----------+

//creating temparary View for dataframe "df" so that we can use it in spark sql.
scala> df.createOrReplaceTempView("dataframe_table")

//Converting List into Temp view
List("2017-01-05","2017-01-12","2017-01-13","2017-01-15").toDF("list").createOrReplaceTempView("date_list")


//Query to retrive all data from dataframe where from_date and to_date are in range of list.

scala> val output =  spark.sql("select * from dataframe_table where from_date  >= (select min(cast(list as date)) from date_list) AND  to_date  <= (select max(cast(list as date)) from date_list)")
output: org.apache.spark.sql.DataFrame = [from_date: date, to_date: date]

scala> output.show()
+----------+----------+                                                         
| from_date|   to_date|
+----------+----------+
|2017-01-10|2017-01-14|
+----------+----------+

Answer 3

如果您想将一个表的多行与另一个表的多行进行比较（让我们将您的日期列表视为具有单列的表），您可以在两个表上使用连接。通常，您会测试表的列是否相等。在这种情况下，您的测试有点特殊，因为您想将第一个表的两列与第二个表的一列进行比较。 您可以为此使用datediff：

scala> val df1 = Seq(("2017-01-10", "2017-01-14")).toDF("start_date","end_date").withColumn("start_date",'start_date.cast("date")).withColumn("end_date",'end_date.cast("date"))

df1: org.apache.spark.sql.DataFrame = [start_date: date, end_date: date]

scala> val df2 = Seq("2017-01-5", "2017-01-12","2017-01-13", "2017-01-15").toDF("from_date").withColumn("from_date",'from_date.cast("date"))
df2: org.apache.spark.sql.DataFrame = [from_date: date]

scala> df2.join(df1, datediff('from_date,'start_date) > 0) && datediff('from_date,'end_date) < 0)).show()