如何用PySpark/Pandas合并嵌套文件夹中结构各异的多CSV文件？

针对你提到的三种场景（列顺序不同、无表头文件、含额外列），下面分别提供Python Pandas和PySpark的实现方案，适配嵌套文件夹的批量处理。

Pandas 实现（适合中小数据量） #

核心逻辑 #

1. 定义统一的目标表头，统一列名格式（比如全部转小写）
2. 递归遍历嵌套文件夹下所有CSV文件
3. 针对不同文件类型做适配：
   • 带表头文件：统一列名后按目标表头筛选，自动对齐列顺序，丢弃多余列，缺失列补NaN
   • 无表头文件：直接用目标表头作为列名读取
   • 列顺序/大小写差异：通过列名转小写+重索引自动对齐

代码实现 #

import pandas as pd
import os

# 定义最终要保留的统一表头（根据实际需求调整）
TARGET_COLUMNS = ['date', 'channel', 'spend', 'clicks']

def process_csv(file_path):
    try:
        # 尝试读取带表头的文件
        df = pd.read_csv(file_path)
        # 列名转小写并去除前后空格，消除大小写与格式差异
        df.columns = df.columns.str.strip().str.lower()
        # 按目标表头筛选列，缺失列补NaN，多余列直接丢弃
        df = df.reindex(columns=TARGET_COLUMNS)
    except pd.errors.ParserError:
        # 读取失败（大概率是无表头文件），用目标表头作为列名读取
        df = pd.read_csv(file_path, names=TARGET_COLUMNS)
    return df

def merge_csvs(root_dir):
    all_dataframes = []
    # 遍历嵌套文件夹
    for dirpath, _, filenames in os.walk(root_dir):
        for file in filenames:
            if file.lower().endswith('.csv'):
                full_path = os.path.join(dirpath, file)
                df = process_csv(full_path)
                all_dataframes.append(df)
    # 合并所有数据
    merged_df = pd.concat(all_dataframes, ignore_index=True)
    # 保存结果
    merged_df.to_csv('merged_pandas_result.csv', index=False)
    print(f"合并完成，共处理 {len(all_dataframes)} 个CSV文件")

# 替换为你的CSV根文件夹路径
merge_csvs('./csv_root_folder')

关键细节 #

• str.strip().str.lower()：处理列名前后空格和大小写差异（比如Date/date）
• reindex(columns=TARGET_COLUMNS)：自动对齐列顺序，不管原文件列顺序如何，都能匹配到目标列
• 异常处理：捕获无表头文件的读取错误，自动切换读取模式

PySpark 实现（适合大数据量） #

如果300+ CSV文件总数据量较大，Pandas可能出现内存不足，推荐用PySpark处理，天然支持分布式和大文件。

核心逻辑 #

1. 定义统一的Schema（指定列名和数据类型）
2. 递归读取嵌套文件夹下所有CSV
3. 自动适配场景：
   • 列顺序不同：通过Schema强制匹配列名，与顺序无关
   • 额外列：Schema会自动忽略多余列
   • 无表头文件：通过判断第一行是否为表头，决定是否跳过首行

代码实现 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("CSV_Merger") \
    .config("spark.sql.caseSensitive", "false")  # 关闭列名大小写敏感
    .getOrCreate()

# 定义统一Schema（根据实际数据类型调整）
TARGET_SCHEMA = StructType([
    StructField("date", StringType(), nullable=True),
    StructField("channel", StringType(), nullable=True),
    StructField("spend", FloatType(), nullable=True),
    StructField("clicks", IntegerType(), nullable=True)
])

def merge_csvs(root_dir):
    # 递归读取所有CSV，用Schema强制对齐列
    df_raw = spark.read.csv(
        path=f"{root_dir}/**/*.csv",
        header=False,
        schema=TARGET_SCHEMA,
        ignoreLeadingWhiteSpace=True,
        ignoreTrailingWhiteSpace=True
    )
    
    # 判断第一行是否是表头（检查是否包含目标列名）
    first_row_values = df_raw.limit(1).collect()[0].asDict().values()
    is_header_row = all(col in first_row_values for col in TARGET_COLUMNS)
    
    if is_header_row:
        # 跳过表头行
        df = df_raw.filter(~df_raw.date.isin(['Date', 'date']))
    else:
        df = df_raw
    
    # 合并为单个文件（数据量极大时建议去掉coalesce(1)，保留分区）
    df.coalesce(1).write.csv(
        path="merged_spark_result",
        header=True,
        mode="overwrite"
    )
    print("合并完成，结果保存到 merged_spark_result 文件夹")

# 替换为你的CSV根文件夹路径
merge_csvs('./csv_root_folder')

# 停止SparkSession
spark.stop()

关键细节 #

• spark.sql.caseSensitive=false：关闭列名大小写敏感，自动匹配Date/date
• schema=TARGET_SCHEMA：强制按指定列名和类型读取，自动忽略多余列，缺失列补null
• coalesce(1)：将多个分区合并为单个文件，方便后续处理（大数据量慎用，会影响性能）

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