Spark 行级错误处理，如何在行级获取错误消息答案

【问题标题】：Spark row level error handling , how to get error message at row levelSpark 行级错误处理，如何在行级获取错误消息
【发布时间】：2021-02-16 21:41:50
【问题描述】：

我有一个通过 spark 加载的 csv 文件。我想区分好记录和坏记录，也想知道坏记录的每一行级别的错误。

我正在指定一个模式并且可以像这样捕获损坏记录，但是如何获取每个不同损坏记录的错误消息？

  --------------+-----------+----------+--------------------+-------+--------------------+
|service_point_number|energy_type|is_enabled|            metadata|testint|     _corrupt_record|
+--------------------+-----------+----------+--------------------+-------+--------------------+
|            90453512|          E|     false|Address1@420#Addr...|     23|                null|
|            14802348|          G|     false|Address1@420#Addr...|     24|                null|
|                null|       null|      null|                null|   null|99944990,E,12,Add...|
|            78377144|          E|     false|                 123|     26|                null|
|            25506816|          G|     false|Address1@420#Addr...|     27|                null|
|            48789905|          E|      true|Address1@420#Addr...|   null|48789905,E,true,A...|
|            20283032|          E|     false|Address1@420#Addr...|     29|                null|
|            67311231|          G|     false|Address1@420#Addr...|     30|                null|
|            18240558|          G|     false|Address1@420#Addr...|     31|18240558,G,false,...|
|            42631153|          E|     false|Address1@420#Addr...|     32|                null|
+--------------------+-----------+----------+--------------------+-------+--------------------+

【问题讨论】：

标签： java apache-spark

【解决方案1】：

2.x 提供的 spark 中集成的 spark-csv 包 https://docs.databricks.com/spark/latest/spark-sql/handling-bad-records.html

badRecordsPath

在Possible to put records that aren't same length as header records to bad_record directory查看CSV 示例

【讨论】：

已经尝试过了，但是找不到任何在 HDFS 中写入不良记录的文件。我使用的是 spark 版本 2.3.1。我使用的是 spark 版本 2.3.1。 val df = spark.read.format("cs").option("header", "true").option("badRecordsPath", "/sampledata/test").schema(schema).csv("/sampledata/test.csv")
我尝试在 HDFS @georg 的 /sampledata/test 中找到
您是否尝试在 Databricks 或开源 spark 中运行它？可能是此功能仅在此处可用。但我对此不是 100% 确定。
我在 EMR 中使用 spark，但看不到 badRecordsPath 选项。

【解决方案2】：

badRecordsPath 可以工作，但在指定路径中找不到它的原因可能是在触发操作之前不会开始实际执行。尝试在您的代码spark.read... 之后使用df.show()，然后再次查看输出文件是否现在显示在路径中。

【讨论】：