PySpark写入HDFS时含双引号字段引发额外分隔符问题

这种列数突然变多的情况，几乎可以肯定是双引号未正确处理引发的CSV解析/写入规则冲突，结合你的场景（GCS读、Dataproc写HDFS、分隔符为|），我来拆解问题根源和解决办法：

问题根源 #

PySpark默认的CSV读写逻辑会把双引号(")当作字段包裹符——也就是说，它认为被双引号包裹的内容是一个完整字段，哪怕里面包含分隔符|。如果你的源数据里某列的双引号没有被转义（比如字段内容是abc"def|ghi而不是abc""def|ghi），Spark就会错误地把这个引号当成字段的起始标记，直到找到下一个引号才结束字段。这会导致后续的|被当成字段内容的一部分，或者错误地拆分出额外的列，最终出现51列变59列的异常。

解决办法 #

1. 读取GCS数据时统一引号处理规则 #

根据你的数据实际情况选择以下两种方案：

方案A：完全禁用引号解析（推荐如果双引号只是普通字符）

如果你的双引号只是字段内容的一部分，不需要作为字段包裹符，直接关闭Spark的引号解析逻辑：

from pyspark.sql import SparkSession

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

# 读取GCS数据时禁用引号处理
df = spark.read.csv(
    "gs://your-bucket/path/to/source-data",
    sep="|",
    quote=None,  # 把双引号当作普通字符，不做包裹符解析
    escape=None,
    header=False,  # 根据你的数据是否有表头调整为True/False
    inferSchema=False  # 建议手动定义Schema，避免推断错误
)

方案B：正确配置转义字符（如果双引号需要被转义）

如果你的源数据里的双引号是用双引号转义的（比如"abc""def"表示实际内容是abc"def），需要指定转义字符为双引号：

df = spark.read.csv(
    "gs://your-bucket/path/to/source-data",
    sep="|",
    quote="\"",  # 保留双引号作为包裹符
    escape="\"",  # 用双引号转义双引号
    header=False,
    inferSchema=False
)

2. 写入HDFS时保持一致的规则 #

写入时必须和读取时的引号规则保持一致，避免二次解析错误：

对应方案A的写入代码

df.write.csv(
    "hdfs://your-nn-host/path/to/output",
    sep="|",
    quote=None,  # 同样禁用引号包裹
    escape=None,
    header=False,
    mode="overwrite"  # 根据需求选择append/overwrite等
)

对应方案B的写入代码

df.write.csv(
    "hdfs://your-nn-host/path/to/output",
    sep="|",
    quote="\"",
    escape="\"",
    header=False,
    mode="overwrite"
)

3. 手动定义Schema（关键步骤） #

因为Spark自动推断Schema时，遇到列数不一致的行会出现错误或数据丢失，建议提前定义好51列的Schema：

from pyspark.sql.types import StructType, StructField, StringType

# 构建51列的Schema，这里用StringType适配所有类型，你可以根据实际调整
schema = StructType([
    StructField(f"column_{idx}", StringType(), nullable=True) for idx in range(51)
])

# 读取时指定Schema
df = spark.read.csv(
    "gs://your-bucket/path/to/source-data",
    sep="|",
    quote=None,
    schema=schema,
    header=False
)

4. 排查验证技巧 #

可以先采样有问题的行，查看原始内容确认引号格式：

# 筛选包含双引号的行，完整显示内容
df.filter(df["column_xxx"].contains('"')).show(truncate=False)

这里的column_xxx替换为你怀疑有问题的列名，通过查看实际内容能更精准地调整参数。

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