使用Databricks Pyspark写入Azure Blob Storage时如何指定覆盖后的CSV文件名

Spark为分布式计算引擎，默认写入操作会根据DataFrame的分区数生成对应数量的part-开头的分片文件，且save()方法传入的参数为目录路径，会自动生成对应目录存放分片文件和_SUCCESS等元数据文件，这是Spark的原生设计，无法直接通过write参数指定输出文件名。

可通过「先写临时目录+移动重命名文件」的方式实现保留原文件名写入，完整实现代码如下：

# 读取原始CSV并修改字段名
df = spark.read.csv("/mnt/ndemo/nsalman/addresses.csv", inferSchema = True)
df = df.toDF("firstName","lastName","street","town","city","code")

# 合并为单分区，确保仅生成1个分片文件（仅适合数据量不大的场景）
df_single_part = df.coalesce(1)

# 写入临时目录（无需提前创建，Spark会自动生成）
temp_dir = "/mnt/ndemo/nsalman/temp_write_dir"
df_single_part.write.format('com.databricks.spark.csv') \
  .mode('overwrite').option("header", "true").save(temp_dir)

# 筛选出临时目录下的CSV分片文件，排除元数据文件
part_file_path = [file.path for file in dbutils.fs.ls(temp_dir) if file.name.startswith("part-") and file.name.endswith(".csv")][0]

# 移动分片文件到原路径，重命名为原始文件名完成覆盖
target_file_path = "/mnt/ndemo/nsalman/addresses.csv"
dbutils.fs.mv(part_file_path, target_file_path)

# 删除临时目录清理冗余文件
dbutils.fs.rm(temp_dir, recurse=True)

注意事项 #

• coalesce(1)会将全量数据拉到单个Executor上处理，仅适合数据量不大的场景，若数据量过大会导致任务运行缓慢甚至内存溢出
• 操作前建议先备份原始CSV文件，避免误操作导致数据丢失
• Databricks环境默认内置dbutils工具，可直接操作已挂载的Azure Blob Storage路径

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