Spark 3.2.1加载含多行记录且表头在第5行的CSV遇问题求解决

你遇到的问题根源在于：当把RDD[String]传给spark.read.csv时，Spark会将RDD的每个元素视为独立行，multiLine参数会被忽略——因为此时多行记录已经被拆分成单个字符串元素，Spark不会再尝试合并它们。

下面提供两种可行的解决思路：

方案一：使用Spark CSV内置的skipRows参数（推荐） #

Spark的CSV数据源原生支持skipRows选项，直接跳过指定行数后读取表头，同时保留multiLine的功能，无需手动处理RDD：

sourcePath = "s3://mybucket/location/file.csv"
skip_rows = 4

df = spark.read.csv(
    sourcePath,
    header=True,
    multiLine=True,
    skipRows=skip_rows
)

这个方法简洁高效，直接利用Spark内置逻辑处理跳过行和多行记录，避免手动操作RDD带来的问题。

方案二：先读取全文件再过滤行（适用于复杂场景） #

如果因为某些原因无法使用skipRows，可以先读取整个文件保留多行结构，再手动提取表头并过滤数据行：

sourcePath = "s3://mybucket/location/file.csv"
skip_rows = 4

# 1. 读取整个文件，保留多行结构，不指定表头
raw_df = spark.read.csv(sourcePath, multiLine=True, header=False)

# 2. 给DataFrame添加索引，方便过滤行
from pyspark.sql.functions import monotonically_increasing_id
raw_df_with_index = raw_df.withColumn("row_index", monotonically_increasing_id())

# 3. 提取表头行（第5行对应索引4）
header_row = raw_df_with_index.filter(raw_df_with_index.row_index == skip_rows).collect()[0]
headers = [str(col) for col in header_row if col != header_row["row_index"]]

# 4. 过滤掉前skip_rows+1行（包括表头行之前的行和表头行）
data_df = raw_df_with_index.filter(raw_df_with_index.row_index > skip_rows).drop("row_index")

# 5. 用表头重命名列
final_df = data_df.toDF(*headers)

注意：monotonically_increasing_id()生成的索引是全局唯一且递增的，但不一定连续，如果你的文件是单个文件，这个方法可以正常工作；如果是多文件分区，可能需要调整索引生成方式。

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