Airflow 工作流调度平台

FreeGuideOnline 15阅读 2026-07-09

python task1 >> task2 >> task3 task1 >> task4


这表示 task1 完成后同时触发 task2 和 task4，task2 完成后执行 task3。

### 2.4 Scheduler（调度器）与 Executor（执行器）

- **Scheduler**：核心组件，负责解析 DAG 文件，根据调度间隔和依赖关系决定何时触发任务，并把任务交给 Executor。
- **Executor**：实际执行任务的组件。常见的执行器：
  - `SequentialExecutor`：开发用，单线程，串行执行。
  - `LocalExecutor`：在本地以多进程方式并行执行。
  - `CeleryExecutor`：分布式工作队列，适合生产环境。
  - `KubernetesExecutor`：每个任务在独立的 Pod 中运行，动态资源分配。

### 2.5 Web Server（Web 服务器）

Airflow 提供了一个强大的 Web 界面，你可以：

- 查看所有 DAG 的运行状态图（树图、甘特图、图形视图）
- 手动触发 DAG 或任务
- 查看历史日志
- 管理变量、连接（Connections）
- 监控任务成败并设置告警

### 2.6 其他关键概念

- **DAG Run**：某一次 DAG 的具体执行实例。
- **Task Instance**：某一次 DAG Run 中某个 Task 的具体执行实例。
- **start_date**：DAG 的开始调度日期，注意这并非任务实际执行的时间，而是调度时间的基准。
- **execution_date**：逻辑执行时间，例如对于每天 2 点运行的 DAG，`执行日期 = 2025-03-20` 的数据任务会在 2025-03-21 02:00 触发。
- **catchup**：是否回填未执行的调度周期，默认 True。
- **Pool**（资源池）：限制某类任务的并行度。
- **Variables**：全局键值对变量，可在 DAG 中动态引用。
- **Connections**：存储外部系统连接信息（数据库、API、云服务等），实现了凭据与代码分离。

## 三、安装 Airflow（快速体验）

推荐使用官方提供的 `pip` 安装方式，也可以使用 Docker 进行标准化部署。下面演示在 Linux/macOS 上的快速搭建。

### 3.1 创建虚拟环境

```bash
python3 -m venv airflow_env
source airflow_env/bin/activate

3.2 安装 Airflow 并设置 Home 目录

AIRFLOW_VERSION=2.8.1
PYTHON_VERSION="$(python3 --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

3.3 初始化数据库

Airflow 默认使用 SQLite（仅用于测试），生产环境需切换到 PostgreSQL 或 MySQL。

airflow db init

3.4 创建管理员用户

airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email [email protected]

3.5 启动所有服务

打开三个终端窗口，分别执行：

# 启动 Web 服务器
airflow webserver -p 8080

# 启动调度器
airflow scheduler

# 如果需要，启动触发器服务（用于可延迟算子）
airflow triggerer

访问 http://localhost:8080，使用刚刚创建的用户登录，就可以看到 Airflow 界面了。

四、编写你的第一个 DAG

在 Airflow 的 dags 文件夹（默认路径为 ~/airflow/dags）下创建一个 Python 文件，例如 my_first_dag.py。

4.1 示例：每天提取、转换、加载数据并发送通知

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.operators.email import EmailOperator

# 默认参数，应用给该 DAG 中的所有任务
default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# 定义 DAG
with DAG(
    dag_id='etl_example_dag',
    default_args=default_args,
    description='一个简单的 ETL 工作流',
    schedule_interval='@daily',          # 每天运行一次
    start_date=datetime(2025, 1, 1),
    catchup=False,
    tags=['tutorial'],
) as dag:

    # 任务1：检查数据源文件是否存在
    check_file = BashOperator(
        task_id='check_source_file',
        bash_command='test -f /data/source.csv && echo "文件存在" || exit 1'
    )

    # 任务2：用 Python 清洗数据
    def clean_data():
        # 假想的清洗逻辑
        print("数据清洗完成")
    
    clean = PythonOperator(
        task_id='clean_data',
        python_callable=clean_data
    )

    # 任务3：加载到数据库（用 bash 模拟）
    load = BashOperator(
        task_id='load_to_db',
        bash_command='echo "模拟数据加载到数据库"'
    )

    # 任务4：发送成功邮件
    notify = EmailOperator(
        task_id='send_email',
        to='[email protected]',
        subject='ETL 完成',
        html_content='<h3>每日 ETL 成功执行</h3>'
    )

    # 定义依赖：check >> clean >> load >> notify
    check_file >> clean >> load >> notify

保存文件后，稍等片刻调度器便会扫描并加载这个 DAG（可通过 airflow dags list 或 Web UI 查看）。在 Web 界面中你可以手动触发执行，观察每个任务的颜色变化：浅绿表示成功，红色表示失败。

五、深入理解 DAG 调度与时间窗口

5.1 schedule_interval 参数

支持 cron 表达式或预设别名：

  • None：手动触发
  • @once：只执行一次
  • @hourly：每小时
  • @daily：每天 0:00
  • @weekly：每周日 0:00
  • 0 6-18 * * 1-5：工作日早 6 点到晚 18 点整点

5.2 回填与 catchup

假设 start_date = 2025-01-01，schedule_interval = @daily，而当前是 2025-03-20。如果 catchup=True，Airflow 会为从 1 月 1 日到今天所有未执行的调度周期创建 DAG Run 并依次执行。这通常用于历史数据回填，但如果你不希望执行过去的任务，一定要设置 catchup=False。

5.3 执行日期 (execution_date) 的含义

关键思维转变：对于调度间隔为 @daily 的 DAG，2025-03-20 的 dag run 的实际触发时间是在 2025-03-21 00:00（该周期的结束时间），而逻辑上的 execution_date 是 2025-03-20。这样做是为了覆盖从 2025-03-20 00:00 至 2025-03-21 00:00 这一整天的数据。在你的 Operator 代码中，可以通过 {{ ds }} 或 {{ execution_date }} 获取这个日期，通常用于处理分区数据。

六、生产环境最佳实践

6.1 使用官方 Docker 镜像快速部署

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.8.1/docker-compose.yaml'
docker-compose up -d

该 compose 文件包含 PostgreSQL 数据库、Redis 中间件、Celery worker 等，适合快速搭建生产级环境。

6.2 将凭据外部化：Connections 与 Variables

不要在 DAG 代码中硬编码数据库密码、API 密钥。应使用 Web UI 中的 Admin → Connections 添加连接，然后在 DAG 中通过连接 ID 引用：

from airflow.providers.postgres.operators.postgres import PostgresOperator

run_query = PostgresOperator(
    task_id='query_pg',
    postgres_conn_id='my_postgres_conn',
    sql='SELECT count(*) FROM users;'
)

6.3 使用 TaskGroup 组织复杂 DAG

当任务数量变多时，可以用 TaskGroup 进行分组，使 Web 界面更清晰。

from airflow.utils.task_group import TaskGroup

with TaskGroup("processing_section") as processing:
    task_a = PythonOperator(...)
    task_b = PythonOperator(...)
    task_a >> task_b

6.4 动态 DAG 生成

利用 Python 循环和模板引擎生成大量类似 DAG，避免代码重复。例如为每个网域生成一个监控 DAG：

domains = ['news.example.com', 'shop.example.com']
for domain in domains:
    dag_id = f'monitor_{domain.replace(".", "_")}'
    # 动态创建 DAG