Spark 读取 CSV - 不显示损坏记录答案

【问题标题】：Spark read CSV - Not showing corroupt RecordsSpark 读取 CSV - 不显示损坏记录
【发布时间】：2020-02-26 02:32:49
【问题描述】：

Spark 有一个Permissive 模式用于读取CSV 文件，该模式将损坏记录存储到名为_corroupt_record 的单独列中。

允许 - 遇到损坏的记录时将所有字段设置为空，并将所有损坏的记录放在字符串列中称为_corrupt_record

但是，当我尝试以下示例时，我没有看到任何名为 _corroupt_record 的列。与架构不匹配的记录似乎是null

data.csv

data
10.00
11.00
$12.00
$13
gaurang

代码

import org.apache.spark.sql.types.{StructField, StructType, StringType, LongType, DecimalType}
val schema = new StructType(Array(
new StructField("value", DecimalType(25,10), false)
))
val df = spark.read.format("csv") 
  .option("header", "true") 
  .option("mode", "PERMISSIVE") 
  .schema(schema) 
  .load("../test.csv")

架构

scala> df.printSchema()
root
 |-- value: decimal(25,10) (nullable = true)


scala> df.show()
+-------------+
|        value|
+-------------+
|10.0000000000|
|11.0000000000|
|         null|
|         null|
|         null|
+-------------+

如果我将模式更改为FAILFAST，我在尝试查看数据时会出错。

【问题讨论】：

您必须将 _corrupt_record 添加到您的架构中。
add("_corrupt_record", StringType, true) // schema 包含一个特殊的列 _corrupt_record，在数据中是不存在的。此列捕获未正确解析的行。见docs.azuredatabricks.net/_static/notebooks/…

标签： apache-spark apache-spark-sql databricks

【解决方案1】：

按照 Andrew 和 Prateek 的建议添加 _corroup_record 解决了该问题。

import org.apache.spark.sql.types.{StructField, StructType, StringType, LongType, DecimalType}
val schema = new StructType(Array(
  new StructField("value", DecimalType(25,10), true),
  new StructField("_corrupt_record", StringType, true)
))
val df = spark.read.format("csv") 
  .option("header", "true") 
  .option("mode", "PERMISSIVE") 
  .schema(schema) 
  .load("../test.csv")

查询数据

scala> df.show()
+-------------+---------------+
|        value|_corrupt_record|
+-------------+---------------+
|10.0000000000|           null|
|11.0000000000|           null|
|         null|         $12.00|
|         null|            $13|
|         null|        gaurang|
+-------------+---------------+

【讨论】：