火花窗口函数：筛选出在其他行的开始和结束日期范围内的开始和结束日期的行

Question

我有一个类似于以下内容的DataFrame（sqlDF）（在本例中进行了简化），其中我试图删除在另一行的开始日期和结束日期范围内具有start_date和end_date的所有行：

+-------+-------------+-------------------+-------------------+
|    id |         type|         start_date|           end_date|
+-------+-------------+-------------------+-------------------+
|  1    |      unknown|2018-11-14 16:03:47|2018-12-06 21:23:22| (remove as it's within the next rows start and end dates)
|  1    |          ios|2018-10-13 14:58:22|2019-08-26 15:50:45|
|  1    |      android|2019-08-29 02:41:40|2019-09-05 23:03:20|
|  2    |          ios|2017-12-19 02:25:34|2019-08-09 15:41:30|
|  2    |      windows|2018-07-10 05:30:52|2018-07-13 10:11:34| (remove as it's within the previous row's start and end dates)
|  2    |      android|2019-05-14 18:33:15|2019-08-27 06:10:53| (remove as it's within another row's start and end dates)

首先，最终用户要求我删除start_date和end_date之间间隔小于5天的所有记录，我使用了以下方法：

val dfWithoutTempHandsets = sqlDF.filter(datediff(col("end_date"), col("start_date")) > 5)

从而产生如下所示的数据frame：

+-------+-------------+-------------------+-------------------+
|    id |         type|         start_date|           end_date|
+-------+-------------+-------------------+-------------------+
|  1    |      unknown|2018-11-14 16:03:47|2018-12-06 21:23:22| 
|  1    |          ios|2018-10-13 14:58:22|2019-08-26 15:50:45|
|  1    |      android|2019-08-29 02:41:40|2019-09-05 23:03:20|
|  2    |          ios|2017-12-19 02:25:34|2019-08-09 15:41:30|
|  2    |      android|2019-05-14 18:33:15|2019-06-27 06:10:53|

现在，我需要筛选出开始日期和结束日期在同一id的另一行开始日期和结束日期“内”的行，这样得到的DataFrame看起来如下所示：

+-------+-------------+-------------------+-------------------+
|    id |         type|         start_date|           end_date|
+-------+-------------+-------------------+-------------------+
|  1    |          ios|2018-10-13 14:58:22|2019-08-26 15:50:45|
|  1    |      android|2019-08-29 02:41:40|2019-09-05 23:03:20|
|  2    |          ios|2017-12-19 02:25:34|2019-08-09 15:41:30|

在阅读了几篇关于spark窗口函数的博客文章和堆栈溢出文章后，我知道这就是答案。但是我很难找到类似用例的例子，其中多个日期以这种方式与另一行的日期进行比较。我相信我有一个关闭的windowSpec：

val windowSpec = Window.partitionBy("id", "type").orderBy("start_date")

但是，从那里我不确定如何利用windowSpec只选择那些在其他行中没有开始日期和结束日期的行。

编辑：我得到了一个新的要求，只对具有“null”或“unknown”类型的行应用上述逻辑。但这里的答案让我离得更近了！

Answer 1

下面是我考虑使用的逻辑：

在按ID且start_date升序排列的窗口分区下，如果当前行中的end_date早于或等于前面任何行中的end_date，则当前行中的日期范围必须包含在前面行中的某个日期范围内。

将其转换为示例代码（还包括>5天筛选）：

import java.sql.Timestamp
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
import spark.implicits._

val df = Seq(
  (1, "unknown", Timestamp.valueOf("2018-11-14 16:03:47"), Timestamp.valueOf("2018-12-06 21:23:22")),
  (1, "ios", Timestamp.valueOf("2018-10-13 14:58:22"), Timestamp.valueOf("2019-08-26 15:50:45")),
  (1, "android", Timestamp.valueOf("2019-08-29 02:41:40"), Timestamp.valueOf("2019-09-05 23:03:20")),
  (2, "ios", Timestamp.valueOf("2017-12-19 02:25:34"), Timestamp.valueOf("2019-08-09 15:41:30")),
  (2, "unknown", Timestamp.valueOf("2018-07-10 05:30:52"), Timestamp.valueOf("2018-07-13 10:11:34")),
  (2, "android", Timestamp.valueOf("2019-05-14 18:33:15"), Timestamp.valueOf("2019-06-27 06:10:53"))
).toDF("id", "type", "start_date", "end_date")

val win = Window.partitionBy("id").orderBy($"start_date").
  rowsBetween(Window.unboundedPreceding, -1)

df.
  where(unix_timestamp($"end_date") - unix_timestamp($"start_date") > 5*24*3600).
  withColumn("isContained",
    when($"end_date" <= max($"end_date").over(win), true).otherwise(false)
  ).
  where(! $"isContained").
  show
// +---+-------+-------------------+-------------------+-----------+
// | id|   type|         start_date|           end_date|isContained|
// +---+-------+-------------------+-------------------+-----------+
// |  1|    ios|2018-10-13 14:58:22|2019-08-26 15:50:45|      false|
// |  1|android|2019-08-29 02:41:40|2019-09-05 23:03:20|      false|
// |  2|    ios|2017-12-19 02:25:34|2019-08-09 15:41:30|      false|
// +---+-------+-------------------+-------------------+-----------+

请注意，对于>5天筛选，我使用的是UNIX_TIMESTAMP而不是datediff，后者仅机械地比较day值的差异（例如datediff(2019-01-06 12:00:00，2019-01-01 00:00:00)>5为false）。