如何用Spark分离字段数不符的行，生成正确数据集与错误文件

问题根源 #

你的代码存在两个关键问题：

1. 读取配置冲突：重复设置escape选项，且未正确配置来解析跨行带引号的字段；
2. 过滤逻辑错误：Spark读取CSV时，会自动为多字段行新增列、为少字段行补null，直接用len(row)无法判断原始行的实际字段数。

实现步骤 #

1. 读取原始行文本 #

先读取CSV的原始行内容，避免Spark自动解析字段时的列调整：

# 读取所有行的原始文本
raw_df = spark.read.text("input.csv")

# 获取表头并计算预期字段数
header = raw_df.first()[0].strip()
expected_field_count = len(header.split(","))  # 此处为3

2. 计算每行实际字段数 #

用Python标准库csv解析每行，确保带引号的跨行字段不被错误拆分：

import csv
from io import StringIO
from pyspark.sql.functions import udf, col, monotonically_increasing_id
from pyspark.sql.types import IntegerType

# 定义UDF：计算每行的实际字段数
def get_field_count(row_str):
    if not row_str.strip():
        return 0
    reader = csv.reader(StringIO(row_str.strip()), quotechar='"', escapechar='"')
    return len(next(reader))

count_fields_udf = udf(get_field_count, IntegerType())

# 给原始数据添加行号和字段数字段
raw_with_metadata = raw_df.withColumn("row_num", monotonically_increasing_id() + 1) \
                          .withColumn("actual_fields", count_fields_udf(col("value")))

3. 分离正确数据与错误数据 #

生成正确数据DataFrame

# 筛选字段数符合要求的行
correct_rows = raw_with_metadata.filter(col("actual_fields") == expected_field_count)

# 解析为结构化DataFrame
from pyspark.sql.types import StructType, StructField, StringType

schema = StructType([
    StructField("emp_id", StringType(), True),
    StructField("emp_name", StringType(), True),
    StructField("salary", StringType(), True)
])

correct_df = spark.read.csv(
    correct_rows.rdd.map(lambda x: x["value"]),
    schema=schema,
    header=True,
    multiline=True,
    quote='"',
    escape='"'
)

# 导出正确数据
correct_df.write.csv("Correct_input_data_frame", header=True, mode="overwrite")

生成错误数据CSV

from pyspark.sql.functions import concat, lit, concat_ws

# 筛选字段数不符的行并添加错误原因
wrong_rows = raw_with_metadata.filter(col("actual_fields") != expected_field_count) \
                             .withColumn(
                                 "reason",
                                 concat(
                                     lit(col("actual_fields")),
                                     lit(" fields in row "),
                                     lit(str(expected_field_count)),
                                     lit(" fields expected")
                                 )
                             )

# 拼接原始行与错误原因，生成符合要求的格式
wrong_final = wrong_rows.withColumn(
    "full_line",
    concat_ws(",", col("value"), col("reason"))
)

# 写入带表头的错误文件
wrong_header = f"{header},no_of_fields"

# 先写表头，再写入内容
with open("wrong_file.csv", "w") as f:
    f.write(wrong_header + "\n")
    for line in wrong_final.select("full_line").collect():
        f.write(line["full_line"] + "\n")

核心要点 #

• 用csv模块解析字段数，能准确识别跨行的带引号字段，避免误判；
• 保留原始行文本是关键，绕开Spark自动调整列的逻辑；
• 错误数据直接拼接原始行与原因，确保输出格式完全匹配需求。

内容的提问来源于stack exchange，提问作者K Soumya