如何在Spark中读取单元格含多行文本的CSV文件

当CSV文件中某列包含带换行符的多行文本时，默认的Spark CSV读取器会将换行符识别为行分隔符，导致字段错位。以下是几种可行的解决方法：

方法一：启用多行读取模式（推荐） #

Spark内置的CSV数据源支持multiLine配置，开启后会将被引号包裹的多行内容识别为单个字段值，避免换行符拆分行。同时确保指定正确的引号字符（通常是双引号）：

from pyspark.sql import SparkSession

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

# 读取含多行文本的CSV文件
df = spark.read \
    .option("header", "true") \
    .option("multiLine", "true") \
    .option("quote", "\"") \  # 匹配文件中包裹多行文本的引号
    .option("escape", "\"") \  # 处理引号转义（如用""表示单个"）
    .csv("你的文件路径.csv")

df.show(truncate=False)  # 查看完整字段内容

如果文件使用其他分隔符（如制表符），需额外添加.option("delimiter", "\t")指定分隔符。

方法二：预处理文件合并多行文本 #

如果多行文本未被引号包裹，需要先通过RDD预处理合并字段内的换行：

from pyspark.sql import SparkSession

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

# 读取原始文本RDD
text_rdd = spark.sparkContext.textFile("你的文件路径.csv")

def merge_multi_line(iterator):
    # 提取表头
    header = next(iterator)
    yield header
    current_row = ""
    # 按表头的分隔符数量，判断正常行的字段数（需根据你的文件结构调整）
    expected_delimiters = header.count(",")
    
    for line in iterator:
        if line.count(",") < expected_delimiters:
            # 当前行是字段内的换行，合并到上一行
            current_row += " " + line.strip()
        else:
            if current_row:
                yield current_row
                current_row = ""
            yield line
    # 处理最后一行未完成的合并
    if current_row:
        yield current_row

# 处理后的RDD转换为DataFrame
processed_rdd = text_rdd.mapPartitions(merge_multi_line)
df = spark.read.csv(processed_rdd, header=True)

df.show()

这种方法需要根据文件的列数（分隔符数量）判断哪些换行属于字段内，逻辑需适配你的具体文件结构。

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