Apache Airflow – 开源数据管道编排平台，ETL工作流管理利器-壹联网络

Apache Airflow是由Apache软件基金会管理的开源工作流编排平台，目前在GitHub上拥有超过4.5万颗星标，被全球数千家企业用于构建、调度和监控数据管道。作为数据工程领域的标准工具，Airflow采用Python编写，允许多节点分布式执行任务。

核心概念

Airflow有三个核心概念需要理解：

DAG（有向无环图）：定义任务及其依赖关系的Python文件
Operator（操作符）：代表单个任务，如执行Python代码、调用SQL、发送邮件
Task（任务）：Operator的具体实例

安装Airflow

pip install apache-airflow

# 快速启动（单节点）
airflow db init
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com

airflow webserver --port 8080 &
airflow scheduler &

也可以使用Docker快速部署：

pip install apache-airflow[standalone]
airflow standalone

创建第一个DAG

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

default_args = {
    'owner': 'airflow',
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

def extract():
    print("从数据源提取数据...")

def transform():
    print("数据转换处理...")

def load():
    print("加载到目标数据库...")

with DAG(
    'my_first_etl_dag',
    default_args=default_args,
    description='简单ETL示例',
    schedule_interval=timedelta(hours=1),
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    t1 = PythonOperator(task_id='extract', python_callable=extract)
    t2 = PythonOperator(task_id='transform', python_callable=transform)
    t3 = PythonOperator(task_id='load', python_callable=load)
    t1 >> t2 >> t3