Python使用Airflow实现自动化任务调度的构建方式【教程】

Airflow通过DAG文件定义任务调度，需满足文件命名、全局dag变量、必要导入等要求；用PythonOperator封装函数，设置依赖、重试、触发规则及敏感参数管理；支持本地调试与生产部署。

用 Airflow 实现 Python 任务的自动化调度，核心是把业务逻辑封装成可被 Airflow 管理的 Operator，再通过 DAG 定义执行顺序、触发条件和重试策略。它不直接运行脚本，而是调度“任务实例”，靠 Scheduler 和 Executor 协同驱动。

定义一个基础 DAG 文件

DAG 是 Airflow 的调度蓝图，本质是一个 Python 文件（通常放在 dags/ 目录下），需满足几个硬性要求：

• 文件名不能含空格或特殊字符，推荐小写加下划线（如 etl_daily_job.py）
• 必须包含一个全局变量 dag = DAG(...)，且变量名固定为 dag
• 需导入必要模块：from airflow import DAG、from airflow.operators.python import PythonOperator 等
• DAG 参数中 schedule_interval（新版推荐用 schedule）决定触发频率，支持 cron 表达式（如 "0 2 * * *" 表示每天凌晨2点）或 timedelta（如 timedelta(days=1)）

用 PythonOperator 封装你的函数

这是最常用的方式，适合已有现成的 Python 函数。Airflow 会在任务运行时调用它，并自动传入上下文（**context）：

• 函数本身不能带括号调用，只写函数名（例如 task1 = PythonOperator(task_id='run_clean', python_callable=clean_data)）
• 若需传参，用 op_kwargs 字典（如 op_kwargs={"table": "users", "days_back": 7}），函数签名要匹配
• 函数返回值默认被序列化进 XCom，供下游任务读取（用 context["ti"].xcom_pull(task_ids="upstream_task")）
• 避免在函数里写长时间阻塞操作（如 time.sleep(300)），应拆成多个短任务或改用 TimeDeltaSensor 等传感器

设置依赖关系与容错机制

任务不是孤立运行的，DAG 要明确谁先谁后、失败怎么处理：

• 用 >> 或 设置上下游（如 extract >> transform >> load），也可用 set_downstream() 方法
• retries=3 指定失败后重试次数；retry_delay=timedelta(minutes=5) 控制重试间隔
• trigger_rule="all_success"（默认）表示所有上游成功才运行；换成 "all_done" 可让任务无论上游成败都执行（适合清理类任务）
• 敏感参数（如数据库密码）不要硬编码，用 Airflow Connections 或 Variables 管理，代码里通过 BaseHook.get_connection("my_db").password 获取

本地调试与部署要点

别等部署到生产才发现问题：

• 启动 Webserver 和 Scheduler 后，在 UI 中点击 DAG 名称右侧的「Trigger DAG」手动测试一次
• 用命令行快速验证语法：airflow dags list 看是否识别；airflow tasks list my_dag_id 查任务；airflow tasks test my_dag_id task_name 2025-01-01 模拟单次运行（不走调度器，纯本地执行）
• 生产环境建议用 CeleryExecutor 或 KubernetesExecutor，避免默认的 SequentialExecutor 只能串行跑任务
• DAG 文件修改后，Scheduler 默认 30 秒扫描一次 dags/ 目录，无需重启服务（但语法错误会导致 DAG 显示为「paused」并报红）

基本上就这些。Airflow 强大但不复杂，关键在把逻辑切分成职责清晰的小任务，再用 DAG 连起来。写完一个能跑通的最小 DAG，后面扩展就顺了。