Apache Airflow中HttpHook与requests库对比及DAG刷新相关问题咨询

关于HttpHook改善Session重建的问题 #

默认情况下，HttpHook每次实例化确实会创建新的requests.Session，但关键在于你把Hook的实例化放在哪个作用域：

• 如果在DAG文件的全局范围实例化Hook，那每次DAG文件刷新（受min_file_process_interval控制）时都会重建Hook和对应的Session，这和你当前用requests.Session的问题一样。
• 但如果把Hook的实例化放在Operator的execute方法内部，Hook和Session只会在任务实际执行时创建，完全不受DAG刷新流程的影响。这样就能避免每30秒重建Session的问题，因为DAG刷新只是解析DAG结构，不会触发Operator的执行逻辑。

Hook与DAG刷新流程的关系 #

Hook本身不是天生不受DAG刷新影响，影响与否取决于你实例化它的位置：

• 全局作用域（DAG定义时）：会随DAG文件解析刷新重建。
• 任务执行作用域（execute方法内）：仅在任务运行时初始化，和DAG刷新无关。

HttpHook底层虽然也是用requests.Session，但只要用对作用域，就能解决你当前Session频繁重建的问题。

令牌缓存避免DAG刷新清除的方案 #

因为你的令牌和Session都在Airflow插件的utils.py模块中，而插件是在Airflow启动时加载的，不会随DAG文件刷新重新加载。利用这一点，你可以在utils.py中用类级别的缓存来维护令牌和Session，避免每次DAG刷新被清除：

示例实现（utils.py） #

import requests
from airflow.providers.http.hooks.http import HttpHook
from datetime import datetime, timedelta

class APIHelper:
    _session = None
    _token = None
    _token_expiry = datetime.min

    @classmethod
    def get_valid_token(cls):
        # 检查令牌是否过期
        if datetime.now() >= cls._token_expiry:
            # 重新获取令牌的逻辑
            auth_hook = HttpHook(http_conn_id="api_auth_conn", method="POST")
            response = auth_hook.run("/auth", data={"username": "xxx", "password": "xxx"})
            cls._token = response.json()["access_token"]
            # 假设令牌有效期是1小时，可根据实际调整
            cls._token_expiry = datetime.now() + timedelta(hours=1)
        return cls._token

    @classmethod
    def get_session(cls):
        if not cls._session:
            cls._session = requests.Session()
            # 设置默认Headers，包含令牌
            cls._session.headers.update({"Authorization": f"Bearer {cls.get_valid_token()}"})
        else:
            # 每次获取Session时检查令牌是否过期，更新Headers
            cls._session.headers["Authorization"] = f"Bearer {cls.get_valid_token()}"
        return cls._session

在Operator中使用 #

from airflow.models.baseoperator import BaseOperator
from utils import APIHelper

class APICallOperator(BaseOperator):
    def execute(self, context):
        session = APIHelper.get_session()
        response = session.get("https://your-api-endpoint/data")
        # 处理响应逻辑

这样：

• APIHelper的类变量会在插件加载时初始化，不会随DAG刷新重置，只有Airflow服务重启时才会重新初始化。
• 每次调用API前会自动检查令牌是否过期，仅在过期时重新获取，避免无效请求。

内容的提问来源于stack exchange，提问作者tomomomo