Airflow问题：使用PythonVirtualenvOperator抛AirflowSkipException任务未跳过却失败

你遇到的问题根源在于：PythonVirtualenvOperator会启动独立的Python子进程执行你的callable函数，子进程内抛出的AirflowSkipException无法被主Airflow进程识别，反而会导致子进程异常退出（非0退出码），最终任务被标记为失败。

要让任务正确触发跳过状态，你可以在子进程中使用Airflow约定的退出码99（Airflow会将这个退出码解析为任务跳过），具体实现方式如下：

修改后的代码示例 #

import sys
from airflow.exceptions import SkipExitCode

def execute():
    a = 0
    if a > 0:
        print('continue running task')
    else: 
        # 用Airflow约定的跳过退出码替代直接抛出AirflowSkipException
        sys.exit(SkipExitCode)

def execute_task():
    return PythonVirtualenvOperator(
        task_id='test skipping',
        op_kwargs={},
        python_callable=execute,
        provide_context=True,
        system_site_packages=True,
        dag=dag
     )

补充说明 #

• SkipExitCode是Airflow提供的常量，值为99，直接使用它比硬编码数字更规范。
• 因为你设置了system_site_packages=True，虚拟环境可以访问主环境的Airflow包，所以能正常导入SkipExitCode；如果虚拟环境未共享系统包，需要确保Airflow依赖已在虚拟环境中安装，或者直接使用sys.exit(99)。

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