Spark SQL数据帧:跨行对计算的最佳方式

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")  

然而，很少有情况会导致错误，例如设备停机但从未恢复，在这种情况下，停机时间是当前时间和上次停机时间之间的差值。

此外，如果特定设备的第一个条目以"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|
// +---+-------------------+-----+----------+------+--------+

以类似的方式，您可以向前看，如果组中没有更多记录，则可以使用当前时间戳调整总uptime/downtime。

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

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

最后是来自Cloudera的spark-timeseries包。文档几乎不存在，但测试足够全面，可以让你知道如何使用它

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