Spark与不寻常编码的CSV文件不一致

Question

上下文：

• 作为数据管道的一部分，我正在处理一些平面CSV文件，
• 这些文件具有不寻常的编码和逃避规则，
• 我的意图太预处理了，然后将其转换为parquets，以进行后续管道步骤

MCVE：

spark = SparkSession.builder.appName("...").getOrCreate()

min_schema = StructType(
    [
        StructField("dummy_col", StringType(), True),
        StructField("record_id", IntegerType(), nullable=False),
        StructField("dummy_after", StringType(), nullable=False),
    ]
)


df = (
    spark.read.option("mode", "FAILFAST")
    .option("quote", '"')
    .option("escape", '"')
    .option("inferSchema", "false")
    .option("multiline", "true")
    .option("ignoreLeadingWhiteSpace", "true")
    .option("ignoreTrailingWhiteSpace", "true")
    .schema(min_schema)
    .csv(f'min_repro.csv', header=True)
)

dummy_col,record_id,dummy_after
"",1,", Unusual value with comma included"
B,2,"Unusual value with escaped quote and comma ""like, this"

CSV PARSES：CSV PARSES FING：

df.collect()

[Row(dummy_col=None, record_id=1, dummy_after=', Unusual value with comma included'),
Row(dummy_col='B', record_id=2, dummy_after='Unusual value with escaped quote and comma "like, this')]

但是，在同一同一spard spark code上DF失败而失败的错误错误：

if df.count() != df.select('record_id').distinct().count():
    pass

Py4JJavaError: An error occurred while calling o357.count.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 17.0 failed 1 times, most recent failure: Lost task 0.0 in stage 17.0 (TID 13, localhost, executor driver): org.apache.spark.SparkException: Malformed records are detected in record parsing. Parse Mode: FAILFAST.
...
Caused by: java.lang.NumberFormatException: For input string: "Unusual value with comma included""
    at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)

我不明白.collect（）在同一df上可以提供正确的行，但是在同一的任何查询DF失败了。

创建了上游错误： https://issues.apache.apache.org/jira/jira/browse/browse/spark/spark/spark/spark/spark -39842

Answer 1

正确忽略的方式，数据中的数据是

1. 双引号中的数据。
2. 使用选项“逃生Quotes”，“ True”

df = ( spark.read.option("mode", "FAILFAST") .option("escapeQuotes", "true") .option("inferSchema", "false") .option("multiline", "true") .option("ignoreLeadingWhiteSpace", "true") .option("ignoreTrailingWhiteSpace", "true") .schema(min_schema) .csv(f'C:/Users/pc/Desktop/sample2.csv', header=True) )
------------------------------------------------------------------------

    >>> df.select('dummy_after').show(truncate=False)
    +-----------------------------------+
    |dummy_after                        |
    +-----------------------------------+
    |, Unusual value with comma included|
    +-----------------------------------+

    >>> if df.count() != df.select('record_id').distinct().count():
    ...    pass