编辑“︁Airflow API触发”︁

= Airflow API触发 =

== 介绍 ==  
'''Airflow API触发'''是指通过Airflow的REST API接口手动触发DAG（有向无环图）运行的技术。与依赖调度器自动触发不同，API触发提供了灵活的外部控制能力，适用于以下场景：  
* 外部系统（如CI/CD管道）需要按需启动工作流  
* 人工干预测试或紧急任务执行  
* 与其他服务集成实现事件驱动架构  

Airflow 2.0+ 提供了稳定的REST API（基于Flask AppBuilder），取代了旧版的实验性API。

== API端点概览 ==  
关键端点（需认证）：  

{| class="wikitable"  
! HTTP方法 !! 端点 !! 功能  
|-  
| POST || <code>/api/v1/dags/{dag_id}/dagRuns</code> || 触发指定DAG的新运行  
|-  
| GET || <code>/api/v1/dags/{dag_id}/dagRuns</code> || 列出DAG的所有运行记录  
|}  

== 认证方式 ==  
Airflow API支持多种认证：  
* Basic Auth（用户名/密码）  
* JWT（JSON Web Token）  
* OAuth（企业版）  

配置示例（<code>airflow.cfg</code>）：  
<syntaxhighlight lang="ini">  
[api]  
auth_backend = airflow.api.auth.backend.basic_auth  
</syntaxhighlight>  

== 触发DAG运行 ==  

=== 基本请求示例 ===  
使用<code>curl</code>触发DAG：  
<syntaxhighlight lang="bash">  
curl -X POST \  
  "http://localhost:8080/api/v1/dags/example_dag/dagRuns" \  
  -H "Content-Type: application/json" \  
  -H "Authorization: Basic $(echo -n 'admin:admin' | base64)" \  
  -d '{"conf": {"key": "value"}}'  
</syntaxhighlight>  

'''参数说明'''：  
* <code>conf</code>: 传递给DAG的配置字典（可选）  
* <code>logical_date</code>: 指定逻辑日期（ISO 8601格式，可选）  

=== Python客户端示例 ===  
使用官方Python客户端：  
<syntaxhighlight lang="python">  
from airflow.api.client.local_client import Client  

client = Client(api_base_url='http://localhost:8080')  
response = client.trigger_dag(  
    dag_id='example_dag',  
    run_id='manual__2023-01-01T00:00:00',  
    conf={'param': 'test'}  
)  
print(response)  
</syntaxhighlight>  

'''输出示例'''：  
<syntaxhighlight lang="json">  
{  
  "conf": {"param": "test"},  
  "dag_id": "example_dag",  
  "dag_run_id": "manual__2023-01-01T00:00:00",  
  "end_date": null,  
  "logical_date": "2023-01-01T00:00:00+00:00",  
  "state": "running"  
}  
</syntaxhighlight>  

== 状态检查 ==  
通过API检查DAG运行状态：  
<syntaxhighlight lang="bash">  
curl -X GET \  
  "http://localhost:8080/api/v1/dags/example_dag/dagRuns" \  
  -H "Authorization: Basic ..."  
</syntaxhighlight>  

响应包含<code>state</code>字段，可能值：  
* <code>queued</code>  
* <code>running</code>  
* <code>success</code>  
* <code>failed</code>  

== 实际案例 ==  

=== 场景：CI/CD管道集成 ===  
当GitHub代码库收到推送时，通过Webhook触发Airflow测试DAG：  

<mermaid>  
sequenceDiagram  
    participant GitHub  
    participant WebhookServer  
    participant AirflowAPI  
    GitHub->>WebhookServer: POST push event  
    WebhookServer->>AirflowAPI: POST /api/v1/dags/test_pipeline/dagRuns  
    AirflowAPI-->>WebhookServer: 202 Accepted  
    WebhookServer-->>GitHub: 200 OK  
</mermaid>  

=== 代码实现 ===  
Webhook处理器（Python Flask示例）：  
<syntaxhighlight lang="python">  
@app.route('/webhook', methods=['POST'])  
def handle_webhook():  
    if request.headers.get('X-GitHub-Event') == 'push':  
        response = requests.post(  
            'http://airflow-server/api/v1/dags/test_pipeline/dagRuns',  
            headers={'Authorization': 'Bearer YOUR_TOKEN'},  
            json={"conf": {"commit_id": request.json['after']}}  
        )  
        return response.json(), response.status_code  
</syntaxhighlight>  

== 高级配置 ==  

=== 自定义触发逻辑 ===  
在DAG文件中添加触发条件检查：  
<syntaxhighlight lang="python">  
from airflow.decorators import dag  

@dag(schedule_interval=None)  
def api_triggered_dag():  
    @task  
    def check_trigger_params(conf):  
        if 'required_key' not in conf:  
            raise ValueError("Missing required parameter")  

    check_trigger_params(conf="{{ dag_run.conf }}")  
</syntaxhighlight>  

=== 安全加固 ===  
1. 启用HTTPS  
2. 限制IP访问（Web服务器配置）  
3. 使用短期有效的JWT令牌  

== 常见问题 ==  

'''Q: 触发请求返回403错误'''  
A: 检查：  
1. 用户是否具有"DAG Edit"权限  
2. 认证凭据是否过期  

'''Q: DAG未按预期执行'''  
A: 检查：  
1. <code>is_paused_upon_creation</code>参数  
2. 调度器是否正常运行  

== 数学表示 ==  
API触发可视为函数映射：  

<math>  
\text{API\_Trigger}: (DAG\_ID \times Config) \rightarrow \text{DAG\_Run\_State}  
</math>  

其中：  
* <math>Config \subseteq \text{Key-Value Pairs}</math>  
* <math>\text{DAG\_Run\_State} \in \{\text{queued}, \text{running}, \text{success}, \text{failed}\}</math>  

== 总结 ==  
Airflow API触发提供了强大的外部集成能力，适合需要动态控制工作流的场景。通过合理设计认证机制和参数传递，可以实现安全的自动化触发流程。

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow调度与触发]]