Apache Spark结合Azure Data Lake Storage实现文件恰好一次处理问询

你当前考量的两类方案都存在可优化的空间：

• 直接建Delta表存已处理文件名+读全量数据做EXCEPT：需要每次全量枚举输入目录下数万个文件的元数据，在ADLS的网络延迟下枚举开销很高，且会把未处理和已处理的文件内容全部加载到Spark内存中再过滤，资源浪费严重。
• 文件迁移/重命名：ADLS Gen2的元数据操作是按请求计费、逐次触发的，10万级文件的重命名/迁移单次操作就会消耗数千秒的等待时间，重跑时还要重复执行这类操作，成本极高。

优化后的可落地方案 #

核心逻辑是完全不修改源目录的文件状态，用轻量的纯元数据追踪实现文件跳过，过滤动作下推到Spark文件扫描阶段，不做全量数据加载后再过滤，具体流程：

1. 提前创建一个仅存储文件元信息的轻量Delta追踪表
   • 表结构只需要两列：file_path STRING 存已处理文件的全路径，processed_ts TIMESTAMP 存处理时间，单条记录不到100字节，哪怕存储100万条已处理文件也仅占用几十MB存储空间，查询响应在毫秒级。
2. 调整原有数据读取逻辑，把过滤动作从“读完再筛”改成“扫描时跳过”
   • 启动Spark任务时，先从追踪表中拉取全部已处理的文件路径，做成内存集合广播到所有Executor（数万个路径的集合仅几MB大小，广播开销可忽略）。
   • 读取数据时，基于Spark内置的input_file_name()函数做下推过滤，Spark在列举文件阶段就会直接跳过匹配到的已处理路径，根本不会把这类文件的元数据分配到任务分区，更不会加载文件内容，相比“全量读入再EXCEPT”的模式性能提升10~100倍。
3. 保障幂等性的收尾逻辑
   • 完成数据转换、写入落地后，再把本次任务实际处理的所有文件路径，通过Delta的MERGE语法写入追踪表：匹配到已存在的文件路径就跳过写入，仅插入未记录的路径，避免任务重试导致追踪表出现重复记录。
   • 整个流程的幂等性可以完全保障：如果任务执行中途失败，只要还没写入追踪表，下次重试时这批文件就会被重新加载处理；如果任务成功后执行追踪表写入，哪怕同一个批次重复跑两次，第二次任务也会因为所有文件都在追踪表中，直接返回空结果，不会产生重复处理的开销。

配套的性能优化参数 #

因为场景是数万个小文件，需要搭配Spark的内置参数降低小文件处理的额外开销：

• 配置spark.sql.files.ignoreMissingFiles=true，避免处理过程中源文件被意外修改导致任务失败
• 配置spark.sql.files.maxPartitionBytes=134217728（即128MB），让Spark自动把分散的小文件合并到同一个分区，避免一个小文件对应一个任务的调度开销
• 开启spark.sql.files.openCostInBytes=8388608，调整小文件的打开开销阈值，让合并逻辑更适配数MB大小的JSON文件

重跑场景的适配 #

如果后续需要重新处理某一批历史文件，只需要在追踪表中按processed_ts的时间范围删除对应记录即可，相当于把这批文件标记为未处理状态，整个操作仅需要执行几行SQL，不需要在ADLS上做任何文件迁移、重命名的操作，重跑耗时和首次处理完全一致。

调整后的核心代码示例 #

from pyspark.sql.functions import col, input_file_name, current_timestamp

# 1. 广播已处理文件列表
processed_path_set = {row.file_path for row in spark.read.table("processed_file_tracker").collect()}
processed_path_bc = spark.sparkContext.broadcast(processed_path_set)

# 2. 带过滤的读取操作（过滤下推到文件扫描阶段）
raw_df = (
    spark.read
    .json("/mnt/input_location/*.json")
    .where(~col(input_file_name()).isin(processed_path_bc.value))
)

# 3. 执行自定义转换逻辑
processed_df = custom_transform_logic(raw_df)

# 4. 结果落地写入
processed_df.write.format("delta").mode("append").save("/mnt/output_location")

# 5. 合并写入本次处理的文件名单到追踪表
current_batch_files = raw_df.selectExpr("DISTINCT input_file_name() as file_path").withColumn("processed_ts", current_timestamp())
current_batch_files.createOrReplaceTempView("current_batch")
spark.sql("""
MERGE INTO processed_file_tracker t
USING current_batch s
ON t.file_path = s.file_path
WHEN NOT MATCHED THEN INSERT *
""")

方案对比：

• 相比原有的“读全量做EXCEPT”思路：跳过了全量文件元数据枚举、全量内容加载的开销，过滤动作在扫描阶段完成，处理10万级小文件的元数据耗时从分钟级降到秒级。
• 相比原有的“文件迁移/重命名”思路：全程不修改源存储的文件结构，没有ADLS元数据操作的额外请求开销，重跑成本几乎为零。

内容的提问来源于stack exchange，提问作者Łukasz Kastelik