PySpark读取含双引号内分隔符的文本文件解决方案咨询

你的问题核心在于PySpark默认CSV读取器的quote参数是用来包裹单个字段的完整内容，但你的原始数据中引号跨了多个字段（role、exp、task_desc），导致默认解析逻辑无法正确拆分字段。以下两种方法可以解决这个问题：

方法一：RDD手动行处理 #

先按行读取文本，再手动拆分字段，精准处理引号内的分隔符：

# 1. 读取文本文件，指定编码ISO-8859-1
text_rdd = spark.sparkContext.textFile("test.txt", use_unicode=False).map(lambda x: x.decode('ISO-8859-1'))

# 2. 提取表头和数据行
header = text_rdd.first().split('þ')
data_rdd = text_rdd.filter(lambda line: line != text_rdd.first())

# 3. 定义行处理函数
def parse_line(line):
    start_quote = line.find('"')
    end_quote = line.rfind('"')
    
    # 拆分引号外、引号内、引号后的部分
    left_section = [p.strip() for p in line[:start_quote].split('þ') if p.strip()]
    quote_content = line[start_quote+1:end_quote].split('þ')
    right_section = [p.strip() for p in line[end_quote+1:].split('þ') if p.strip()]
    
    # 拼接成目标字段格式
    return (
        left_section[0],
        left_section[1],
        f'"{quote_content[0].strip()}',
        quote_content[1].strip(),
        f'{quote_content[2].strip()}"',
        right_section[0]
    )

# 4. 转换为DataFrame
from pyspark.sql import Row
result_df = data_rdd.map(parse_line).map(lambda x: Row(
    id=x[0], name=x[1], role=x[2], exp=x[3], task_desc=x[4], comp=x[5]
)).toDF()

# 查看结果
result_df.show(truncate=False)

方法二：正则表达式提取字段 #

利用Spark的正则表达式函数直接匹配并提取每个字段，代码更简洁：

from pyspark.sql import functions as F

# 1. 读取文本文件
df = spark.read.text("test.txt", encoding='ISO-8859-1')

# 2. 提取表头并过滤表头行
header_str = df.first()[0]
df = df.filter(df.value != header_str)

# 3. 用正则表达式提取每个字段
result_df = df \
    .withColumn("id", F.regexp_extract("value", r'^(\d+)þ', 1)) \
    .withColumn("name", F.regexp_extract("value", r'þ([^þ]+)þ"', 1)) \
    .withColumn("role", F.concat(F.lit('"'), F.regexp_extract("value", r'þ"([^þ]+)þ', 1))) \
    .withColumn("exp", F.regexp_extract("value", r'þ\s*(\d+)þ', 1)) \
    .withColumn("task_desc", F.concat(F.regexp_extract("value", r'þ\s*([^"]+)"', 1), F.lit('"'))) \
    .withColumn("comp", F.regexp_extract("value", r'"þ\s*(\w+)$', 1)) \
    .drop("value")

# 查看结果
result_df.show(truncate=False)

两种方法都能输出你期望的DataFrame结构，处理引号内包含分隔符的特殊场景。

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