Airflow调度器重复触发API登录请求致DAG导入失败求助

核心原因 #

Airflow调度器会定期（默认每30秒）重新解析所有DAG文件，用于检测DAG定义的更新。你把get_conn()直接写在DAG定义的顶层代码中，每次解析DAG都会执行这个函数发起登录请求——哪怕DAG的schedule设为None、从未实际运行，也会因为频繁触发API请求而被限流，最终导致DAG导入失败。

解决办法 #

不需要纠结是否要在每个PythonOperator单独初始化，正确的做法是把连接初始化逻辑移到PythonOperator的执行函数内部，而非放在DAG定义的顶层：

• 改写后的代码示例：

def task1_func():
    conn = get_conn()  # 仅在任务实际运行时才执行登录
    # 执行op1的业务逻辑

def task2_func():
    conn = get_conn()
    # 执行op2的业务逻辑

with dag:
    op1 = PythonOperator(
        task_id="task1",
        python_callable=task1_func
    )

    op2 = PythonOperator(
        task_id="task2",
        python_callable=task2_func
    )

调整后，只有当DAG任务被实际触发运行时，才会调用get_conn()发起登录请求，DAG解析阶段不会再执行该逻辑，从根源上避免了限流问题。

额外优化 #

如果想减少重复登录的次数，可以在get_conn()中加入缓存机制（比如用functools.lru_cache），但要注意API token的有效期，避免使用过期连接。另外，Airflow的每个任务可能运行在不同的Worker进程中，进程间缓存不共享，这种优化仅能在单个任务进程内生效。

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