如何为PySpark DataFrame生成的CSV文件自定义命名？

核心思路 #

PySpark默认生成part-*.csv是分布式计算的特性导致——数据分散在多个分区，每个分区对应一个part文件。要生成单个自定义命名文件，核心是先将数据合并到一个分区，再通过文件系统操作重命名生成的part文件，全程无需依赖Pandas中转。

具体实现步骤 #

方法1：Spark文件系统API（跨环境通用） #

这种方法不依赖外部命令，纯Spark API实现，适配本地、YARN、K8s等各类运行环境：

1. 合并数据到单分区：用coalesce(1)（无shuffle，性能优于repartition(1)）将DataFrame合并为一个分区，写入临时目录。
2. 定位part文件：遍历临时目录，筛选出唯一的part-*.csv文件。
3. 重命名+清理：用Spark内置的Hadoop文件系统API将part文件改名为目标格式，最后删除临时目录及冗余文件（如_SUCCESS）。

代码示例：

from pyspark.sql import SparkSession
from datetime import datetime

# 初始化SparkSession
spark = SparkSession.builder.appName("CustomCSVExport").getOrCreate()

# 假设你的业务DataFrame为df
# df = spark.read.table("your_source_table")

# 构造目标文件名（按YYYYMMDD格式生成）
current_date = datetime.now().strftime("%Y%m%d")
target_filename = f"extraction_on_{current_date}.csv"
target_full_path = f"/your/target/dir/{target_filename}"

# 临时存储目录（后续会删除）
temp_dir = "/your/temp/dir"

# 写入临时目录，开启表头
df.coalesce(1).write.mode("overwrite").csv(temp_dir, header=True)

# 获取Hadoop文件系统实例
fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
temp_path = spark._jvm.org.apache.hadoop.fs.Path(temp_dir)

# 找到临时目录里的part文件
part_file_path = None
for file_status in fs.listStatus(temp_path):
    file_name = file_status.getPath().getName()
    if file_name.startswith("part-") and file_name.endswith(".csv"):
        part_file_path = file_status.getPath().toString()
        break

# 重命名part文件到目标路径
if part_file_path:
    fs.rename(
        spark._jvm.org.apache.hadoop.fs.Path(part_file_path),
        spark._jvm.org.apache.hadoop.fs.Path(target_full_path)
    )
    # 清理临时目录
    fs.delete(temp_path, True)

方法2：结合Shell命令（适合本地/可控集群环境） #

如果你的运行环境允许执行shell命令，可以简化文件查找和重命名步骤：

import os
import shutil
from datetime import datetime

current_date = datetime.now().strftime("%Y%m%d")
target_filename = f"extraction_on_{current_date}.csv"
temp_dir = "/your/temp/dir"
target_dir = "/your/target/dir"

# 写入临时目录
df.coalesce(1).write.mode("overwrite").csv(temp_dir, header=True)

# 找到part文件并改名（本地环境示例，HDFS环境替换为`hdfs dfs`命令）
for file in os.listdir(temp_dir):
    if file.startswith("part-") and file.endswith(".csv"):
        os.rename(f"{temp_dir}/{file}", f"{target_dir}/{target_filename}")
        break

# 删除临时目录
shutil.rmtree(temp_dir)

关键注意事项 #

• 数据量限制：coalesce(1)会把所有数据集中到单个Executor节点，数据量过大时可能引发内存溢出或性能瓶颈。如果数据量极大，建议和下游系统沟通是否支持多part文件，或者拆分数据分批导出。
• 性能对比：原方法用Pandas中转需要将全量数据拉到Driver节点，不仅容易OOM，性能也远低于Spark分布式处理，建议优先用上述纯Spark方案。

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