如何编写函数将Spark DataFrame列表按序转Pandas并导出CSV

当然可以实现，以下是符合需求的函数实现：

import pyspark
import os

def export_spark_dfs_to_csv(spark_dfs, output_dir):
    # 确保输出目录存在，不存在则自动创建
    os.makedirs(output_dir, exist_ok=True)
    
    # 遍历DataFrame列表，同时获取顺序索引
    for idx, spark_df in enumerate(spark_dfs, start=1):
        # 转换为Pandas DataFrame
        pandas_df = spark_df.toPandas()
        # 生成按顺序命名的文件路径
        csv_path = os.path.join(output_dir, f"data_{idx:02d}.csv")
        # 导出CSV文件（可选index=False去掉行索引）
        pandas_df.to_csv(csv_path, index=False)

使用示例 #

# 假设你已准备好Spark DataFrame列表dfs
dfs = [df1, df2, df3, df4, df5, df6, df7, df8, df9, df10, df11, df12, df13, df14, df15]
# 调用函数，指定目标保存目录
export_spark_dfs_to_csv(dfs, "D:/data")

关键说明 #

• 顺序命名：通过enumerate从1开始计数，生成data_01.csv、data_02.csv这类有序文件名，完全匹配输入列表的顺序
• 目录兼容：os.makedirs的exist_ok=True参数避免了目录已存在时的报错
• 避免覆盖：每个文件使用唯一序号命名，解决了你原代码中循环覆盖同一文件的问题
• 性能优化选项：如果Spark DataFrame数据量较大，转Pandas可能占用过多内存，可直接用Spark原生导出方法跳过Pandas转换，效率更高：

  def export_spark_dfs_to_csv_direct(spark_dfs, output_dir):
      os.makedirs(output_dir, exist_ok=True)
      for idx, spark_df in enumerate(spark_dfs, start=1):
          csv_path = os.path.join(output_dir, f"data_{idx:02d}")
          spark_df.write.csv(csv_path, header=True, mode="overwrite")
  

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