如何在使用模式 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

如果您正在读取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使用schema和option作为读取文件

  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 的更多详细信息

希望这可以帮助！

Answer 2

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

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

• PERMISSIVE: 将其他字段设置为null当遇到损坏的记录时，并将格式错误的字符串放入由columnNameOfCorruptRecord. 当用户设置模式时，它会设置null额外的字段。

• DROPMALFORMED : 忽略整个损坏的记录。

• FAILFAST : 遇到损坏的记录时抛出异常。