更新火花数据框中多个火花列的日期格式

Question

我有一个 Spark 数据框，其中几列具有不同类型的日期格式。
为了解决这个问题，我编写了下面的代码，以保持所有日期列的格式类型一致。
由于日期列的日期格式每次都可能发生变化，因此我在dt_formats 中定义了一组日期格式。

def to_timestamp_multiple(s: Column, formats: Seq[String]): Column = {
    coalesce(formats.map(fmt => to_timestamp(s, fmt)):_*)
}

val dt_formats= Seq("dd-MMM-yyyy", "MMM-dd-yyyy", "yyyy-MM-dd","MM/dd/yy","dd-MM-yy","dd-MM-yyyy","yyyy/MM/dd","dd/MM/yyyy")

val newDF =  df.withColumn("ETD1", date_format(to_timestamp_multiple($"ETD",Seq("dd-MMM-yyyy", dt_formats)).cast("date"), "yyyy-MM-dd")).drop("ETD").withColumnRenamed("ETD1","ETD")

但是在这里我必须创建一个新列然后我必须删除旧列然后重命名新列。 这使代码变得不必要非常笨拙，因此我想从此代码中覆盖。

我正在尝试通过在函数下面编写一个 Scala 来实现类似的功能，但它抛出异常 org.apache.spark.sql.catalyst.parser.ParseException:，但我无法确定我应该进行哪些更改才能使其工作..

val CleansedData= rawDF.selectExpr(rawDF.columns.map( 
x => { x match {
  case "ETA" => s"""date_format(to_timestamp_multiple($x, dt_formats).cast("date"), "yyyy-MM-dd") as ETA"""
  case _ => x
}  }   ) : _*)

因此寻求帮助。 提前致谢。

Answer 1

创建一个UDF 以便与select 一起使用。 The select method 获取列并生成另一个 DataFrame。

此外，与其使用coalesce，不如简单地构建一个处理所有格式的解析器可能更直接。您可以为此使用DateTimeFormatterBuilder。

import java.time.format.DateTimeFormatter
import java.time.format.DateTimeFormatterBuilder
import org.apache.spark.sql.functions.udf
import java.time.LocalDate
import scala.util.Try
import java.sql.Date

val dtFormatStrings:Seq[String] = Seq("dd-MMM-yyyy", "MMM-dd-yyyy", "yyyy-MM-dd","MM/dd/yy","dd-MM-yy","dd-MM-yyyy","yyyy/MM/dd","dd/MM/yyyy")

// use foldLeft with appendOptional method, which for each format,
// returns a new builder with that additional possible format

val initBuilder = new DateTimeFormatterBuilder()
val builder: DateTimeFormatterBuilder = dtFormatStrings.foldLeft(initBuilder)(
  (b: DateTimeFormatterBuilder, s:String) => b.appendOptional(DateTimeFormatter.ofPattern(s)))
val formatter = builder.toFormatter()

// Create the UDF, which just takes
// any function returning a sql-compatible type (java.sql.Date, here)

def toTimeStamp2(dateString:String): Date = {
  val dateTry: Try[Date] = Try(java.sql.Date.valueOf(LocalDate.parse(dateString, formatter)))
  dateTry.toOption.getOrElse(null)
}

val timeConversionUdf = udf(toTimeStamp2 _)

// example DF and new DF
val df = Seq(("05/08/20"), ("2020-04-03"), ("unparseable")).toDF("ETD")
df.select(timeConversionUdf(col("ETD"))).toDF("ETD2").show

输出：

+----------+
|      ETD2|
+----------+
|2020-05-08|
|2020-04-03|
|      null|
+----------+

请注意，不可解析的值最终为 null，如图所示。

Answer 2

尝试withColumn(...) 同名并合并如下-

val dt_formats= Seq("dd-MMM-yyyy", "MMM-dd-yyyy", "yyyy-MM-dd","MM/dd/yy","dd-MM-yy","dd-MM-yyyy","yyyy/MM/dd","dd/MM/yyyy")

val newDF =  df.withColumn("ETD", coalesce(dt_formats.map(fmt => to_date($"ETD", fmt)):_*))