为何HttpOperator导致本地Airflow调度器崩溃？求助排查

问题概述 #

在macOS上以独立模式运行Airflow（搭配SQLite数据库）做测试，BashOperator可正常执行，但每次运行HttpSensor任务时调度器都会崩溃。已尝试删除所有Airflow进程并通过airflow standalone重启，问题仍重复出现，不确定是SQLite/SequentialExecutor的限制还是Http组件问题。

环境信息 #

• Airflow版本：apache-airflow==2.5.1
• 目标API端点已验证可正常访问
• 已配置HTTP连接：test_api，指向目标服务的public/v2路径

任务日志 #

PycharmProjects/composer/logs/dag_id=test_api/run_id=scheduled__2024-01-30T09:00:00+00:00/task_id=is_api_active/attempt=4.log
[2024-01-31, 11:40:28 UTC] {taskinstance.py:1083} INFO - Dependencies all met for <TaskInstance: test_api.is_api_active scheduled__2024-01-30T09:00:00+00:00 [queued]>
[2024-01-31, 11:40:28 UTC] {taskinstance.py:1083} INFO - Dependencies all met for <TaskInstance: test_api.is_api_active scheduled__2024-01-30T09:00:00+00:00 [queued]>
[2024-01-31, 11:40:28 UTC] {taskinstance.py:1279} INFO - 
--------------------------------------------------------------------------------
[2024-01-31, 11:40:28 UTC] {taskinstance.py:1280} INFO - Starting attempt 4 of 4
[2024-01-31, 11:40:28 UTC] {taskinstance.py:1281} INFO - 
--------------------------------------------------------------------------------
[2024-01-31, 11:40:28 UTC] {taskinstance.py:1300} INFO - Executing <Task(HttpSensor): is_api_active> on 2024-01-30 09:00:00+00:00
[2024-01-31, 11:40:28 UTC] {standard_task_runner.py:55} INFO - Started process 44845 to run task
[2024-01-31, 11:40:28 UTC] {standard_task_runner.py:82} INFO - Running: ['airflow', 'tasks', 'run', 'test_api', 'is_api_active', 'scheduled__2024-01-30T09:00:00+00:00', '--job-id', '29', '--raw', '--subdir', 'DAGS_FOLDER/test_api.py', '--cfg-path', '/var/folders/84/5q_1yj555494bdjdy7m78pv40000gp/T/tmp6pg6_8t6']
[2024-01-31, 11:40:28 UTC] {standard_task_runner.py:83} INFO - Job 29: Subtask is_api_active
[2024-01-31, 11:40:28 UTC] {task_command.py:388} INFO - Running <TaskInstance: test_api.is_api_active scheduled__2024-01-30T09:00:00+00:00 [running]> on host pc-156.home
[2024-01-31, 11:40:28 UTC] {taskinstance.py:1507} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_EMAIL=xxxxxx
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=test_api
AIRFLOW_CTX_TASK_ID=is_api_active
AIRFLOW_CTX_EXECUTION_DATE=2024-01-30T09:00:00+00:00
AIRFLOW_CTX_TRY_NUMBER=4
AIRFLOW_CTX_DAG_RUN_ID=scheduled__2024-01-30T09:00:00+00:00
[2024-01-31, 11:40:28 UTC] {http.py:122} INFO - Poking: posts/
[2024-01-31, 11:40:28 UTC] {base.py:73} INFO - Using connection ID 'test_api' for task execution.

DAG代码 #

import json

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

from airflow.providers.http.sensors.http import HttpSensor
from airflow.utils.dates import days_ago
from airflow.providers.http.operators.http import SimpleHttpOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['Benoit.FAGOT.ext@ag2rlamondiale.fr'],
    'email_on_failure': True,
    'retries': 0,
}

with DAG(
    'test_api',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval="0 9 * * *",
    start_date=days_ago(2),
    catchup=False,
    tags=['example'],

) as dag:

    def debug(response):
        print(response)
        if "200" in response:
            return True
        return False

    task_is_api_active = HttpSensor(
        task_id="is_api_active",
        http_conn_id='test_api',
        endpoint='posts/'
    )

排查与解决方案 #

1. SQLite/SequentialExecutor限制 #

SQLite搭配SequentialExecutor为单进程执行模式，HttpSensor作为周期性触发的传感器任务，可能因持续占用数据库连接或进程资源导致调度器崩溃。

• 临时调整传感器参数：增加poke_interval（如设为30秒），降低请求频率，减少资源占用
• 切换执行器：修改airflow.cfg中executor = LocalExecutor，重启Airflow后测试

2. HttpSensor配置优化 #

当前代码未显式指定response_check逻辑，默认的状态码检查可能因响应处理异常引发问题。建议显式配置：

task_is_api_active = HttpSensor(
    task_id="is_api_active",
    http_conn_id='test_api',
    endpoint='posts/',
    response_check=lambda response: response.status_code == 200,
    poke_interval=10,
    timeout=300  # 设置超时时间，避免无限等待
)

3. 检查Http Provider依赖 #

确保HTTP provider版本与Airflow兼容：

pip install apache-airflow-providers-http==2.4.0

4. 补充日志排查 #

当前日志仅记录到任务启动阶段，未包含崩溃时的异常信息。需查看调度器日志（logs/scheduler目录下的文件），获取完整异常栈以定位具体问题。

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