Airflow中如何让运行数小时的常驻任务自动跳过执行下游任务？

你的场景核心问题是Airflow默认将「进程正常退出且返回码为0」作为任务成功的判定标准，你写的inotifywait常驻监听逻辑没有主动退出机制，才会一直卡在运行中，直接用execution_timeout会因为进程被强制杀死返回非0码被标记为失败，以下是可落地的实现方案：

这个方案完全不需要修改Airflow任务配置，也不会产生失败任务标记，是最符合Airflow设计规范的实现。
直接修改Task1的bash脚本，用系统自带的timeout命令给监听流程加最长运行时间限制，到点自动终止监听、正常返回0退出码，任务会被标记为成功，下游任务按默认规则自动触发。
修改后的脚本参考：

#!/bin/bash
# 配置最长监听时长，单位为秒，比如x小时就换算为x*3600填入
WATCH_MAX_DURATION=$((4*3600)) # 示例为4小时，替换为实际需要的时长
WATCH_PATH="/<path>"
OUTPUT_PATH="/<output_path>"

# 用timeout包裹监听命令，到点自动发送终止信号结束进程
timeout --signal=SIGTERM $WATCH_MAX_DURATION inotifywait -m $WATCH_PATH -e create -e moved_to |
    while read dir action file; do
        echo "The file '$file' appeared in directory '$dir' via '$action'"
        unzip -o -q "$WATCH_PATH/$file" "*.csv" -d $OUTPUT_PATH
        # 原脚本此处路径写死，补全为变量避免误删文件
        rm "$WATCH_PATH/$file"
    done

# 兜底返回0退出码，确保任务被标记为成功
exit 0

如果暂时不能修改原有bash脚本，可以通过Airflow任务配置组合实现效果，但是属于补丁式写法，后续维护成本更高：

• 给Task1配置execution_timeout为你需要的x小时
• 给Task1增加失败回调函数，超时触发失败判定时，直接把任务状态强制改为成功
• 把下游Task2的触发规则改为「上游无失败即可触发」，跳过默认的所有上游必须成功的校验
  对应的Python配置代码参考：

from airflow.utils.state import State
from airflow.utils.trigger_rule import TriggerRule
from datetime import timedelta

def force_mark_success(context):
    task_instance = context["ti"]
    task_instance.set_state(State.SUCCESS)

task1 = BashOperator(
    task_id="listen_and_unzip",
    bash_command="sh /path/to/your/script.sh",
    execution_timeout=timedelta(hours=4), # 替换为实际需要的x小时
    on_failure_callback=force_mark_success,
    dag=dag
)

task2 = PythonOperator(
    task_id="clean_and_load_to_mysql",
    python_callable=your_etl_function,
    trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
    dag=dag
)

Airflow是批量工作流调度系统，不是常驻服务托管平台，不建议把长时间运行的监听类逻辑放在DAG节点中执行：

• 长期运行的文件监听逻辑建议单独用systemd、supervisor等进程管理工具托管，和调度流程解耦
• DAG中的Task1可以改为单次执行逻辑：每次调度运行时只扫描对应调度周期内到达的zip文件，解压处理完成后直接退出，不需要常驻监听，流程更稳定也更容易排查问题

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