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 要明确谁先谁后、失败怎么处理：
立即学习“Python免费学习笔记（深入）”；
- 用
>>或设置上下游（如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 2024-01-01模拟单次运行（不走调度器，纯本地执行） - 生产环境建议用 CeleryExecutor 或 KubernetesExecutor，避免默认的 SequentialExecutor 只能串行跑任务
- DAG 文件修改后，Scheduler 默认 30 秒扫描一次
dags/目录，无需重启服务（但语法错误会导致 DAG 显示为「paused」并报红）
基本上就这些。Airflow 强大但不复杂，关键在把逻辑切分成职责清晰的小任务，再用 DAG 连起来。写完一个能跑通的最小 DAG，后面扩展就顺了。
