spark 解析的时间戳 csv 不正确

Question

我有一个格式如下的 CSV 文件：

574,REF009,3213,16384,3258,111,512,2013-12-07 21:03:12.567+01,2013-12-07 21:03:12.567+01,2013-12-31 23:33:15.821+01,/data/ath/athdisk/ro/user/bas/b6/c0
48,REF010,456,32768,3258,111,2175850,2018-07-10 04:37:06.495+02,2018-07-10 04:37:06.459+02,2018-07-10 04:37:06.648+02,/data/ath/athdisk/ro/mc15/b9/dc/lo.log.tgz.1
1758,REF011,123,32768,3258,111,31691926,2017-04-21 22:29:30.315+02,2017-10-20 05:55:03.959+02,2017-04-21 22:29:31+02,/data/ath/athdisk/ro/dataV/1f/00/D0293.pool.root

在尝试导入如此庞大（11B 行长）的文件时，我有大约 400 万行充满了空值。我意识到我的文件有问题，所以我尝试运行使用选项 FAILFAST 导入，如下所示：

val inodes_schema = StructType(
    Array(
        StructField("testID",LongType,false),
        StructField("ref",StringType,false),
        StructField("iref",IntegerType, false),
        StructField("flag",IntegerType, false),
        StructField("iuid",IntegerType, false),
        StructField("igid",IntegerType, false),
        StructField("isize",LongType, false),
        StructField("icrtime",TimestampType,false),
        StructField("iatime",TimestampType,false),
        StructField("ictime",TimestampType,false),
        StructField("path",StringType,false)
    )
)

val inodes_table = spark.read.option("mode", "FAILFAST")
.option("timestampFormat", "yyyy-MM-dd HH:mm:ss.SSSX")
.schema(inodes_schema)
.option("delimiter",",")
.option("header",false).csv("/my/csv/file.csv")

这让我可以确定包含 59+02 的行导致了问题。有很多包含 59+02 的行，如果我使用通常的 PERMISSIVE 模式，我最终设法识别出一个未正确导入的行：

 1758,REF011,123,32768,3258,111,31691926,2017-04-21 22:29:30.315+02,2017-10-20 05:55:03.959+02,2017-04-21 22:29:31+02,/data/ath/athdisk/ro/dataV/1f/00/D0293.pool.root

我不明白为什么 Spark 没有正确解析这一行？ 05:55:03.959+02 关于我的时间戳，小时格式是正确的，但行将无法正确导入，而且可能有很多行。

Answer 1

上述问题似乎是由您的数据中包含多种时间戳格式引起的。一种解决方法是将 TimestampType 列设置为 StringType 以读取 CSV，然后将它们转换回 TimestampType：

// /path/to/csvfile:
1,2017-04-21 22:29:30.315+02
2,2017-10-20 05:55:03.959+02
3,2017-04-21 22:29:31+02

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

val schema = StructType(Array(
  StructField("id", IntegerType, false),
  StructField("dt", StringType, false)
))

val df = spark.read.
  option("mode", "FAILFAST").
  option("delimiter", ",").
  option("header", false).
  schema(schema).
  csv("/path/to/csvfile")

df.select($"id", $"dt".cast(TimestampType)as("dt")).
  show(false)
// +---+-----------------------+
// |id |dt                     |
// +---+-----------------------+
// |1  |2017-04-21 13:29:30.315|
// |2  |2017-10-19 20:55:03.959|
// |3  |2017-04-21 13:29:31    |
// +---+-----------------------+

spark 解析的时间戳 csv 不正确

incorrect timestamp csv parsing by spark

scala

dataframe

apache-spark

apache-zeppelin