apache-spark-sql - 如何在使用选项 ("dateFormat"读取 SPARK 中的 Dataframe 时处理 csv 文件中的不同日期格式)？

Question

我有以下输入文件

INPUTFILE_CNTRY_CODE|INPUTFILE_CTY_CODE|INPUTFILE_ID|INPUTFILE_LTY_ID|INPUTFILE_CNSM_ID|INPUTFILE_DATE|INPUTFILE_TIME|INPUTFILE_TRDATE
GBR|263|326735246||I034867789V|15/11/30|2015-11-30 00:00:00.000000|2016-22-06
GBR|263|397802068|PC7135361|PC7135361|16/05/20|2016-10-06 11:50:05.000000|2016-22-07

我正在尝试按以下方式阅读。

val registeration_schema = StructType(List(
      StructField("INPUTFILE_CNTRY_CODE", StringType),
      StructField("INPUTFILE_CTY_CODE", IntegerType),
      StructField("INPUTFILE_ID", IntegerType),
      StructField("INPUTFILE_LTY_ID", StringType),
      StructField("INPUTFILE_CNSM_ID", StringType),
      StructField("INPUTFILE_DATE", DateType),
      StructField("INPUTFILE_TIME", TimestampType),
      StructField("INPUTFILE_TRDATE", DateType)
    ))
    val registerationDF = spark.read
      .option("header", "true")
      .option("delimiter", "|")
      .option("mode", "FAILFAST")
      .schema(registeration_schema)
      .option("dateFormat", "yy/M/d")
      .option("timestampFormat", "yyyy-MM-dd HH:mm:ss.SSSSSS")
      .csv("registration2.csv")

而且我低于 Error 。

Caused by: org.apache.spark.sql.catalyst.util.BadRecordException: java.time.format.DateTimeParseException: Text '2016-22-06' could not be parsed at index 2
    at org.apache.spark.sql.catalyst.csv.UnivocityParser.org$apache$spark$sql$catalyst$csv$UnivocityParser$$convert(UnivocityParser.scala:262)
    at org.apache.spark.sql.catalyst.csv.UnivocityParser.$anonfun$doParse$2(UnivocityParser.scala:200)
    at org.apache.spark.sql.catalyst.csv.UnivocityParser.parse(UnivocityParser.scala:207)
    at org.apache.spark.sql.catalyst.csv.UnivocityParser$.$anonfun$parseIterator$1(UnivocityParser.scala:347)
    at org.apache.spark.sql.catalyst.util.FailureSafeParser.parse(FailureSafeParser.scala:60)
    ... 27 more
Caused by: java.time.format.DateTimeParseException: Text '2016-22-06' could not be parsed at index 2

发生这种情况是因为日期格式不同，我在加载 Dataframe 时只指定了一种日期格式。有人可以指导如何在将 CSV 读入数据帧时处理多种日期格式。

Answer 1

加载 CSV 时，您不能为 DateType 定义多种格式。但是您可以通过使用 spark 2.2+ 提供的 date_format() 和 to_date() 函数来实现这一点。

高级步骤如下-

1. 在原始模式中将这两列之一定义为 string。在我的演示中，我选择将 INPUTFILE_DATE 定义为 string。
2. 使用 date_format() 和 to_date() 函数将其数据类型转换为日期，以适当的格式格式化 INPUTFILE_DATE。

使用 INPUTFILE_DATE 定义原始模式 --> StringType

val registeration_schema = StructType(List(
      StructField("INPUTFILE_CNTRY_CODE", StringType),
      StructField("INPUTFILE_CTY_CODE", IntegerType),
      StructField("INPUTFILE_ID", IntegerType),
      StructField("INPUTFILE_LTY_ID", StringType),
      StructField("INPUTFILE_CNSM_ID", StringType),
      StructField("INPUTFILE_DATE", StringType),
      StructField("INPUTFILE_TIME", TimestampType),
      StructField("INPUTFILE_TRDATE", DateType)
    ))
    val registerationDF = spark.read
      .option("header", "true")
      .option("delimiter", "|")
      .option("mode", "FAILFAST")
      .schema(registeration_schema)
      .option("dateFormat", "yyyy-dd-MM")
      .option("timestampFormat", "yyyy-MM-dd HH:mm:ss.SSSSSS")
      .csv("registration2.csv")

这个解决方案的核心部分是——

val targetDF  = registerationDF.withColumn("INPUTFILE_DATE",to_date((date_format(to_date(col("INPUTFILE_DATE"),"yy/MM/dd"),"yyyy-dd-MM")),"yyyy-dd-MM"))

最终结果-

scala> targetDF.printSchema()
root
 |-- INPUTFILE_CNTRY_CODE: string (nullable = true)
 |-- INPUTFILE_CTY_CODE: integer (nullable = true)
 |-- INPUTFILE_ID: integer (nullable = true)
 |-- INPUTFILE_LTY_ID: string (nullable = true)
 |-- INPUTFILE_CNSM_ID: string (nullable = true)
 |-- INPUTFILE_DATE: date (nullable = true)
 |-- INPUTFILE_TIME: timestamp (nullable = true)
 |-- INPUTFILE_TRDATE: date (nullable = true)


scala> targetDF.show()

+--------------------+------------------+------------+----------------+-----------------+--------------+-------------------+----------------+
|INPUTFILE_CNTRY_CODE|INPUTFILE_CTY_CODE|INPUTFILE_ID|INPUTFILE_LTY_ID|INPUTFILE_CNSM_ID|INPUTFILE_DATE|     INPUTFILE_TIME|INPUTFILE_TRDATE|
+--------------------+------------------+------------+----------------+-----------------+--------------+-------------------+----------------+
|                 GBR|               263|   326735246|            null|      I034867789V|    2015-11-30|2015-11-30 00:00:00|      2017-10-06|
|                 GBR|               263|   397802068|       PC7135361|        PC7135361|    2016-05-20|2016-10-06 11:50:05|      2017-10-07|
+--------------------+------------------+------------+----------------+-----------------+--------------+-------------------+----------------+