在 scala spark 数据框中提取时间间隔答案

【问题标题】：Extract time intervals in a scala spark dataframe在 scala spark 数据框中提取时间间隔
【发布时间】：2019-03-08 16:34:02
【问题描述】：

我正在尝试根据 scala 和 spark 中的时间序列提取组合数据间隔

我在数据框中有以下数据：

Id | State | StartTime           | EndTime
---+-------+---------------------+--------------------
 1 |   R   | 2019-01-01T03:00:00 | 2019-01-01T11:30:00
 1 |   R   | 2019-01-01T11:30:00 | 2019-01-01T15:00:00
 1 |   R   | 2019-01-01T15:00:00 | 2019-01-01T22:00:00
 1 |   W   | 2019-01-01T22:00:00 | 2019-01-02T04:30:00
 1 |   W   | 2019-01-02T04:30:00 | 2019-01-02T13:45:00
 1 |   R   | 2019-01-02T13:45:00 | 2019-01-02T18:30:00
 1 |   R   | 2019-01-02T18:30:00 | 2019-01-02T22:45:00

我需要根据 id 和 state 将数据提取到时间间隔中。生成的数据需要如下所示：

Id | State | StartTime           | EndTime
---+-------+---------------------+--------------------
 1 |   R   | 2019-01-01T03:00:00 | 2019-01-01T22:00:00
 1 |   W   | 2019-01-01T22:00:00 | 2019-01-02T13:45:00
 1 |   R   | 2019-01-02T13:45:00 | 2019-01-02T22:45:00

请注意，前三个记录已分组在一起，因为设备从 2019-01-01T03:00:00 到 2019-01-01T22:00:00 连续处于 R 状态，然后切换到 W 状态接下来的两条记录从 2019-01-01T22:00:00 到 2019-01-02T13:45:00，然后返回到最后两条记录的 R 状态。

【问题讨论】：

Spark SQL window function with complex condition的可能重复
我看了那个问题，这是一个非常不同的问题
在这种情况下可以编辑您的问题并详细解释所需的逻辑吗？此外，我们始终欢迎 reproducible example 加入 apache-spark。提前谢谢你。
原来的解决方案是这样的：stackoverflow.com/questions/7420618/… 翻译成 spark
嗨@JeffHornby，您是否设法将其转换为 Spark 代码？

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

【解决方案1】：

所以事实证明，这个问题的答案是 Combine rows when the end time of one is the start time of another (Oracle) 翻译成 Spark。

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{col,row_number}
import spark.implicits._

val idSpec = Window.partitionBy('Id).orderBy('StartTime)
val idStateSpec = Window.partitionBy('Id,'State).orderBy('StartTime)
val df2 = df
  .select('Id,'State,'StartTime,'EndTime,
          row_number().over(idSpec).as("idRowNumber"),
          row_number().over(idStateSpec).as("idStateRowNumber"))
  .groupBy('Id,'State,'idRowNumber - 'idStateRowNumber)
  .agg(min('StartTime).as("StartTime"), max('EndTime).as("EndTime"))

【讨论】：