Spark处理JSON文件后如何移动文件避免重复处理？

针对你遇到的问题，以下是适配Databricks+Blob存储场景的可行解决方法：

方法一：批处理场景手动移动文件 #

核心思路是先捕获所有被处理的JSON文件路径，在Driver端用dbutils.fs.mv完成移动（规避Executor端操作的限制）。

完整代码示例 #

from pyspark.sql.functions import input_file_name
import os

# 1. 读取JSON时记录每行的源文件路径
df = spark.read.option("multiline", "true").json(f"/mnt/bronze/{something}*") \
    .withColumn("source_file", input_file_name())

# 2. 执行你的数据处理逻辑
rows = df.select(...)  # 替换为实际处理代码

# 3. 写入Delta表
rows.write.format("delta").mode("overwrite").save(f"/mnt/silver/{something}")

# 4. 获取所有已处理的唯一文件路径
processed_files = df.select("source_file").distinct().rdd.map(lambda x: x[0]).collect()

# 5. 批量移动文件到归档目录
archive_base = "/mnt/bronze/processed/"
for file_path in processed_files:
    # 构造目标路径，保留原文件层级结构
    relative_path = file_path.replace("/mnt/bronze/", "")
    target_path = f"{archive_base}{relative_path}"
    
    # 创建目标父目录，避免路径不存在报错
    dbutils.fs.mkdirs(os.path.dirname(target_path))
    
    # 执行移动操作
    dbutils.fs.mv(file_path, target_path)

之前方法失败的原因 #

• rdd.foreach()中用dbutils：rdd.foreach()在Executor节点执行，而dbutils是Driver端工具，Executor无对应上下文环境，无法调用。必须在Driver端收集文件路径后统一操作。
• shutil.move()：shutil是本地文件系统工具，无法识别Databricks挂载的Blob存储（分布式文件系统）路径，必须用dbutils.fs系列工具或对应云厂商存储SDK操作。

方法二：增量处理场景用Auto Loader（推荐） #

如果是持续增量文件处理，Databricks的Auto Loader可自动跟踪已处理文件并完成归档，无需手动管理路径：

代码示例 #

from pyspark.sql.functions import input_file_name

# 初始化Auto Loader流读取
df_stream = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .option("multiline", "true") \
    .option("cloudFiles.schemaLocation", f"/mnt/silver/{something}_schema") \
    .load(f"/mnt/bronze/{something}") \
    .withColumn("source_file", input_file_name())

# 处理数据
processed_stream = df_stream.select(...)  # 你的处理逻辑

# 写入Delta表并自动归档已处理文件
write_query = processed_stream.writeStream \
    .format("delta") \
    .option("checkpointLocation", f"/mnt/silver/{something}_checkpoint") \
    .option("cloudFiles.moveAfterRead", "/mnt/bronze/processed/") \
    .start(f"/mnt/silver/{something}")

write_query.awaitTermination()

优势 #

• 自动识别新增文件，无需手动过滤
• 内置文件跟踪机制，避免重复处理
• 自动完成文件归档，无需额外代码

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