airflow

Apache Airflow 调度平台

补充说明

Airflow 用 Python DAG 描述任务依赖，适合批处理、ETL 和定时调度。核心组件包括 scheduler、webserver、worker、metadata database 和 executor。生产环境通常使用 CeleryExecutor 或 KubernetesExecutor。

常用示例

本地快速启动：

airflow standalone
# 访问 http://localhost:8080

查看 DAG 和任务：

airflow dags list
airflow tasks list example_dag
airflow dags trigger example_dag
airflow dags state example_dag 2026-10-18

测试单个任务：

airflow tasks test example_dag extract 2026-10-18

回填历史数据：

airflow dags backfill example_dag \
  --start-date 2026-10-01 \
  --end-date 2026-10-07

清理元数据：

airflow db clean --clean-before-timestamp '2026-01-01'

DAG 示例：

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG('demo', start_date=datetime(2026, 1, 1),
         schedule='0 2 * * *', catchup=False) as dag:
    extract = BashOperator(task_id='extract', bash_command='echo extract')
    load = BashOperator(task_id='load', bash_command='echo load')
    extract >> load

易错点与排错

• DAG 文件有语法错误，整文件不会出现在 UI，先看 scheduler 日志。
• start_date 和 schedule 配错，任务没有按预期触发。
• 时区不一致，调度时间偏移；统一使用 UTC 或明确时区。
• 元数据库连接池耗尽，scheduler 心跳丢失。
• Worker 没有安装 DAG 依赖，任务报 ModuleNotFoundError。
• 任务重复运行没有幂等设计，回填时产生重复数据。
• 日志目录磁盘满，任务状态无法更新。
• 直接修改已运行的 DAG 结构，历史任务实例状态混乱。

读完自测

合上页面，先回答，再展开对照。

1. 新 DAG 在 UI 里完全看不到，先查什么？
2. 任务报 ModuleNotFoundError，原因通常在哪？
3. 回填历史数据前最该确认什么？