Spark：解析数据帧同一列中不同格式的日期/时间戳（MM-dd-yyyy HH:mm、MM/dd/yy H:mm）

Question

问题是：我有一个数据集，其中一列具有两种或多种日期格式。一般来说，我选择所有值作为字符串类型，然后使用to_date来解析日期。但我不知道如何解析具有两种或多种日期格式的列。

val DF= Seq(("02-04-2020 08:02"),("03-04-2020 10:02"),("04-04-2020 09:00"),("04/13/19 9:12"),("04/14/19 2:13"),("04/15/19 10:14"), ("04/16/19 5:15")).toDF("DOB")

import org.apache.spark.sql.functions.{to_date, to_timestamp}
val DOBDF = DF.withColumn("Date", to_date($"DOB", "MM/dd/yyyy"))

上述命令的输出：

null
null
null
0019-04-13
0019-04-14
0019-04-15
0019-04-16

我编写的上面的代码不适用于该格式MM/dd/yyyy，并且未提供我将其null作为输出提供的格式。

因此寻求帮助来解析具有不同日期格式的文件。如果可能的话，还请分享一些处理日期格式的教程或注释。请注意：我使用 Scala 作为 Spark 框架。

提前致谢。

Answer 1

检查编辑部分以使用列函数而不是 UDF 来获得此解决方案后面部分的性能优势 -

好吧，让我们以 try-catch 的方式进行操作。尝试针对每种格式进行列转换并保留成功值。您可能必须从外部提供所有可能的格式作为参数，或者在代码本身的某个位置保留所有可能格式的主列表。

这是可能的解决方案..（我使用新库 - java.time.format.DateTimeFormatter，而不是有时在时间戳上出现问题的 SimpleDateFormatter）

创建一个 to_timestamp 函数，它接受字符串转换为时间戳和所有可能的格式

  import java.time.LocalDate
  import java.time.LocalDateTime
  import java.time.LocalTime
  import java.time.format.DateTimeFormatter
  import scala.util.Try

def toTimestamp(date: String, tsformats: Seq[String]): Option[java.sql.Timestamp] = {

    val out = (for (tsft <- tsformats) yield {
      val formatter = new DateTimeFormatterBuilder()
        .parseCaseInsensitive()
        .appendPattern(tsft).toFormatter()
      if (Try(java.sql.Timestamp.valueOf(LocalDateTime.parse(date, formatter))).isSuccess)
        Option(java.sql.Timestamp.valueOf(LocalDateTime.parse(date, formatter)))
      else None

    }).filter(_.isDefined)
    if (out.isEmpty) None else out.head
  }

在其之上创建一个 UDF -（此 udf 将格式字符串的 Seq 作为参数）

 def UtoTimestamp(tsformats: Seq[String]) = org.apache.spark.sql.functions.udf((date: String) => toTimestamp(date, tsformats))

现在，只需在您的 Spark 代码中使用它即可。这是对您的数据的测试 -

    val DF = Seq(("02-04-2020 08:02"), ("03-04-2020 10:02"), ("04-04-2020 09:00"), ("04/13/19 9:12"), ("04/14/19 2:13"), ("04/15/19 10:14"), ("04/16/19 5:15")).toDF("DOB")

    val tsformats = Seq("MM-dd-yyyy HH:mm", "MM/dd/yy H:mm")

    DF.select(UtoTimestamp(tsformats)('DOB)).show

这是输出 -

+-------------------+
|           UDF(DOB)|
+-------------------+
|2020-02-04 08:02:00|
|2020-03-04 10:02:00|
|2020-04-04 09:00:00|
|2019-04-13 09:12:00|
|2019-04-14 02:13:00|
|2019-04-15 10:14:00|
|2019-04-16 05:15:00|
+-------------------+

最重要的是避免为数据框中的许多列编写 UtoTimestamp(colname) 。让我们编写一个函数，它接受数据帧、所有时间戳列的列表以及源数据可能编码时间戳的所有可能格式。

它会为您解析所有时间戳列，并尝试针对格式。

def WithTimestampParsed(df: DataFrame, tsCols: Seq[String], tsformats: Seq[String]): DataFrame = {

    val colSelector = df.columns.map {
      c =>
        {
          if (tsCols.contains(c)) UtoTimestamp(tsformats)(col(c)) alias (c)
          else col(c)
        }
    }

像这样使用它 -

// You can pass as many column names in a sequence to be parsed
WithTimestampParsed(DF, Seq("DOB"), tsformats).show

输出 -

+-------------------+
|                DOB|
+-------------------+
|2020-02-04 08:02:00|
|2020-03-04 10:02:00|
|2020-04-04 09:00:00|
|2019-04-13 09:12:00|
|2019-04-14 02:13:00|
|2019-04-15 10:14:00|
|2019-04-16 05:15:00|
+-------------------+

编辑- 我看到了最新的 Spark 代码，他们现在也使用 java.time._ utils 来解析日期和时间戳，从而能够处理超过毫秒的数据。早期这些函数基于 SimpleDateFormat （我之前并不依赖于 Spark 的 to_timestamps由于此限制）。

因此，现在 to_date 和 to_timestamp 函数非常可靠。让我们使用它们，而不必编写 UDF。让我们编写一个对列进行操作的函数。

def to_timestamp_simple(col: org.apache.spark.sql.Column, formats: Seq[String]): org.apache.spark.sql.Column = {
    coalesce(formats.map(fmt => to_timestamp(col, fmt)): _*)
  }

有了这个 WithTimestampParsed 看起来像 -

def WithTimestampParsedSimple(df: DataFrame, tsCols: Seq[String], tsformats: Seq[String]): DataFrame = {

    val colSelector = df.columns.map {
      c =>
        {
          if (tsCols.contains(c)) to_timestamp_simple(col(c), tsformats) alias (c)
          else col(c)
        }
    }

    df.select(colSelector: _*)
  }

并像这样使用它 -

DF.select(to_timestamp_simple('DOB,tsformats)).show

//OR

WithTimestampParsedSimple(DF, Seq("DOB"), tsformats).show

输出看起来像 -

+---------------------------------------------------------------------------------------+
|coalesce(to_timestamp(`DOB`, 'MM-dd-yyyy HH:mm'), to_timestamp(`DOB`, 'MM/dd/yy H:mm'))|
+---------------------------------------------------------------------------------------+
|                                                                    2020-02-04 08:02:00|
|                                                                    2020-03-04 10:02:00|
|                                                                    2020-04-04 09:00:00|
|                                                                    2019-04-13 09:12:00|
|                                                                    2019-04-14 02:13:00|
|                                                                    2019-04-15 10:14:00|
|                                                                    2019-04-16 05:15:00|
+---------------------------------------------------------------------------------------+

+-------------------+
|                DOB|
+-------------------+
|2020-02-04 08:02:00|
|2020-03-04 10:02:00|
|2020-04-04 09:00:00|
|2019-04-13 09:12:00|
|2019-04-14 02:13:00|
|2019-04-15 10:14:00|
|2019-04-16 05:15:00|
+-------------------+