如何从Scala Spark DataFrameReader csv记录格式错误的行

Question

Scala\u Spark\u DataFrameReader\u csv的文档表明，Spark可以记录在读取时检测到的格式错误的行。csv文件
-如何记录格式错误的行
-是否可以获取包含格式错误行的val或var？

链接文档中的选项是：maxMalformedLogPerPartition（默认值10）：设置Spark将为每个分区记录的最大错误行数。超过此数字的格式错误记录将被忽略

Answer 1

我在这里通过加载csv、从未损坏的记录中创建模式、添加损坏的记录列、使用新模式加载csv，然后查找损坏的记录来扩展clucar的答案。

from pyspark.sql.types import StructField, StringType
from pyspark.sql.functions import col

file_path = "/path/to/file"
mode = "PERMISSIVE"

schema = spark.read.options(mode=mode).csv(file_path).schema
schema = schema.add(StructField("_corrupt_record", StringType(), True))

df = spark.read.options(mode=mode).schema(schema).csv(file_path)
df.cache()
df.count()
df.filter(col("_corrupt_record").isNotNull()).show()

Answer 2

如果你正在使用火花2.3检查_corrupt_error特殊列...根据几个火花讨论“它应该工作”，所以在读取过滤器后，那些非空的科勒-应该有你的错误...你也可以检查input_file_name（）sql func

如果未使用低于2.3版的版本，则应实现自定义读取、记录解决方案，因为根据我的测试，csv数据源不存在\u corrupt\u错误。。。

Answer 3

基于此databricks示例，在读取文件时，需要将“\u corrupt\u record”列显式添加到架构定义中。在pyspark 2.4.4中，类似的内容对我很有用：

from pyspark.sql.types import *

my_schema = StructType([
  StructField("field1", StringType(), True),
  ...
  StructField("_corrupt_record", StringType(), True)
])

my_data = spark.read.format("csv")\
  .option("path", "/path/to/file.csv")\
  .schema(my_schema)
  .load()

my_data.count()  # force reading the csv

corrupt_lines = my_data.filter("_corrupt_record is not NULL")
corrupt_lines.take(5)