PySpark写出DataFrame为CSV后如何重命名生成的part前缀文件

PySpark中同样可以通过调用Hadoop FileSystem接口实现文件重命名，和Scala版本逻辑一致，实现代码如下：

from org.apache.hadoop.fs import FileSystem, Path

# 获取Hadoop配置及FileSystem实例
hadoop_conf = spark._jsc.hadoopConfiguration()
fs = FileSystem.get(hadoop_conf)

# 手动指定源文件和目标路径方式
src_path = Path("csvDirectory/data.csv/part-00000-xxxx-xxxx-xxxx-xxxxxxxxxxxx-c000.csv")
dest_path = Path("csvDirectory/newData.csv")

if fs.exists(src_path):
    fs.rename(src_path, dest_path)
    # 重命名完成后可按需删除原输出目录
    # fs.delete(Path("csvDirectory/data.csv"), True)
else:
    print("源文件不存在，请检查路径配置")

如果不知道完整的part文件名，可以通过遍历目录自动匹配符合规则的文件：

output_dir = Path("csvDirectory/data.csv/")
file_status_list = fs.listStatus(output_dir)

for file_status in file_status_list:
    file_path = file_status.getPath()
    file_name = file_path.getName()
    # 匹配part前缀、csv后缀的文件
    if file_name.startswith("part-") and file_name.endswith(".csv"):
        fs.rename(file_path, Path("csvDirectory/newData.csv"))
        break

注意事项 #

• 执行重命名前建议先对DataFrame执行df.coalesce(1).write.csv(输出目录)操作，保证输出目录下只有1个CSV文件，避免多文件匹配错误。
• 上述代码支持本地文件路径和HDFS路径，无需额外适配不同存储介质。
• 若业务场景需要输出多分区CSV文件，可遍历所有part文件后按业务规则逐个重命名即可。

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