如何基于DataFrame中的文件名更新Databricks指定目录下的JSON文件？

我有如下Spark DataFrame：

input_df = self.spark.createDataFrame(
            data=[
                ("01", "file_name_1"),
                ("02", "file_name_2"),
                ("05", "file_name_5"),
            ],
            schema=(
                "RECORD_ID: string, FILE_NAME: string"
            ),
        )

目录/mnt/data/project/integration_test/下当前存在这些文件：

file_name_1.json
file_name_2.json
file_name_3.json
file_name_4.json

我需要更新该目录中与input_df内文件名匹配的JSON文件，计划执行以下流程：

• 删除文件名出现在input_df中的JSON文件
• 将input_df的每一行保存为单独的JSON文件（此步骤已实现）

最终目录下的文件应变为：

file_name_1.json (已更新)
file_name_2.json (已更新)
file_name_3.json 
file_name_4.json
file_name_5.json (新建)

方法1：Databricks环境下使用dbutils实现删除 #

先从input_df提取需删除的文件名并拼接.json后缀，再遍历执行删除：

# 提取目标文件名并补全后缀
file_names_to_delete = [row.FILE_NAME + ".json" for row in input_df.select("FILE_NAME").collect()]
target_dir = "/mnt/data/project/integration_test/"

# 遍历删除存在的目标文件
for file_name in file_names_to_delete:
    file_path = target_dir + file_name
    if dbutils.fs.exists(file_path):
        dbutils.fs.rm(file_path)

方法2：本地/可直接访问的挂载目录使用Python原生操作 #

借助os模块完成文件删除：

import os

target_dir = "/mnt/data/project/integration_test/"
file_names_to_delete = [row.FILE_NAME + ".json" for row in input_df.select("FILE_NAME").collect()]

# 遍历删除存在的目标文件
for file_name in file_names_to_delete:
    file_path = os.path.join(target_dir, file_name)
    if os.path.exists(file_path):
        os.remove(file_path)

注意事项 #

• 删除前建议先打印待删文件路径，确认无误后再执行操作，避免误删
• 针对分布式文件系统（如HDFS），需确保执行代码的节点具备对应删除权限

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