scala - Spark : Parse a Date/Timestamps with different Formats (MM-dd-yyyy HH:mm, MM/dd/yy H:mm ) 在 Dataframe 的同一列

Question

问题是:我有一个数据集，其中一列具有 2 种或更多类型的日期格式。通常我选择所有值作为 String 类型，然后使用 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

检查 EDIT 部分以使用 Column 函数而不是 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|
+-------------------+

Cherry on top 将避免为数据框中的许多列编写 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|
+-------------------+