如何高效在PySpark中读取多CSV文件并跳过首尾行

我有多个无表头行、数据行数不一致的CSV文件，希望将它们读取为单个PySpark DataFrame。CSV文件结构示例如下：

data1,data2
data1,data2,data3
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2,data3,data4,data5,data6
data1,data2
data1,data2
data1,data2,data3

我需要在合并为单个DataFrame前，跳过每个文件的前2行和后3行。目前使用的方法可行，但因数据集较大，寻求更优化的方案：

def concat(df_list: list):
    df = df_list[0]
    for i in df_list[1:]:
        df = df.unionByName(i, allowMissingColumns=True)
    return df


def __read_with_separators(self, spark: SparkSession, field_details: List[Dict[str, Any]], file_path_list: List[str], kwargs: dict) -> DataFrame:
    df_list = []
    for file_path in file_path_list:
        rdd = spark.sparkContext.textFile(file_path)

        total_rows = rdd.count()
        start_index = kwargs.get("skiprows", 0)
        end_index = total_rows - kwargs.get("skipfooter", 0)

        rdd_filtered = rdd.zipWithIndex().filter(lambda x: start_index <= x[1] < end_index).map(lambda x: x[0]).map(lambda line: line.split(delimiter))
        temp_df = rdd_filtered.toDF(schema)
        df_list.append(temp_df)

    return concat(df_list)

疑问 #

1. 是否存在更高效的方法，可一次性读取多CSV文件并跳过指定首尾行？
2. 针对当前方法，有哪些优化措施可更高效处理大数据集？

针对问题1：一次性读取多文件并跳过首尾行的高效方法 #

可以利用Spark的wholeTextFiles实现批量读取+行过滤，避免逐个文件count()带来的重复全量扫描开销，具体思路：

• 用spark.sparkContext.wholeTextFiles()一次性读取所有目标文件，每个文件对应一个键值对（文件路径: 完整文本内容）
• 对每个文件的文本按换行符拆分为行列表，直接计算有效行范围（跳过前N行、后M行）
• 过滤有效行后拆分字段，最终生成单个DataFrame

示例代码：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType

def read_multiple_csv_with_skip(spark: SparkSession, file_paths: list, skiprows=2, skipfooter=3, delimiter=",", schema: StructType=None):
    # 批量读取所有文件
    whole_rdd = spark.sparkContext.wholeTextFiles(",".join(file_paths))
    
    # 处理每个文件的内容，过滤有效行并拆分字段
    processed_rdd = whole_rdd.flatMap(lambda file_data:
        lines = file_data[1].split("\n")
        # 计算有效行的起止索引，同时跳过空行
        start_idx = skiprows
        end_idx = len(lines) - skipfooter
        [line.split(delimiter) for line in lines[start_idx:end_idx] if line.strip()]
    )
    
    # 生成DataFrame，优先使用预定义schema避免自动推断开销
    return processed_rdd.toDF(schema) if schema else processed_rdd.toDF()

该方法仅需扫描一次所有文件，相比逐个文件处理的IO效率提升显著。

针对问题2：当前方法的优化措施 #

1. 移除逐个文件的count()操作：原方法中rdd.count()会触发全量扫描，大文件场景下耗时极高。可改用wholeTextFiles读取时直接通过行列表长度获取总行数，或提前用文件系统命令（如hadoop fs -count）批量获取文件行数。
2. 批量处理文件，减少中间DataFrame：原方法逐个生成小DataFrame再合并，会增加内存开销和合并成本。改为一次性处理所有文件生成单个RDD后转成DataFrame，避免多次中间对象创建。
3. 优化union操作：原concat函数循环调用unionByName，当DataFrame数量较多时会形成长依赖链。可改用reduce函数合并，或直接从RDD生成单个DataFrame彻底避免union操作：

   from functools import reduce
   def concat(df_list: list):
       return reduce(lambda df1, df2: df1.unionByName(df2, allowMissingColumns=True), df_list)
   

4. 固化schema：确保使用预定义的StructType作为schema，避免Spark自动推断schema带来的额外扫描和类型不一致问题。
5. 合并RDD转换步骤：原代码中多次map操作可合并，减少RDD转换次数，优化后的RDD处理逻辑：

   rdd_filtered = rdd.zipWithIndex().filter(lambda x: start_index <= x[1] < end_index).map(lambda x: x[0].split(delimiter))
   

6. 使用广播变量：将delimiter、schema等固定参数设为广播变量，减少节点间的数据传输开销。

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