Spark DataFrame导出带表头CSV并指定文件名（AWS S3环境）

我来帮你搞定这个问题！Spark作为分布式框架，默认会生成带随机后缀的分区文件，但我们完全可以不用os.system，直接通过Spark和Hadoop的API来实现指定文件名+保留表头的需求，下面给你两种可行的方案：

方案一：合并分区+HDFS API重命名（通用所有Spark版本） #

这个方法的核心是先把数据合并到单个分区，再通过Hadoop的文件系统API重命名生成的part文件，全程不需要调用外部系统命令。

Python代码示例 #

from pyspark.sql import SparkSession

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

# 假设你的目标DataFrame是df，这里用读取示例数据代替
df = spark.read.parquet("s3://your-input-bucket/your-data.parquet")

# 1. 先将数据合并到1个分区，写入临时路径并保留表头
temp_s3_path = "s3://your-output-bucket/temp-csv-folder"
df.coalesce(1).write.mode("overwrite").option("header", "true").csv(temp_s3_path)

# 2. 使用Hadoop FileSystem API找到临时路径下的part文件并重命名
hadoop_conf = spark._jsc.hadoopConfiguration()
fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(hadoop_conf)
temp_path = spark._jvm.org.apache.hadoop.fs.Path(temp_s3_path)

# 过滤出真正的CSV part文件（排除_SUCCESS、.crc等辅助文件）
part_files = []
for status in fs.listStatus(temp_path):
    file_name = status.getPath().getName()
    if file_name.startswith("part-") and file_name.endswith(".csv"):
        part_files.append(status.getPath())

if part_files:
    # 重命名为你想要的文件名
    target_path = spark._jvm.org.apache.hadoop.fs.Path("s3://your-output-bucket/final-folder/part-00000.csv")
    fs.rename(part_files[0], target_path)
    
    # 删除临时文件夹，清理冗余文件
    fs.delete(temp_path, True)

关键细节说明 #

• coalesce(1)：将所有数据合并到一个分区，这样只会生成一个part文件，注意：如果数据量极大，这会导致单个Executor压力过大，适合中小数据集使用；如果是大数据集，建议接受多分区文件，或者考虑其他方式。
• option("header", "true")：必须添加这个参数才能在CSV中保留表头。
• 用Hadoop的FileSystem API操作S3：Spark本身集成了Hadoop的文件系统客户端，不需要额外依赖，也避开了os.system的限制，而且能直接跨集群操作S3文件。

方案二：Spark 3.3+专属：使用Spark SQL的COPY TO命令（更简洁） #

如果你使用的是Spark 3.3及以上版本，可以利用新增的功能直接指定输出文件名，无需处理临时文件：

# 注册DataFrame为临时视图
df.createOrReplaceTempView("temp_view")

# 使用COPY TO命令直接导出到指定文件名
spark.sql("""
    COPY TO 's3://your-output-bucket/final-folder/part-00000.csv'
    FROM temp_view
    WITH (FORMAT = 'CSV', HEADER = true)
""")

这个方法代码更简洁，但仅限Spark 3.3及以上版本使用。

注意事项 #

• 确保你的Spark作业拥有S3的读写权限（比如通过IAM角色、Access Key等配置）。
• 如果使用coalesce(1)处理超大数据集，可能会引发性能瓶颈，此时建议放弃单个文件的需求，或者先对数据做过滤/抽样后再导出。

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