PySpark读取CSV时字段含分隔符的问题解决求助

在Databricks中使用PySpark批量读取1000+个以|为分隔符的CSV文件，部分文件单文件超10GB，无法直接修改原文件。原读取代码如下：

df = spark.read.options(delimiter='|', header=True, multiLine=True).schema(schema).csv('/mnt/file.csv')

遇到的问题：部分文件的字段内容包含|（无引号包裹），导致DataFrame结构异常，甚至Timestamp字段被非法字符串填充，无法保存。尝试过以下代码但无效：

df =spark.read.options(delimiter=',',header=True, multiLine=True, inferSchema=True).csv('/mnt/file.csv')
df.withColumn('header|in|one|row', regexp_replace('header|in|one|row', 'text||||text2', 'text text2'))

1. 先定位问题文件与异常行 #

先批量扫描所有文件，找出字段数不符合预期的行及对应文件：

from pyspark.sql.functions import split, size, col, input_file_name

# 读取所有文件为纯文本行，保留文件名
raw_lines = spark.read.text('/mnt/*.csv') \
    .withColumn("filename", input_file_name())

# 从预定义schema获取预期字段数
expected_cols = len(schema.fields)

# 筛选出字段数不符的行（排除header行）
problem_rows = raw_lines \
    .withColumn("split_count", size(split(col("value"), "\\|"))) \
    .filter((col("split_count") != expected_cols) & (col("value") != ",".join([f.name for f in schema.fields])))

# 查看问题详情
problem_rows.select("filename", "value").show(truncate=False)

2. 修复异常行字段数 #

针对字段内包含|导致的字段数超标问题，将多余的|合并到最后一个字段，保证每行字段数与schema一致：

from pyspark.sql.functions import split, concat_ws, element_at, array, slice

# 读取文本行并保留文件名
raw_data = spark.read.text('/mnt/*.csv') \
    .withColumn("filename", input_file_name())

# 拆分后重构字段：前N-1个字段正常取，剩余部分合并为最后一个字段
fixed_data = raw_data \
    .withColumn("split_values", split(col("value"), "\\|")) \
    .withColumn("fixed_values", 
                array(*[element_at(col("split_values"), i) for i in range(1, expected_cols)] + 
                      [concat_ws("|", slice(col("split_values"), expected_cols, size(col("split_values"))))]
               )) \
    .withColumn("fixed_line", concat_ws("|", col("fixed_values")))

# 从修复后的行加载为符合schema的DataFrame
fixed_rdd = fixed_data.select("fixed_line").rdd.map(lambda x: x[0])
df_fixed = spark.read.options(delimiter='|', header=True, multiLine=True).schema(schema).csv(fixed_rdd)

3. 自定义解析逻辑（灵活适配特殊场景） #

如果特定字段才会出现|，可以写自定义UDF针对性解析：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, ArrayType

def parse_line(line, expected_cols):
    parts = line.split("|")
    if len(parts) == expected_cols:
        return parts
    # 示例：仅最后一个字段允许包含|，合并剩余部分
    elif len(parts) > expected_cols:
        return parts[:expected_cols-1] + ["|".join(parts[expected_cols-1:])]
    # 字段数不足时补空值
    else:
        return parts + [""]*(expected_cols - len(parts))

parse_udf = udf(lambda x: parse_line(x, expected_cols), ArrayType(StringType()))

# 解析并转换为对应schema类型
parsed_df = raw_data \
    .filter(col("value") != ",".join([f.name for f in schema.fields]))  # 排除header行
    .withColumn("parsed", parse_udf(col("value"))) \
    .select(*[col("parsed")[i].alias(field.name) for i, field in enumerate(schema.fields)])

# 转换字段类型匹配schema
for field in schema.fields:
    parsed_df = parsed_df.withColumn(field.name, col(field.name).cast(field.dataType))

4. 修复Timestamp字段异常 #

如果是Timestamp字段无法保存，先定位并处理非法值：

from pyspark.sql.functions import to_timestamp, when, col

# 尝试转换Timestamp字段，无效值设为None（或替换为默认值）
df = df.withColumn("valid_timestamp", 
                   when(to_timestamp(col("timestamp_col"), "yyyy-MM-dd HH:mm:ss").isNotNull(), 
                        to_timestamp(col("timestamp_col"), "yyyy-MM-dd HH:mm:ss"))
                   .otherwise(None))

# 过滤无效行（可选）
df_clean = df.filter(col("valid_timestamp").isNotNull())

# 查看无效Timestamp对应的文件和内容
invalid_timestamps = df.filter(col("valid_timestamp").isNull())
invalid_timestamps.select("filename", "timestamp_col").show(truncate=False)

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