如何在用户上传任意文件时自动触发 Apache Airflow 任务

本文介绍一种轻量、可控的方案：通过自定义 pythonoperator 监控指定上传目录，结合文件时间戳与 airflow 内置时间变量（如 data_interval_start）识别新增文件，并动态触发下游任务，避免依赖固定文件名或外部服务。

本文介绍一种轻量、可控的方案：通过自定义 pythonoperator 监控指定上传目录，结合文件时间戳与 airflow 内置时间变量（如 data_interval_start）识别新增文件，并动态触发下游任务，避免依赖固定文件名或外部服务。

在 Apache Airflow 中，原生 FileSensor 仅支持对已知路径+确定文件名的轮询等待，无法满足“监听任意用户上传文件并即时响应”的场景。若需实现“上传即触发”，推荐采用主动扫描 + 时间戳比对 + 条件跳过的组合策略——无需引入外部消息队列（如 RabbitMQ）或复杂文件系统事件库（如 watchdog），兼顾可靠性与部署简洁性。

核心思路

1. 约定上传目录：如 /opt/airflow/user_upload/，所有用户文件统一存放于此；
2. 获取文件元信息：使用 os.path.getmtime()（最后修改时间）或 os.path.getctime()（创建时间，Windows 下为创建时间，Unix 下常为状态变更时间）；
3. 判定“新文件”：将文件时间戳与当前 DAG 运行的时间窗口（推荐使用 data_interval_start）对比，仅处理时间晚于该窗口起始时间的文件；
4. 空数据优雅退出：若无匹配文件，抛出 AirflowSkipException 跳过后续任务，避免无效执行。

示例 DAG 实现

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.exceptions import AirflowSkipException
from airflow.utils.dates import days_ago
import os
import glob

UPLOAD_DIR = "/opt/airflow/user_upload"

def scan_and_trigger(**context):
    # 获取当前任务所属 DAG 的数据区间起始时间（推荐，语义清晰）
    data_start = context["data_interval_start"]

    # 扫描所有文件（支持通配符，如 *.csv, *.*）
    files = glob.glob(os.path.join(UPLOAD_DIR, "*"))
    new_files = []

    for f in files:
        if os.path.isfile(f):
            try:
                # 使用 mtime：更可靠，因多数上传操作会更新修改时间
                mtime = os.path.getmtime(f)
                if mtime >= data_start.timestamp():
                    new_files.append(f)
            except OSError:
                continue  # 忽略权限错误或已删除文件

    if not new_files:
        raise AirflowSkipException(f"No new files found in {UPLOAD_DIR} since {data_start}")

    # 将新文件列表传递给下游任务（通过 XCom）
    context["ti"].xcom_push(key="uploaded_files", value=new_files)
    print(f"Found {len(new_files)} new file(s): {new_files}")

def process_uploaded_file(**context):
    files = context["ti"].xcom_pull(key="uploaded_files", task_ids="scan_upload")
    for f in files:
        print(f"Processing: {f}")
        # 此处插入实际处理逻辑，如加载到数据库、触发转换等

with DAG(
    "trigger_on_any_upload",
    default_args={"start_date": days_ago(1)},
    schedule_interval="@hourly",  # 按需调整频率，如每5分钟："*/5 * * * *"
    catchup=False,
    tags=["file-trigger"],
) as dag:

    scan_task = PythonOperator(
        task_id="scan_upload",
        python_callable=scan_and_trigger,
        provide_context=True,
    )

    process_task = PythonOperator(
        task_id="process_files",
        python_callable=process_uploaded_file,
        provide_context=True,
    )

    scan_task >> process_task

关键注意事项

• ✅ 时间基准选择：优先使用 data_interval_start（DAG v2.2+ 推荐），而非 execution_date（已弃用）或 ts（含时区，需谨慎解析）；
• ✅ 文件时间可靠性：getmtime() 在绝大多数 FTP/SFTP/HTTP 上传场景下均被更新，比 getctime 更具可移植性；
• ⚠️ 并发安全：本方案不保证“恰好一次”处理（如文件在扫描间隙被重复上传）。若需严格幂等，建议配合文件移动（shutil.move 到 processed/）或数据库记录指纹（如 hashlib.md5(file.read())）；
• ⚠️ 权限与路径：确保 Airflow worker 进程对 UPLOAD_DIR 具有读取权限，且路径在所有 worker 节点上一致（若多节点部署，建议使用共享存储或 NFS）；
• ? 性能优化：对海量小文件场景，可增加 max_files_to_process 限制，或改用 pathlib.Path().iterdir() 配合 is_file() 提升可读性。

该方案以最小侵入性达成“任意文件上传即触发”的业务目标，兼具 Airflow 原生兼容性、运维可见性与调试友好性，是生产环境中值得优先验证的实用模式。

Chromox

Chromox是一款领先的AI在线生成平台，专为喜欢AI生成技术的爱好者制作的多种图像、视频生成方式的内容型工具平台。