Spark：使用 PERMISSIVE 读取文件并提供模式 - 记录列损坏的问题答案

【问题标题】：Spark: reading files with PERMISSIVE and provided schema - issues with corrupted records columnSpark：使用 PERMISSIVE 读取文件并提供模式 - 记录列损坏的问题
【发布时间】：2019-01-21 16:45:35
【问题描述】：

我正在阅读 spark CSV。我正在为我阅读的文件提供一个模式，并且我在许可模式下阅读它。我想将所有记录保留在 columnNameOfCorruptRecord 中（在我的情况下是损坏的记录）。

我经历了地狱般的设置，但仍然收到警告说我无法抑制我想念的东西。

因此，首先为了获得损坏的记录列，我需要将其作为 StringType 添加到架构中。这记录得很好。但是，每当我读取文件时，都会收到一个警告，即架构不匹配，因为列的数量不同。这只是一个警告，但它正在填满我的日志。

此外，当有一个不可为空的字段并且有一条损坏的记录时，损坏的记录将进入损坏的记录列，并且它的所有字段都设置为空，因此我得到一个异常，因为我有一个不可为空的字段。解决此问题的唯一方法是将列不可为空设置为可空。这很奇怪。

我错过了什么吗？

回顾：

有没有办法在我添加时忽略警告架构中的损坏记录列
有没有办法使用具有模式的 PERMISSIVE 模式和损坏的记录列不可为空的字段。

谢谢！

【问题讨论】：

标签： apache-spark apache-spark-sql

【解决方案1】：

以下文档可能会有所帮助。如果您至少提供您编写的代码，那就太好了。 https://docs.databricks.com/spark/latest/data-sources/read-csv.html

【讨论】：

【解决方案2】：

读取json代码sn-p的demo

df= self.spark.read.option("mode", "PERMISSIVE").option("primitivesAsString", True).json(self.src_path)

【讨论】：

【解决方案3】：

要回答你的第 2 点，你应该更好地研究第 1 点。

第 1 点：您应该对文件进行分析，并将您的架构与文件中的所有字段进行映射。将您的 csv 文件导入 DataFrame 后，我会选择您感兴趣的领域，然后继续您正在做的事情。

第 2 点：您将解决定义架构的问题，如下所示（我会使用 scala）：

    import pyspark.sql.types as types
    yourSchema = (types.StructType()
                  .add('field0', types.IntegerType(), True)
                     # all your .add(fieldsName, fieldsType, True which let your field be nullable)
                  .add('corrupted_records', types.StringType(), True) #your corrupted date will be here
                  )

定义好后，您可以将您的 csv 文件 导入 DataFrame，如下所示：

    df = ( spark.read.format("csv")
             .schema(yourSchema)
             .option("mode", "PERMISSIVE")
             .option("columnNameOfCorruptRecord", "corrupted_records")
             load(your_csv_files)
         )

同样的操作还有其他的方法，处理bad bad的方式也不同；看看这篇有见地的文章：https://python.plainenglish.io/how-to-handle-bad-data-in-spark-sql-5e0276d37ca1

【讨论】：