如何基于变量条件在Airflow DAG中不显示指定任务？

我明白你的需求——你想要的不只是跳过任务执行，而是让任务在DAG UI里直接“消失”，对吧？你当前的代码只是调整了任务依赖关系，但middle_name_task还是被实例化了，所以它依然会出现在UI中。

要实现动态隐藏任务的效果，关键在于只在条件满足时才创建这个任务实例，而不是先创建再通过依赖跳过。下面是修改后的优雅实现：

from airflow.operators import PythonOperator
from datetime import datetime, timedelta
from airflow.models import Variable

middle_name_var = Variable.get('middle_name')

default_args = {
    'owner': 'test',
    'depends_on_past': False,
    'start_date': datetime(2018, 6, 18),
    'email': ['tes@abc.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=1)
}

dag = DAG(
    'name',
    default_args=default_args,
    schedule_interval="@once")

def first_name():
    print('John')

def middle_name():
    print('Smith')

def last_name():
    print('Doe')

first_name_task = PythonOperator(
    task_id='first_name',
    provide_context=False,
    python_callable=first_name,
    dag=dag
)

last_name_task = PythonOperator(
    task_id='last_name',
    provide_context=False,
    python_callable=last_name,
    dag=dag
)

# 仅当条件满足时才创建中间任务
if middle_name_var == 'true':
    middle_name_task = PythonOperator(
        task_id='middle_name',
        provide_context=False,
        python_callable=middle_name,
        dag=dag
    )
    # 连接完整的任务链
    first_name_task >> middle_name_task >> last_name_task
else:
    # 直接连接首尾任务
    first_name_task >> last_name_task

关键说明： #

1. 任务实例化时机：把middle_name_task的创建逻辑放到了if判断块内，当middle_name_var为false时，这个任务根本不会被创建，自然不会在Airflow UI中显示。
2. 依赖动态调整：根据是否存在中间任务，分别设置不同的依赖关系，保证任务链始终正确。

这种方式利用了Airflow DAG的解析时生成特性，从根源上控制任务的存在与否，是最直接且优雅的实现方式。

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