编辑“︁Airflow HttpOperator 详解”︁

= Airflow HttpOperator 详解 =

== 简介 ==
'''HttpOperator''' 是 Apache Airflow 中用于执行 HTTP 请求的核心 Operator，属于 `airflow.providers.http` 包。它允许工作流通过 REST API 与外部服务交互，支持 GET/POST/PUT/DELETE 等标准 HTTP 方法，是数据管道集成 Web 服务的桥梁。

== 核心特性 ==
* '''HTTP 方法支持'''：完整覆盖 RESTful 操作
* '''认证集成'''：支持 Basic Auth/OAuth2 等
* '''请求定制'''：可配置 Headers/Params/Body
* '''响应处理'''：支持 JSON/XML 解析和 XCom 推送
* '''错误处理'''：HTTP 状态码检查和重试机制

== 基础用法 ==
以下是最简单的 GET 请求示例：

<syntaxhighlight lang="python">
from airflow.providers.http.operators.http import SimpleHttpOperator

task_get = SimpleHttpOperator(
    task_id="get_api_data",
    method="GET",
    endpoint="/users",
    http_conn_id="my_http_connection",
    response_filter=lambda response: response.json()
)
</syntaxhighlight>

关键参数说明：
* '''http_conn_id'''：在 Airflow Connections 中预定义的 HTTP 连接
* '''endpoint'''：API 路径（不包含基础URL）
* '''response_filter'''：响应处理函数（本例解析 JSON）

== 连接配置 ==
需先在 Airflow UI 配置 HTTP 连接：
<mermaid>
graph LR
    A[Connections] --> B[Add new]
    B --> C[Connection ID: my_http_connection]
    C --> D[Conn Type: HTTP]
    D --> E[Host: https://api.example.com]
</mermaid>

== 进阶功能 ==
=== POST 请求示例 ===
<syntaxhighlight lang="python">
task_post = SimpleHttpOperator(
    task_id="create_user",
    method="POST",
    endpoint="/users",
    data=json.dumps({"name": "John", "email": "john@example.com"}),
    headers={"Content-Type": "application/json"},
    http_conn_id="my_http_connection"
)
</syntaxhighlight>

=== 响应处理 ===
通过 XCom 获取响应数据：
<syntaxhighlight lang="python">
def process_response(**kwargs):
    ti = kwargs['ti']
    response = ti.xcom_pull(task_ids='get_api_data')
    return response['total_count']

task_process = PythonOperator(
    task_id="process_response",
    python_callable=process_response
)
</syntaxhighlight>

=== 错误处理 ===
自动重试失败请求：
<syntaxhighlight lang="python">
task_with_retry = SimpleHttpOperator(
    task_id="retry_task",
    retries=3,
    retry_delay=timedelta(minutes=1),
    # ...其他参数...
)
</syntaxhighlight>

== 实际案例 ==
=== 场景：每日天气数据采集 ===
1. 调用气象局 API 获取数据
2. 解析 JSON 响应
3. 存储到数据库

<syntaxhighlight lang="python">
weather_task = SimpleHttpOperator(
    task_id="fetch_weather",
    method="GET",
    endpoint="/v1/forecast",
    data={"city": "Beijing", "units": "metric"},
    response_filter=lambda r: r.json()['hourly'],
    http_conn_id="weather_api"
)
</syntaxhighlight>

== 性能优化 ==
* 使用 '''connection pooling''' 减少连接开销
* 对高频请求启用 '''cache'''（需配置 Airflow 后端）
* 批量处理时考虑使用 '''Paginated API''' 分页获取

== 数学基础 ==
HTTP 请求延迟模型（简单估算）：
<math>
T_{total} = T_{connect} + T_{send} + T_{process} + T_{receive}
</math>
其中：
* <math>T_{connect}</math>：TCP 握手时间
* <math>T_{send}</math>：请求发送时间
* <math>T_{process}</math>：服务端处理时间
* <math>T_{receive}</math>：响应接收时间

== 常见问题 ==
{| class="wikitable"
|-
! 问题 !! 解决方案
|-
| 证书验证失败 || 添加 `extra={"verify":false}` 到连接配置（仅开发环境）
|-
| 响应中文乱码 || 设置 `headers={"Accept-Charset": "utf-8"}` 
|-
| 大文件下载超时 || 调整 `timeout` 参数或分块下载
|}

== 最佳实践 ==
* 敏感信息通过 '''Airflow Variables''' 管理
* 为不同环境（dev/stage/prod）配置独立的连接
* 对关键 API 实现 '''Circuit Breaker''' 模式
* 监控 HTTP 状态码分布（4xx/5xx 比例）

== 版本兼容性 ==
不同 Airflow 版本的差异：
* 2.0+：模块路径改为 `airflow.providers.http`
* 2.2+：新增 `response_check` 参数用于自定义响应验证
* 2.5+：支持异步 HTTP 请求（实验性功能）

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow Operators详解]]