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

【问题标题】：Spark 2.0 Timestamp Difference in Milliseconds using Scala使用 Scala 的 Spark 2.0 时间戳差异（以毫秒为单位）
【发布时间】：2018-03-14 09:28:13
【问题描述】：

我正在使用 Spark 2.0 并正在寻找一种在 Scala 中实现以下目标的方法：

需要两个 Data-frame 列值之间的时间戳差异（以毫秒为单位）。

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

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

注意：在两个值上应用函数 unix_timestamp(Column s) 并减去有效，但不能达到要求的毫秒值。

最终查询如下所示：

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

这应该返回以下输出：

1023 毫秒

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

【问题讨论】：

标签： scala timestamp apache-spark-sql user-defined-functions apache-spark-2.0

【解决方案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")

【讨论】：

【解决方案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 通常很慢，因此您可能会遇到性能问题。

【讨论】：

【解决方案3】：

聚会晚了，但希望它仍然有用。

import org.apache.spark.sql.Column
def getUnixTimestamp(col: Column): Column = (col.cast("double") * 1000).cast("long")

df.withColumn("diff", getUnixTimestamp(col("col2")) - getUnixTimestamp(col("col1")))

当然你可以为区别定义一个单独的方法：

def timestampDiff(col1: Column, col2: Column): Column = getUnixTimestamp(col2) - getUnixTimestamp(col1)

df.withColumn("diff", timestampDiff(col("col1"), col("col2")))

为了让生活更轻松，可以为Strings 定义一个重载方法，默认名称为diff：

def timestampDiff(col1: String, col2: String): Column = timestampDiff(col(col1), col(col2)).as("diff")

现在在行动：

scala> df.show(false)
+-----------------------+-----------------------+
|min_time               |max_time               |
+-----------------------+-----------------------+
|1970-01-01 01:00:02.345|1970-01-01 01:00:04.786|
|1970-01-01 01:00:23.857|1970-01-01 01:00:23.999|
|1970-01-01 01:00:02.325|1970-01-01 01:01:07.688|
|1970-01-01 01:00:34.235|1970-01-01 01:00:34.444|
|1970-01-01 01:00:34.235|1970-01-01 01:00:34.454|
+-----------------------+-----------------------+


scala> df.withColumn("diff", timestampDiff("min_time", "max_time")).show(false)
+-----------------------+-----------------------+-----+
|min_time               |max_time               |diff |
+-----------------------+-----------------------+-----+
|1970-01-01 01:00:02.345|1970-01-01 01:00:04.786|2441 |
|1970-01-01 01:00:23.857|1970-01-01 01:00:23.999|142  |
|1970-01-01 01:00:02.325|1970-01-01 01:01:07.688|65363|
|1970-01-01 01:00:34.235|1970-01-01 01:00:34.444|209  |
|1970-01-01 01:00:34.235|1970-01-01 01:00:34.454|219  |
+-----------------------+-----------------------+-----+


scala> df.select(timestampDiff("min_time", "max_time")).show(false)
+-----+
|diff |
+-----+
|2441 |
|142  |
|65363|
|209  |
|219  |
+-----+

【讨论】：