如何通过Databricks将多个PySpark DataFrame批量写入Azure容器指定文件夹

当然可以实现！我来给你分享两种实用的方案，不管是当前的3个DataFrame还是以后要扩展更多，都能轻松搞定：

方案一：循环批量处理+文件重命名 #

这种方法适合快速上手，把需要写入的DataFrame和对应的输出文件名配对，然后循环处理每个DataFrame——先写入临时目录，再把生成的单个part文件移动并重命名到目标文件夹，避免文件覆盖。

前置准备：确保Azure容器已挂载到Databricks #

首先你需要先把Azure Blob容器挂载到Databricks的文件系统中（如果还没挂载的话），示例代码如下（替换成你的存储账户信息）：

container_name = "你的容器名称"
storage_account_name = "你的存储账户名称"
sas_token = "你的SAS令牌"

# 挂载容器
dbutils.fs.mount(
    source=f"wasbs://{container_name}@{storage_account_name}.blob.core.windows.net/",
    mount_point="/mnt/myazurecontainer",
    extra_configs={f"fs.azure.sas.{container_name}.{storage_account_name}.blob.core.windows.net": sas_token}
)

批量写入代码 #

# 1. 定义目标文件夹路径（已挂载的Azure容器下的df_cleaned文件夹）
target_folder = "/mnt/myazurecontainer/df_cleaned"
# 创建目标文件夹（如果不存在）
dbutils.fs.mkdirs(target_folder)

# 2. 把DataFrame和对应的输出文件名配对成列表
df_file_pairs = [
    (df1, "df1_cleaned.csv"),
    (df2, "df2_cleaned.csv"),
    (df3, "df3_cleaned.csv")
]

# 3. 循环处理每个DataFrame
for df, filename in df_file_pairs:
    # 临时文件夹路径，每个DataFrame用独立临时目录避免冲突
    temp_path = f"/tmp/temp_{filename}"
    # 将DataFrame写入临时目录，coalesce(1)确保只生成一个part文件
    df.coalesce(1).write.format("csv").mode("overwrite").option("header", "true").save(temp_path)
    # 找到临时目录中的part文件（过滤出包含"part-"的文件）
    part_files = [file.path for file in dbutils.fs.ls(temp_path) if "part-" in file.name]
    if part_files:
        # 移动part文件到目标文件夹并重命名为指定文件名
        dbutils.fs.mv(part_files[0], f"{target_folder}/{filename}")
    # 删除临时文件夹释放空间
    dbutils.fs.rm(temp_path, recurse=True)

方案二：封装成可复用函数 #

如果以后需要频繁处理这类需求，可以把写入逻辑封装成函数，代码更简洁，复用性更强：

def write_df_to_azure(df, target_filename, target_folder):
    # 确保目标文件夹存在
    dbutils.fs.mkdirs(target_folder)
    # 临时存储路径
    temp_path = f"/tmp/temp_{target_filename}"
    # 写入临时目录
    df.coalesce(1).write.format("csv").mode("overwrite").option("header", "true").save(temp_path)
    # 查找part文件
    part_files = [file.path for file in dbutils.fs.ls(temp_path) if "part-" in file.name]
    if part_files:
        # 移动并重命名到目标路径
        dbutils.fs.mv(part_files[0], f"{target_folder}/{target_filename}")
    # 清理临时文件
    dbutils.fs.rm(temp_path, recurse=True)

# 定义目标文件夹
target_folder = "/mnt/myazurecontainer/df_cleaned"

# 调用函数处理每个DataFrame
write_df_to_azure(df1, "df1_cleaned.csv", target_folder)
write_df_to_azure(df2, "df2_cleaned.csv", target_folder)
write_df_to_azure(df3, "df3_cleaned.csv", target_folder)

额外补充：合并DataFrame后写入（如果需求是合并成单个文件） #

如果你不需要保留三个DataFrame的独立文件，而是希望把它们合并成一个文件写入目标文件夹，可以用unionByName（确保三个DataFrame的列名完全一致）：

# 合并三个DataFrame（列名必须一致，否则需要调整列顺序或重命名）
combined_df = df1.unionByName(df2).unionByName(df3)
# 写入目标文件夹
combined_df.coalesce(1).write.format("csv").mode("overwrite").option("header", "true").save("/mnt/myazurecontainer/df_cleaned")

备注：内容来源于stack exchange，提问作者Debasis