Spark2.2/2.3中PySpark读取含换行符CSV时multiLine选项失效

我明白你遇到的困扰——明明设置了multiLine=True，但Spark还是把数据里的换行当成了新记录的分隔符，导致完整数据被拆得七零八落。咱们一步步来排查和解决这个问题：

首先搞懂multiLine=True的生效条件 #

Spark的multiLine选项只对被引号包裹的字段内的换行有效。也就是说，如果你的数据里包含换行的字段没有用引号（默认是双引号）括起来，multiLine=True根本不会起作用。看你提供的数据示例，字段里的换行是直接裸露的，这就解释了为什么设置了参数还是没用。

针对性解决方案 #

方案1：给含换行的字段添加引号（如果能修改数据的话） #

如果可以预处理数据，把包含换行的字段用引号包裹起来，比如把原数据：

name,test,12345,

,desc
name2,test2,12345,

,desc2

改成：

name,test,12345,"

",desc
name2,test2,12345,"

",desc2

然后再用你的原代码读取，multiLine=True就能正确识别字段内的换行，不会拆分记录了：

spark.read.csv(file_path, schema=schema, sep=delimiter, multiLine=True)

方案2：自定义记录分隔符（如果记录有明确边界） #

如果数据里的记录之间有唯一的分隔标识（比如每个记录都以name,开头），可以先读取整个文件的文本内容，再手动拆分出完整记录：

步骤1：读取整个文件为单条文本

df_whole = spark.read.text(file_path, wholetext=True)

步骤2：按记录起始标识拆分并展开

用正则表达式的正向预查，匹配每个记录的开头（比如name,），把大文本拆成单个记录：

from pyspark.sql.functions import split, explode

df_split = df_whole.withColumn(
    "records", 
    split(df_whole.value, "(?=name,)")  # 正向预查，确保拆分后每个记录以name,开头
).select(explode("records").alias("record"))

步骤3：解析单个记录为CSV

把拆分后的每条记录传给CSV读取器，这时候multiLine=True就能正常工作了：

df = spark.read.csv(
    df_split.rdd.map(lambda x: x.record), 
    schema=schema, 
    sep=delimiter, 
    multiLine=True
)

方案3：指定精确的行分隔符（如果记录和字段换行不同） #

如果你的记录之间用的是CRLF（\r\n）分隔，而字段内的换行是LF（\n），可以直接指定lineSep参数来区分：

spark.read.csv(
    file_path, 
    schema=schema, 
    sep=delimiter, 
    multiLine=True, 
    lineSep="\r\n"
)

要是记录之间是两个连续的CRLF（也就是空行分隔），那就把lineSep设为"\r\n\r\n"。

方案4：用RDD手动合并行 #

如果以上方法都不适用，可以用RDD的mapPartitions来逐行合并，把属于同一条记录的行拼接起来：

raw_data = spark.sparkContext.textFile(file_path)

def combine_records(records):
    current_record = []
    for line in records:
        # 假设每条记录以"name,"开头，判断是否是新记录的起始
        if line.strip().startswith("name,"):
            if current_record:
                yield "\n".join(current_record)
                current_record = []
            current_record.append(line)
        else:
            current_record.append(line)
    # 处理最后一条记录
    if current_record:
        yield "\n".join(current_record)

# 合并后得到完整记录的RDD
combined_rdd = raw_data.mapPartitions(combine_records)

# 再解析为DataFrame
df = spark.read.csv(combined_rdd, schema=schema, sep=delimiter, multiLine=True)

额外检查点 #

• 确认你的Spark版本是2.2及以上，因为multiLine选项是从这个版本开始支持的；
• 检查sep=delimiter里的delimiter是否正确设置为数据的分隔符（比如逗号）。

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