apache-spark - Spark SQL 数据帧 : best way to compute across rowpairs

Question

我有一个像这样的 Spark DataFrame“deviceDF”:

ID    date_time            state  
a     2015-12-11 4:30:00     up  
a     2015-12-11 5:00:00     down  
a     2015-12-11 5:15:00     up  
b     2015-12-12 4:00:00     down  
b     2015-12-12 4:20:00     up  
a     2015-12-12 10:15:00    down  
a     2015-12-12 10:20:00    up  
b     2015-12-14 15:30:00    down  

我正在尝试计算每个 ID 的停机时间。我从基于 id 的分组开始，然后分别计算所有正常运行时间和停机时间的总和。然后计算正常运行时间和停机时间之和。

val downtimeDF = deviceDF.filter($"state" === "down")
  .groupBy("ID")
  .agg(sum(unix_timestamp($"date_time")) as "down_time")  

val uptimeDF = deviceDF.filter($"state" === "up")
  .groupBy("ID")
  .agg(sum(unix_timestamp($"date_time")) as "up_time")  

val updownjoinDF = uptimeDF.join(downtimeDF, "ID")  

val difftimeDF = updownjoinDF
  .withColumn("diff_time", $"up_time" - $"down_time")  

然而，很少有情况会导致错误，例如设备出现故障但从未恢复，在这种情况下，down_time 是 current_time 和 last_time 之间的差异。

此外，如果特定设备的第一个条目以“up”开头，则 down_time 是第一个条目与此分析开始时的时间之差，例如 2015-12-11 00:00:00。使用数据框处理这些边界条件的最佳方法是什么？我需要编写自定义 UDAF 吗？

Answer 1

您可以尝试的第一件事是使用窗口函数。虽然这通常不是最快的解决方案，但它简洁且极具表现力。以你的数据为例:

import org.apache.spark.sql.functions.unix_timestamp

val df = sc.parallelize(Array(
    ("a", "2015-12-11 04:30:00", "up"), ("a", "2015-12-11 05:00:00", "down"), 
    ("a", "2015-12-11 05:15:00", "up"), ("b", "2015-12-12 04:00:00", "down"), 
    ("b", "2015-12-12 04:20:00", "up"), ("a", "2015-12-12 10:15:00", "down"),
    ("a", "2015-12-12 10:20:00", "up"), ("b", "2015-12-14 15:30:00", "down")))
  .toDF("ID", "date_time", "state")
  .withColumn("timestamp", unix_timestamp($"date_time"))

让我们定义示例窗口:

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{coalesce, lag, when, sum}

val w = Window.partitionBy($"ID").orderBy($"timestamp")

一些辅助列

val previousTimestamp = coalesce(lag($"timestamp", 1).over(w), $"timestamp")
val previousState = coalesce(lag($"state", 1).over(w), $"state")

val downtime = when(
  previousState === "down",
  $"timestamp" - previousTimestamp
).otherwise(0).alias("downtime")

val uptime = when(
  previousState === "up",
  $"timestamp" - previousTimestamp
).otherwise(0).alias("uptime")

最后是一个基本查询:

val upsAndDowns = df.select($"*", uptime, downtime)
upsAndDowns.show

// +---+-------------------+-----+----------+------+--------+
// | ID|          date_time|state| timestamp|uptime|downtime|
// +---+-------------------+-----+----------+------+--------+
// |  a|2015-12-11 04:30:00|   up|1449804600|     0|       0|
// |  a|2015-12-11 05:00:00| down|1449806400|  1800|       0|
// |  a|2015-12-11 05:15:00|   up|1449807300|     0|     900|
// |  a|2015-12-12 10:15:00| down|1449911700|104400|       0|
// |  a|2015-12-12 10:20:00|   up|1449912000|     0|     300|
// |  b|2015-12-12 04:00:00| down|1449889200|     0|       0|
// |  b|2015-12-12 04:20:00|   up|1449890400|     0|    1200|
// |  b|2015-12-14 15:30:00| down|1450103400|213000|       0|
// +---+-------------------+-----+----------+------+--------+

以类似的方式，您可以向前看，如果组中没有更多记录，您可以使用当前时间戳调整总的正常运行时间/停机时间。

窗口函数提供了一些其他有用的功能，例如带有 ROWS BETWEEN 和 RANGE BETWEEN 子句的窗口定义。

另一种可能的解决方案是将数据移动到 RDD 并使用 RangePartitioner、mapPartitions 和滑动窗口进行低级操作。对于基本的事情，您甚至可以 groupBy。这需要更多的努力，但也更加灵活。

最后有一个spark-timeseries来自 Cloudera 的软件包。文档几乎不存在，但测试足够全面，可以让您了解如何使用它。

关于自定义 UDAF，我不会乐观。 UDAF API 相当具体且不完全灵活。