如何在 Spark 中使用日期和时间值对列进行排序

Question

注意：我把它作为火花中的数据帧。此时间/日期值构成数据帧中的单个列。

输入：

04-11月-16 03.36.13.000000000 PM
06-十一月-15 03.42.21.000000000 下午
05-十一月-15 03.32.05.000000000 下午
06-十一月-15 03.32.14.000000000 AM

预期输出：

05-NOV-15 03.32.05.000000000 PM
06-NOV-15 03.32.14.000000000 AM
06-NOV-15 03.42.21.000000000 PM
04-NOV-16 03.36.13.000000000 PM

Answer 1

由于这种格式不是标准的，你需要使用 unix_timestamp 函数来解析字符串并转换为时间戳类型：

import org.apache.spark.sql.functions._
// Example data
val df = Seq(
  Tuple1("04-NOV-16 03.36.13.000000000 PM"),
  Tuple1("06-NOV-15 03.42.21.000000000 PM"),
  Tuple1("05-NOV-15 03.32.05.000000000 PM"),
  Tuple1("06-NOV-15 03.32.14.000000000 AM")
).toDF("stringCol")
// Timestamp pattern found in string
val pattern = "dd-MMM-yy hh.mm.ss.S a"
// Creating new DataFrame and ordering
val newDF = df
  .withColumn("timestampCol", unix_timestamp(df("stringCol"), pattern).cast("timestamp"))
  .orderBy("timestampCol")
newDF.show(false)

结果：

+-------------------------------+---------------------+
|stringCol                      |timestampCol         |
+-------------------------------+---------------------+
|05-NOV-15 03.32.05.000000000 PM|2015-11-05 15:32:05.0|
|06-NOV-15 03.32.14.000000000 AM|2015-11-06 03:32:14.0|
|06-NOV-15 03.42.21.000000000 PM|2015-11-06 15:42:21.0|
|04-NOV-16 03.36.13.000000000 PM|2016-11-04 15:36:13.0|
+-------------------------------+---------------------+

有关unix_timestamp和其他实用程序功能的更多信息，请参阅此处。

要构建时间戳格式，可以参考 简单日期格式化程序文档

---

编辑 1：正如 pheeleeppoo 所说，您可以直接按表达式排序，而不是创建新列，假设您只想在数据帧中保留字符串类型的列：

val newDF = df.orderBy(unix_timestamp(df("stringCol"), pattern).cast("timestamp"))

---

编辑 2：请注意，unix_timestamp函数的精度以秒为单位，因此如果毫秒真的很重要，可以使用 udf：

def myUDF(p: String) = udf(
  (value: String) => {
    val dateFormat = new SimpleDateFormat(p)
    val parsedDate = dateFormat.parse(value)
    new java.sql.Timestamp(parsedDate.getTime())
  }
)
val pattern = "dd-MMM-yy hh.mm.ss.S a"
val newDF = df.withColumn("timestampCol", myUDF(pattern)(df("stringCol"))).orderBy("timestampCol")

Answer 2

您还可以在将字符串转换为时间戳后使用排序函数：

   df.sort(unix_timestamp(df("dateColumn"), "dd-MMM-yy hh.mm.ss.S a").cast("timestamp"))
     .show(false)