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

Question

当我使用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”可以很容易地从我的数据集中过滤出来。谢谢！

Answer 1

<代码>。选项（“mode”，“DROPMALFORMED”）应该可以完成这项工作。

<代码>模式（默认允许）：允许在解析期间处理损坏记录的模式。

>

允许：当遇到损坏的记录时，将其他字段设置为null，并将格式错误的字符串放入由columnNameOfCorruptRecord配置的新字段中。当用户设置架构时，它会为额外字段设置null。

DROPMALFORMED：忽略整个损坏的记录。

FAILFAST：遇到损坏记录时引发异常。

Answer 2

如果您正在读取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)
))

使用模式和选项读取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的更多详细信息

希望这有帮助！