编辑“︁Airflow ETL流程设计”︁（章节）

= Airflow ETL流程设计 =

'''ETL（Extract, Transform, Load）'''是数据集成中的核心流程，而Apache Airflow作为一个强大的工作流编排工具，能够高效地设计和调度ETL任务。本章将详细介绍如何使用Airflow设计和实现ETL流程，涵盖基础概念、核心组件、代码示例及实际应用场景。

== 1. 概述 ==

ETL流程是指从源系统'''提取（Extract）'''数据，经过'''转换（Transform）'''处理后，'''加载（Load）'''到目标系统的过程。Airflow通过DAG（有向无环图）定义任务依赖关系，并利用Operator执行具体操作，使得ETL流程可调度、可监控。

=== 核心优势 ===
* '''可视化依赖管理'''：通过DAG清晰展示任务执行顺序。
* '''错误处理与重试'''：内置任务失败重试机制。
* '''可扩展性'''：支持自定义Operator以满足特定需求。

== 2. ETL流程设计步骤 ==

=== 2.1 定义DAG ===
在Airflow中，DAG是ETL流程的容器，需指定调度间隔、起始时间等参数。

<syntaxhighlight lang="python">
from airflow import DAG
from datetime import datetime, timedelta

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

dag = DAG(
    'etl_pipeline',
    default_args=default_args,
    description='A simple ETL pipeline',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2023, 1, 1),
    catchup=False,
)
</syntaxhighlight>

=== 2.2 提取（Extract）===
使用Operator从数据库、API或文件中提取数据。例如，用`PythonOperator`调用提取函数：

<syntaxhighlight lang="python">
from airflow.operators.python import PythonOperator

def extract_data():
    # 模拟从CSV提取数据
    import pandas as pd
    data = pd.read_csv('/data/source.csv')
    return data.to_json()

extract_task = PythonOperator(
    task_id='extract',
    python_callable=extract_data,
    dag=dag,
)
</syntaxhighlight>

=== 2.3 转换（Transform）===
对提取的数据进行清洗、聚合等操作。以下示例使用`Pandas`进行转换：

<syntaxhighlight lang="python">
def transform_data(**kwargs):
    ti = kwargs['ti']
    json_data = ti.xcom_pull(task_ids='extract')
    data = pd.read_json(json_data)
    
    # 转换逻辑：过滤无效值并计算平均值
    data = data.dropna()
    data['value'] = data['value'].mean()
    return data.to_json()

transform_task = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    provide_context=True,
    dag=dag,
)
</syntaxhighlight>

=== 2.4 加载（Load）===
将处理后的数据写入目标系统（如数据库、数据仓库）：

<syntaxhighlight lang="python">
def load_data(**kwargs):
    ti = kwargs['ti']
    json_data = ti.xcom_pull(task_ids='transform')
    data = pd.read_json(json_data)
    
    # 写入PostgreSQL
    from sqlalchemy import create_engine
    engine = create_engine('postgresql://user:password@localhost:5432/mydb')
    data.to_sql('results', engine, if_exists='replace')

load_task = PythonOperator(
    task_id='load',
    python_callable=load_data,
    provide_context=True,
    dag=dag,
)
</syntaxhighlight>

=== 2.5 设置依赖关系 ===
通过`>>`符号定义任务执行顺序：

<syntaxhighlight lang="python">
extract_task >> transform_task >> load_task
</syntaxhighlight>

== 3. 实际案例：销售数据ETL ==

=== 场景描述 ===
每日从电商平台API提取销售数据，计算每类商品销售额，并加载到分析数据库。

=== DAG设计 ===
<mermaid>
graph LR
    A[Extract: API调用] --> B[Transform: 按类别聚合]
    B --> C[Load: 写入PostgreSQL]
</mermaid>

=== 关键代码 ===
<syntaxhighlight lang="python">
# 提取任务（伪代码）
def extract_sales():
    response = requests.get('https://api.ecommerce.com/sales')
    return response.json()

# 转换任务
def transform_sales(**kwargs):
    sales_data = kwargs['ti'].xcom_pull(task_ids='extract_sales')
    df = pd.DataFrame(sales_data)
    return df.groupby('category')['revenue'].sum().to_json()

# 加载任务
def load_sales(**kwargs):
    revenue_by_category = pd.read_json(kwargs['ti'].xcom_pull(task_ids='transform_sales'))
    revenue_by_category.to_sql('daily_sales', engine, if_exists='append')
</syntaxhighlight>

== 4. 高级优化 ==

=== 4.1 动态任务生成 ===
使用`TaskGroup`或循环生成并行任务：

<syntaxhighlight lang="python">
from airflow.utils.task_group import TaskGroup

with TaskGroup('dynamic_tasks', dag=dag) as tg:
    for table in ['users', 'products', 'orders']:
        extract_task = PythonOperator(
            task_id=f'extract_{table}',
            python_callable=lambda: extract_from_table(table),
        )
</syntaxhighlight>

=== 4.2 增量ETL ===
通过`execution_date`实现增量加载：

<syntaxhighlight lang="python">
def incremental_load(**kwargs):
    date = kwargs['execution_date']
    query = f"SELECT * FROM sales WHERE date > '{date}'"
    # 执行增量加载...
</syntaxhighlight>

== 5. 常见问题与调试 ==

* '''XCom数据传输限制'''：避免传递大型数据（>48KB），建议使用外部存储（如S3）。
* '''任务并行度控制'''：通过`pool`参数限制并发任务数。
* '''依赖冲突'''：使用`ExternalTaskSensor`等待外部DAG完成。

== 总结 ==
Airflow为ETL流程提供了灵活、可靠的编排能力。通过合理设计DAG、选择适当的Operator，并结合数据处理的业务逻辑，可以构建高效的数据管道。初学者应从简单案例入手，逐步掌握任务依赖、错误处理和性能优化技巧。

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow数据集成]]