编辑“︁Airflow代码风格指南”︁

= Airflow代码风格指南 =

== 介绍 ==  
Apache Airflow 是一个用于编排复杂工作流的开源工具，其核心思想是通过代码（Python）定义任务及其依赖关系。良好的代码风格不仅能提高可读性，还能减少错误并便于团队协作。本指南涵盖 Airflow DAG（有向无环图）和任务定义的最佳实践，包括命名规范、结构设计、错误处理等。

== 核心原则 ==  
1. **可读性优先**：代码应清晰表达业务逻辑，而非过度优化。  
2. **幂等性**：任务可重复执行且结果一致。  
3. **模块化**：将复杂逻辑拆分为可复用的组件。  

== 命名规范 ==  
* **DAG ID**：使用小写字母和下划线，如 `data_pipeline_daily`。  
* **任务 ID**：动词+名词描述操作，如 `extract_user_data`。  
* **变量名**：遵循 PEP 8，如 `default_args` 而非 `defaultArgs`。  

<syntaxhighlight lang="python">  
# 示例：规范的DAG定义  
from airflow import DAG  
from airflow.operators.python import PythonOperator  
from datetime import datetime  

def process_data():  
    print("Processing data...")  

with DAG(  
    dag_id="example_dag_style_guide",  
    start_date=datetime(2023, 1, 1),  
    schedule_interval="@daily",  
    tags=["example"],  
) as dag:  
    task_process = PythonOperator(  
        task_id="process_data_task",  
        python_callable=process_data,  
    )  
</syntaxhighlight>  

== 代码结构 ==  
=== DAG组织 ===  
* 每个DAG文件应只包含一个DAG定义。  
* 将任务逻辑拆分为单独的函数或模块。  

=== 任务依赖 ===  
使用 `>>` 或 `set_downstream` 明确依赖关系：  
<syntaxhighlight lang="python">  
extract_task >> transform_task >> load_task  # 推荐写法  
</syntaxhighlight>  

<mermaid>  
graph LR  
    A[extract] --> B[transform]  
    B --> C[load]  
</mermaid>  

== 错误处理与重试 ==  
* 设置 `retries` 和 `retry_delay`：  
<syntaxhighlight lang="python">  
default_args = {  
    "retries": 3,  
    "retry_delay": timedelta(minutes=5),  
}  
</syntaxhighlight>  

* 使用 `on_failure_callback` 通知异常：  
<syntaxhighlight lang="python">  
def alert_failure(context):  
    print(f"Task failed: {context.get('task_instance').task_id}")  

task = PythonOperator(  
    task_id="failable_task",  
    python_callable=risky_operation,  
    on_failure_callback=alert_failure,  
)  
</syntaxhighlight>  

== 性能优化 ==  
* **避免全局变量**：可能导致调度器性能下降。  
* **使用模板变量**：如 `{{ ds }}` 替代硬编码日期。  
* **任务超时**：设置 `execution_timeout` 防止卡死。  

== 实际案例 ==  
=== 数据管道示例 ===  
以下是一个ETL管道的简化实现：  
<syntaxhighlight lang="python">  
with DAG("etl_pipeline", schedule_interval="@hourly") as dag:  
    extract = PythonOperator(task_id="extract", python_callable=extract_api_data)  
    transform = PythonOperator(task_id="transform", python_callable=clean_data)  
    load = PythonOperator(task_id="load", python_callable=write_to_db)  

    extract >> transform >> load  
</syntaxhighlight>  

=== 动态任务生成 ===  
使用循环创建并行任务：  
<syntaxhighlight lang="python">  
for i in range(5):  
    task = PythonOperator(  
        task_id=f"process_file_{i}",  
        python_callable=process_file,  
        op_kwargs={"file_index": i},  
    )  
</syntaxhighlight>  

== 数学公式（可选） ==  
在计算任务优先级时，可使用加权公式：  
<math>  
priority = \frac{retries \times 2}{execution\_time}  
</math>  

== 总结 ==  
遵循Airflow代码风格指南能显著提升工作流的可维护性。关键点包括：  
# 一致的命名和结构  
# 明确的依赖关系  
# 健壮的错误处理  
# 性能优化意识

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow最佳实践]]