如何在使用模式 Spark 读取 csv 时删除格式错误的行？答案

【问题标题】：How to drop malformed rows while reading csv with schema Spark?如何在使用模式 Spark 读取 csv 时删除格式错误的行？
【发布时间】：2018-09-18 09:01:31
【问题描述】：

当我使用 Spark DataSet 加载 csv 文件时。我更喜欢清楚地指定模式。但我发现有几行不符合我的架构。一列应该是双倍的，但有些行是非数字值。是否可以轻松地从 DataSet 中过滤所有不符合我的模式的行？

val schema = StructType(StructField("col", DataTypes.DoubleType) :: Nil)
val ds = spark.read.format("csv").option("delimiter", "\t").schema(schema).load("f.csv")

f.csv：

a
1.0

我更喜欢“a”可以很容易地从我的 DataSet 中过滤出来。谢谢！

【问题讨论】：

标签： scala csv apache-spark apache-spark-dataset

【解决方案1】：

如果您正在读取 CSV 文件并希望删除与架构不匹配的行。您可以通过将选项 mode 添加为 DROPMALFORMED 来做到这一点

输入数据

a,1.0
b,2.2
c,xyz
d,4.5
e,asfsdfsdf
f,3.1

架构

val schema = StructType(Seq(
  StructField("key", StringType, false),
  StructField("value", DoubleType, false)
))

使用schema 和option 读取csv 文件

  val df = spark.read.schema(schema)
    .option("mode", "DROPMALFORMED")
    .csv("/path to csv file ")

输出：

+-----+-----+
|key  |value|
+-----+-----+
|hello|1.0  |
|hi   |2.2  |
|how  |3.1  |
|you  |4.5  |
+-----+-----+

您可以在此处获取有关spark-csv 的更多详细信息

希望这会有所帮助！

【讨论】：

谢谢！你的两个答案都是正确的。所以我将第一个答案设置为接受的答案。但感谢你们两位的宝贵回答！
如果我希望将损坏的记录用于审计目的，我该如何恢复它们。例如，我想创建一个带有损坏记录的新 df。
对了，格式错误的记录应该反馈给源系统/厂商，超级基础的功能，Spark有这个功能吗？
需要注意的是，您会默默地从输入数据中删除数据

【解决方案2】：

.option("mode", "DROPMALFORMED") 应该做这项工作。

mode（默认PERMISSIVE）：允许在解析过程中处理损坏记录的模式。

PERMISSIVE：当遇到损坏的记录时，将其他字段设置为null，并将格式错误的字符串放入columnNameOfCorruptRecord配置的新字段中。什么时候架构由用户设置，它为额外字段设置null。
DROPMALFORMED ：忽略整个损坏的记录。
FAILFAST ：遇到损坏的记录时抛出异常。

【讨论】：

这不是和其他答案一样吗？
如果 CSV 加载器的结果实际上是 2 个单独的数据帧对象，一个已解析，第二个已损坏，会更好。