编辑“︁Airflow元数据库访问”︁（章节）

= Airflow元数据库访问 =

== 介绍 ==  
'''Apache Airflow元数据库'''是存储工作流元数据的核心组件，包含DAG定义、任务实例状态、变量、连接等关键信息。通过直接访问元数据库，用户可以：  
* 监控/调试工作流  
* 批量修改配置  
* 实现自定义监控工具  
* 进行历史数据分析  

元数据库通常使用关系型数据库（如PostgreSQL、MySQL），其结构由Airflow自动管理。初学者应谨慎操作，高级用户可利用其实现深度定制。

== 元数据库结构 ==  
核心表及其关系：  

<mermaid>  
erDiagram  
    DAG ||--o{ DAG_RUN : "1:N"  
    DAG_RUN ||--o{ TASK_INSTANCE : "1:N"  
    TASK_INSTANCE }|--|| OPERATOR : "extends"  
    DAG {  
        string dag_id PK  
        string owners  
        timestamp last_updated  
    }  
    DAG_RUN {  
        string run_id PK  
        string dag_id FK  
        timestamp execution_date  
        string state  
    }  
    TASK_INSTANCE {  
        string task_id PK  
        string dag_id PK/FK  
        timestamp execution_date PK/FK  
        string state  
    }  
</mermaid>  

主要表说明：  
* '''dag'''：存储DAG定义和调度信息  
* '''dag_run'''：记录每次DAG执行的上下文  
* '''task_instance'''：保存任务实例状态与日志关联  

== 访问方式 ==  

=== 1. 通过Airflow CLI ===  
基础查询示例：  
<syntaxhighlight lang="bash">  
# 列出所有DAG  
airflow dags list  

# 查看特定DAG的运行记录  
airflow dags list-runs --dag-id example_dag  
</syntaxhighlight>  

=== 2. 使用SQLAlchemy（Python API） ===  
连接元数据库的完整示例：  

<syntaxhighlight lang="python">  
from airflow.settings import Session  
from airflow.models import DagRun, TaskInstance  

# 创建会话  
session = Session()  

# 查询最近失败的DAG运行  
failed_runs = (  
    session.query(DagRun)  
    .filter(DagRun.state == 'failed')  
    .order_by(DagRun.execution_date.desc())  
    .limit(10)  
    .all()  
)  

for run in failed_runs:  
    print(f"DAG ID: {run.dag_id}, Execution Date: {run.execution_date}")  

session.close()  
</syntaxhighlight>  

=== 3. 直接SQL查询 ===  
适用于需要复杂分析的场景：  
<syntaxhighlight lang="sql">  
-- 计算各DAG的平均任务执行时间  
SELECT  
    ti.dag_id,  
    AVG(TIMESTAMPDIFF(SECOND, ti.start_date, ti.end_date)) as avg_duration  
FROM task_instance ti  
WHERE ti.state = 'success'  
GROUP BY ti.dag_id;  
</syntaxhighlight>  

== 实际应用案例 ==  

=== 案例1：自定义监控仪表盘 ===  
通过定时查询元数据库，构建可视化报表：  
* 统计每日任务成功率  
* 识别长期运行的任务  
* 跟踪资源使用趋势  

=== 案例2：自动重试机制 ===  
当检测到特定错误模式时自动触发重试：  
<syntaxhighlight lang="python">  
from airflow.models import TaskInstance  

def auto_retry_failed_tasks(dag_id):  
    session = Session()  
    failed_tasks = (  
        session.query(TaskInstance)  
        .filter(TaskInstance.dag_id == dag_id)  
        .filter(TaskInstance.state == 'failed')  
        .all()  
    )  
    for task in failed_tasks:  
        task.state = 'scheduled'  # 将状态重置为待调度  
    session.commit()  
</syntaxhighlight>  

== 注意事项 ==  
* '''备份数据'''：修改前务必备份数据库  
* '''性能影响'''：复杂查询可能影响Airflow调度器性能  
* '''版本兼容性'''：不同Airflow版本的数据库结构可能有差异  
* '''权限控制'''：限制生产环境的直接数据库访问  

== 高级技巧 ==  
* 使用数据库视图简化常用查询  
* 实现自定义的'''BaseHook'''扩展元数据访问  
* 结合'''Event Scheduler'''监听数据库变更  

数学公式示例（计算任务延迟率）：  
<math>  
\text{延迟率} = \frac{\sum \text{实际开始时间} - \text{计划开始时间}}{\text{任务总数}} \times 100\%  
</math>  

通过系统学习元数据库访问，开发者可以解锁Airflow的深度监控和定制能力，但需始终遵循最小权限原则进行操作。

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow高级特性]]