Apache Airflow中如何识别并跳过已处理文件避免重复加工

你目前采用本地日志文件记录已处理文件的思路可以实现基础功能，但存在并发不安全、状态与流程耦合、调度器解析异常执行等问题，不推荐在生产环境使用。Airflow本身提供了多种原生能力可以直接实现已处理文件自动过滤，不需要自行维护本地日志。

可选实现方案 #

1. 2.4及以上版本优先用Dataset数据集能力 #

Airflow 2.4版本推出的Dataset数据感知特性是目前最贴合该场景的原生方案：

• 你可以将raw_data下每个待处理文件标记为输入Dataset，clean_data下转换完成的文件标记为输出Dataset
• Airflow会自动在元数据库中维护所有Dataset的生产、消费映射关系，天然记录哪些原始文件已经完成转换
• 不需要自行编写任何已处理文件判断逻辑，平台自动保证只有未处理的新文件会触发转换流程，同时支持跨DAG的依赖触发。

2. 兼容低版本的通用实现 #

如果使用2.4以下版本，直接复用Airflow自带的元数据存储能力记录已处理文件即可，可靠性远高于本地日志文件：

• 不要在DAG顶层编写目录扫描、文件处理逻辑：Airflow调度器会默认每30秒解析一次DAG文件，顶层代码会被反复执行，导致业务逻辑提前跑、调度器性能被拖慢
• 已处理文件列表存在Airflow Variable中：元数据库写入自带并发锁，不会出现多进程同时写导致文件损坏的问题，持久化可靠性远高于本地日志
• 状态更新放在任务成功后执行：只有文件转换逻辑执行成功，才把文件名追加到已处理列表中，避免处理成功但日志写入失败导致的重复处理。

参考实现代码 #

import os
import shutil
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models.variable import Variable
from datetime import datetime

def process_untreated_files():
    raw_dir = './raw_data'
    clean_dir = './clean_data'
    # 从元数据库读取已处理文件列表，默认返回空列表
    already_treated = Variable.get(
        "processed_raw_files",
        default_var=[],
        deserialize_json=True
    )
    current_files = os.listdir(raw_dir)
    newly_processed = []

    for filename in current_files:
        raw_file_path = os.path.join(raw_dir, filename)
        # 跳过子目录、已处理文件
        if not os.path.isfile(raw_file_path) or filename in already_treated:
            print(f"跳过已处理/非文件项：{filename}")
            continue
        # 执行自定义转换逻辑
        file_suffix = os.path.splitext(filename)[1]
        target_filename = filename.replace(file_suffix, "_transformed.txt")
        target_path = os.path.join(clean_dir, target_filename)
        shutil.copy(raw_file_path, target_path)
        newly_processed.append(filename)
    
    # 批量更新已处理文件列表，减少元数据库连接次数
    if newly_processed:
        Variable.set(
            "processed_raw_files",
            already_treated + newly_processed,
            serialize_json=True
        )

with DAG(
    dag_id="incremental_file_etl",
    start_date=datetime(2024, 1, 1),
    schedule_interval="*/5 * * * *", # 每5分钟扫描一次目录
    catchup=False,
    tags=['file_etl']
) as dag:
    file_process_task = PythonOperator(
        task_id="process_new_raw_files",
        python_callable=process_untreated_files
    )

如果单目录下文件量级超过10万，不建议用Variable存储列表，可以在Airflow连接的数据库中建一张轻量元数据表，存储已处理文件名、处理时间、文件MD5值即可，扩展性更好。

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