一、前言：为什么需要工作流调度？

还记得你刚入行时是怎么跑定时任务的吗？

# 青涩的你：crontab -e
0 2 * * * /home/scripts/etl_step1.sh
30 2 * * * /home/scripts/etl_step2.sh  # 假设step1跑30分钟
0 3 * * * /home/scripts/etl_step3.sh   # 假设step2跑30分钟

# 某天凌晨3点，你被电话叫醒
# "报表数据不对！"
# 原来step1跑了40分钟，step2还没跑完step3就开始了...

Crontab 的致命问题：

工作流调度器的价值：

二、Airflow 是什么？

2.1 一句话定义

Airflow = 以 Python 代码定义工作流（DAG）的分布式任务调度平台

2.2 发展历程

2.3 为什么选择 Airflow？

三、核心概念详解

3.1 概念全景图

3.2 DAG：有向无环图

DAG 的基本定义：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

# DAG 默认参数
default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'email': ['alert@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

# 定义 DAG
with DAG(
    dag_id='etl_daily_pipeline',
    default_args=default_args,
    description='每日ETL数据管道',
    schedule_interval='0 2 * * *',  # 每天凌晨2点
    start_date=datetime(2024, 1, 1),
    catchup=False,  # 不补跑历史
    tags=['etl', 'daily'],
) as dag:
    
    # 定义任务...
    pass

3.3 Operator：任务类型

常用 Operator 示例：

from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.providers.mysql.operators.mysql import MySqlOperator
from airflow.providers.http.operators.http import SimpleHttpOperator

# 1. BashOperator - 执行Shell命令
extract_task = BashOperator(
    task_id='extract_data',
    bash_command='python /scripts/extract.py --date {{ ds }}',
)

# 2. PythonOperator - 执行Python函数
def transform_data(**context):
    execution_date = context['ds']
    # 数据转换逻辑
    return f"Transformed data for {execution_date}"

transform_task = PythonOperator(
    task_id='transform_data',
    python_callable=transform_data,
    provide_context=True,
)

# 3. MySqlOperator - 执行SQL
load_task = MySqlOperator(
    task_id='load_to_mysql',
    mysql_conn_id='mysql_warehouse',
    sql="""
        INSERT INTO daily_stats (date, metric, value)
        SELECT '{{ ds }}', metric, SUM(value)
        FROM raw_data
        WHERE date = '{{ ds }}'
        GROUP BY metric
    """,
)

# 4. HttpOperator - 调用API
notify_task = SimpleHttpOperator(
    task_id='notify_api',
    http_conn_id='notification_api',
    endpoint='/webhook/airflow',
    method='POST',
    data='{"status": "completed", "date": "{{ ds }}"}',
)

3.4 Sensor：等待条件满足

from airflow.sensors.filesystem import FileSensor
from airflow.providers.http.sensors.http import HttpSensor
from airflow.sensors.sql import SqlSensor

# 1. 文件Sensor - 等待文件出现
wait_for_file = FileSensor(
    task_id='wait_for_data_file',
    filepath='/data/input/{{ ds }}/data.csv',
    poke_interval=60,  # 每60秒检查一次
    timeout=3600,      # 最多等待1小时
    mode='poke',       # poke模式占用worker槽位
)

# 2. SQL Sensor - 等待数据就绪
wait_for_data = SqlSensor(
    task_id='wait_for_upstream_data',
    conn_id='mysql_source',
    sql="""
        SELECT COUNT(*) FROM source_table 
        WHERE date = '{{ ds }}' AND status = 'ready'
    """,
    poke_interval=300,
    timeout=7200,
    mode='reschedule',  # reschedule模式释放槽位
)

# 3. HTTP Sensor - 等待API返回成功
wait_for_api = HttpSensor(
    task_id='wait_for_upstream_api',
    http_conn_id='upstream_service',
    endpoint='/status/{{ ds }}',
    response_check=lambda response: response.json()['status'] == 'ready',
    poke_interval=120,
    timeout=3600,
)

3.5 XCom：任务间通信

# 方式1：通过return值自动push
def extract_data(**context):
    data = {'count': 1000, 'status': 'success'}
    return data  # 自动push到XCom

def transform_data(**context):
    # 从上游任务获取XCom
    ti = context['ti']
    upstream_data = ti.xcom_pull(task_ids='extract_data')
    print(f"收到数据: {upstream_data}")  # {'count': 1000, 'status': 'success'}

# 方式2：手动push/pull
def manual_push(**context):
    ti = context['ti']
    ti.xcom_push(key='my_key', value='my_value')

def manual_pull(**context):
    ti = context['ti']
    value = ti.xcom_pull(task_ids='manual_push', key='my_key')

# 方式3：Jinja模板中使用
bash_task = BashOperator(
    task_id='use_xcom',
    bash_command='echo "Count: {{ ti.xcom_pull(task_ids="extract_data")["count"] }}"',
)

四、Airflow 架构详解

4.1 整体架构

4.2 核心组件职责

4.3 Executor 类型

4.4 任务生命周期

五、实战：构建完整的 ETL Pipeline

5.1 项目结构

airflow_project/
├── dags/
│   ├── __init__.py
│   ├── etl_daily_pipeline.py      # DAG定义
│   └── common/
│       ├── __init__.py
│       └── utils.py               # 公共函数
├── plugins/
│   └── custom_operators/          # 自定义Operator
├── tests/
│   └── test_dags.py              # DAG测试
├── docker-compose.yaml
└── requirements.txt

5.2 完整 DAG 示例

"""
ETL Daily Pipeline
每日从MySQL抽取数据，转换后加载到数据仓库
"""
from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.providers.mysql.operators.mysql import MySqlOperator
from airflow.sensors.sql import SqlSensor
from airflow.utils.task_group import TaskGroup
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)

# ============ 默认参数 ============
default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'email': ['data-alerts@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'execution_timeout': timedelta(hours=2),
}

# ============ Python函数 ============
def check_data_quality(**context):
    """数据质量检查"""
    ds = context['ds']
    # 这里执行数据质量检查逻辑
    record_count = 1000  # 模拟获取记录数
    
    if record_count < 100:
        raise ValueError(f"数据量异常: {record_count} < 100")
    
    logger.info(f"数据质量检查通过，记录数: {record_count}")
    return {'record_count': record_count, 'status': 'passed'}

def decide_branch(**context):
    """决定执行哪个分支"""
    ti = context['ti']
    qc_result = ti.xcom_pull(task_ids='quality_check')
    
    if qc_result['record_count'] > 10000:
        return 'process_large_data'
    else:
        return 'process_small_data'

def process_data(data_size: str, **context):
    """处理数据"""
    ds = context['ds']
    logger.info(f"处理{data_size}数据集，日期: {ds}")
    # 实际的数据处理逻辑
    return f"Processed {data_size} data for {ds}"

def send_notification(**context):
    """发送通知"""
    ti = context['ti']
    ds = context['ds']
    # 获取所有上游任务的结果
    logger.info(f"发送完成通知，日期: {ds}")

# ============ DAG定义 ============
with DAG(
    dag_id='etl_daily_pipeline_v2',
    default_args=default_args,
    description='生产级ETL数据管道示例',
    schedule_interval='0 2 * * *',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['etl', 'production', 'daily'],
    doc_md="""
    ## ETL Daily Pipeline
    
    ### 功能说明
    1. 等待上游数据就绪
    2. 抽取MySQL数据
    3. 数据质量检查
    4. 根据数据量选择处理策略
    5. 加载到数据仓库
    6. 发送通知
    
    ### 依赖
    - MySQL源数据库
    - 数据仓库
    - 告警服务
    """,
) as dag:
    
    # ===== 开始节点 =====
    start = EmptyOperator(task_id='start')
    
    # ===== 等待上游数据 =====
    wait_for_source = SqlSensor(
        task_id='wait_for_source_data',
        conn_id='mysql_source',
        sql="""
            SELECT COUNT(*) > 0 
            FROM etl_status 
            WHERE date = '{{ ds }}' AND status = 'ready'
        """,
        poke_interval=300,
        timeout=7200,
        mode='reschedule',
    )
    
    # ===== 抽取数据（任务组） =====
    with TaskGroup(group_id='extract_group') as extract_group:
        extract_users = BashOperator(
            task_id='extract_users',
            bash_command='python /scripts/extract.py --table users --date {{ ds }}',
        )
        
        extract_orders = BashOperator(
            task_id='extract_orders',
            bash_command='python /scripts/extract.py --table orders --date {{ ds }}',
        )
        
        extract_products = BashOperator(
            task_id='extract_products',
            bash_command='python /scripts/extract.py --table products --date {{ ds }}',
        )
        
        # 组内并行执行
        [extract_users, extract_orders, extract_products]
    
    # ===== 数据质量检查 =====
    quality_check = PythonOperator(
        task_id='quality_check',
        python_callable=check_data_quality,
    )
    
    # ===== 分支决策 =====
    branch = BranchPythonOperator(
        task_id='branch_by_data_size',
        python_callable=decide_branch,
    )
    
    # ===== 处理分支 =====
    process_large = PythonOperator(
        task_id='process_large_data',
        python_callable=process_data,
        op_kwargs={'data_size': 'large'},
    )
    
    process_small = PythonOperator(
        task_id='process_small_data',
        python_callable=process_data,
        op_kwargs={'data_size': 'small'},
    )
    
    # ===== 合并分支 =====
    merge = EmptyOperator(
        task_id='merge_branches',
        trigger_rule='none_failed_min_one_success',  # 至少一个成功即可
    )
    
    # ===== 加载到数仓 =====
    with TaskGroup(group_id='load_group') as load_group:
        load_to_dw = MySqlOperator(
            task_id='load_to_warehouse',
            mysql_conn_id='mysql_warehouse',
            sql='/sql/load_warehouse.sql',
        )
        
        update_stats = MySqlOperator(
            task_id='update_statistics',
            mysql_conn_id='mysql_warehouse',
            sql="""
                INSERT INTO etl_statistics (date, table_name, row_count, updated_at)
                VALUES ('{{ ds }}', 'daily_summary', 
                        (SELECT COUNT(*) FROM daily_summary WHERE date = '{{ ds }}'),
                        NOW())
                ON DUPLICATE KEY UPDATE 
                    row_count = VALUES(row_count),
                    updated_at = NOW()
            """,
        )
        
        load_to_dw >> update_stats
    
    # ===== 发送通知 =====
    notify = PythonOperator(
        task_id='send_notification',
        python_callable=send_notification,
        trigger_rule='all_done',  # 无论成功失败都执行
    )
    
    # ===== 结束节点 =====
    end = EmptyOperator(task_id='end')
    
    # ===== 定义依赖关系 =====
    start >> wait_for_source >> extract_group >> quality_check >> branch
    branch >> [process_large, process_small] >> merge
    merge >> load_group >> notify >> end

5.3 DAG 可视化

六、高级特性

6.1 动态 DAG 生成

"""
根据配置动态生成多个相似的DAG
"""
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

# 配置：每个数据源一个DAG
data_sources = [
    {'name': 'mysql_db1', 'schedule': '0 1 * * *', 'conn_id': 'mysql_db1'},
    {'name': 'mysql_db2', 'schedule': '0 2 * * *', 'conn_id': 'mysql_db2'},
    {'name': 'postgres_db1', 'schedule': '0 3 * * *', 'conn_id': 'pg_db1'},
]

def create_dag(source_config):
    """工厂函数：创建DAG"""
    dag_id = f"sync_{source_config['name']}"
    
    dag = DAG(
        dag_id=dag_id,
        schedule_interval=source_config['schedule'],
        start_date=datetime(2024, 1, 1),
        catchup=False,
        tags=['dynamic', 'sync'],
    )
    
    def sync_data(conn_id, **context):
        print(f"Syncing data from {conn_id}")
    
    with dag:
        sync_task = PythonOperator(
            task_id='sync_data',
            python_callable=sync_data,
            op_kwargs={'conn_id': source_config['conn_id']},
        )
    
    return dag

# 动态创建DAG并注册到全局命名空间
for source in data_sources:
    dag_id = f"sync_{source['name']}"
    globals()[dag_id] = create_dag(source)

6.2 TaskFlow API (Airflow 2.0+)

"""
使用TaskFlow API简化DAG编写
更Pythonic，更简洁
"""
from airflow.decorators import dag, task
from datetime import datetime

@dag(
    dag_id='taskflow_example',
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['taskflow'],
)
def taskflow_pipeline():
    """使用装饰器定义DAG"""
    
    @task
    def extract() -> dict:
        """抽取数据"""
        return {'data': [1, 2, 3, 4, 5], 'count': 5}
    
    @task
    def transform(raw_data: dict) -> dict:
        """转换数据"""
        data = raw_data['data']
        transformed = [x * 2 for x in data]
        return {'data': transformed, 'count': len(transformed)}
    
    @task
    def load(transformed_data: dict) -> str:
        """加载数据"""
        print(f"Loading {transformed_data['count']} records")
        return 'success'
    
    @task
    def notify(status: str):
        """发送通知"""
        print(f"Pipeline finished with status: {status}")
    
    # 自动推断依赖关系！
    raw = extract()
    transformed = transform(raw)
    status = load(transformed)
    notify(status)

# 实例化DAG
taskflow_dag = taskflow_pipeline()

6.3 Deferrable Operators (Airflow 2.2+)

from airflow.sensors.filesystem import FileSensor

# 传统模式：占用Worker槽位
file_sensor_poke = FileSensor(
    task_id='wait_file_poke',
    filepath='/data/file.csv',
    mode='poke',  # 持续占用
)

# Reschedule模式：定期释放
file_sensor_reschedule = FileSensor(
    task_id='wait_file_reschedule',
    filepath='/data/file.csv',
    mode='reschedule',  # 每次检查后释放
)

# Deferrable模式：最高效（需要Triggerer组件）
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensorAsync

s3_sensor_async = S3KeySensorAsync(
    task_id='wait_s3_file',
    bucket_key='s3://bucket/path/file.csv',
    deferrable=True,  # 异步等待
)

6.4 数据感知调度 (Data-aware Scheduling)

"""
Airflow 2.4+ Dataset感知调度
上游DAG产出数据集，下游DAG自动触发
"""
from airflow import DAG, Dataset
from airflow.operators.python import PythonOperator
from datetime import datetime

# 定义数据集
orders_dataset = Dataset('s3://bucket/orders/')
users_dataset = Dataset('s3://bucket/users/')

# 生产者DAG
with DAG(
    dag_id='producer_orders',
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
) as producer_dag:
    
    @task(outlets=[orders_dataset])  # 声明产出数据集
    def produce_orders():
        # 生成订单数据
        pass

# 消费者DAG - 当orders和users都更新时自动触发
with DAG(
    dag_id='consumer_report',
    schedule=[orders_dataset, users_dataset],  # 数据集触发
    start_date=datetime(2024, 1, 1),
) as consumer_dag:
    
    @task
    def generate_report():
        # 生成报表
        pass

七、运维与监控

7.1 部署架构（生产环境）

7.2 关键监控指标

7.3 常见问题排查

7.4 配置优化

# airflow.cfg 关键配置

[core]
# DAG解析间隔（秒）
dag_dir_list_interval = 300
# 最大活跃DAG运行数
max_active_runs_per_dag = 16
# 并行任务数
parallelism = 32

[scheduler]
# Scheduler循环间隔
scheduler_heartbeat_sec = 5
# 每次调度的最大DAG数
max_dagruns_to_create_per_loop = 10
# 每次解析的最大DAG文件数
max_dagruns_per_loop_to_schedule = 20

[celery]
# Worker并发数
worker_concurrency = 16

[webserver]
# Web页面刷新间隔
default_dag_run_display_number = 25

八、最佳实践

8.1 DAG 编写规范

8.2 常见反模式

# ❌ 反模式1：在DAG顶层做重计算
import pandas as pd
df = pd.read_csv('large_file.csv')  # 每次解析DAG都会执行！

# ✅ 正确做法：放在任务函数内
def process_data():
    df = pd.read_csv('large_file.csv')
    # ...

# ❌ 反模式2：硬编码日期
sql = "SELECT * FROM table WHERE date = '2024-01-15'"

# ✅ 正确做法：使用模板变量
sql = "SELECT * FROM table WHERE date = '{{ ds }}'"

# ❌ 反模式3：过度使用XCom传递大数据
def return_large_data():
    return pd.read_csv('huge_file.csv').to_dict()  # 可能几百MB

# ✅ 正确做法：XCom只传递引用
def return_data_path():
    path = '/tmp/processed_data.parquet'
    df.to_parquet(path)
    return path  # 只传路径

# ❌ 反模式4：不设置超时
task = BashOperator(
    task_id='might_hang',
    bash_command='python long_running.py',
    # 没有execution_timeout，可能永远挂住
)

# ✅ 正确做法：总是设置超时
task = BashOperator(
    task_id='with_timeout',
    bash_command='python long_running.py',
    execution_timeout=timedelta(hours=2),
)

九、总结

9.1 Airflow 核心价值

9.2 适用场景

9.3 一句话总结

Airflow = Python 定义工作流 + 强大的调度 + 丰富的生态 + 完善的监控

如果你的团队懂 Python，需要管理复杂的数据管道，Airflow 是最佳选择之一。

如果这篇文章对你有帮助，欢迎：

• 👍 点赞：让更多人看到
• ⭐ 收藏：方便日后查阅
• 💬 评论：分享你的 Airflow 使用经验