Spark 2.0时间戳使用Scala以毫秒为单位的差异

Question

我正在使用Spark 2.0并寻找在Scala中实现以下功能的方法：

需要两个数据框列值之间的时间戳差异（以毫秒为单位）。

Value_1 = 06/13/2017 16:44:20.044
Value_2 = 06/13/2017 16:44:21.067

两者的数据类型都是时间戳。

注意：对两个值应用函数unix_timestamp（列s）并减去工作但不要达到要求的毫秒值。

最终查询如下所示：

Select **timestamp_diff**(Value_2,Value_1) from table1

这应该返回以下输出：

1023毫秒

其中timestamp_diff是计算差异的函数，以毫秒为单位。

Answer 1

一种方法是使用Unix纪元时间，即自1970年1月1日以来的毫秒数。下面是使用UDF的示例，它需要两个时间戳并以毫秒为单位返回它们之间的差异。

val timestamp_diff = udf((startTime: Timestamp, endTime: Timestamp) => {
  (startTime.getTime() - endTime.getTime())
})

val df = // dataframe with two timestamp columns (col1 and col2)
  .withColumn("diff", timestamp_diff(col("col2"), col("col1")))

或者，您可以注册要与SQL命令一起使用的函数：

val timestamp_diff = (startTime: Timestamp, endTime: Timestamp) => {
  (startTime.getTime() - endTime.getTime())
}

spark.sqlContext.udf.register("timestamp_diff", timestamp_diff)
df.createOrReplaceTempView("table1")

val df2 = spark.sqlContext.sql("SELECT *, timestamp_diff(col2, col1) as diff from table1")

Answer 2

PySpark也是如此：

import datetime

def timestamp_diff(time1: datetime.datetime, time2: datetime.datetime):
    return int((time1-time2).total_seconds()*1000)

int和*1000只输出毫秒

用法示例：

spark.udf.register("timestamp_diff", timestamp_diff)    

df.registerTempTable("table1")

df2 = spark.sql("SELECT *, timestamp_diff(col2, col1) as diff from table1")

它不是最佳解决方案，因为UDF通常很慢，因此您可能会遇到性能问题。