Airflow EMR DAG依赖配置需求：MWAA调度仅执行指定任务路径

核心需求 #

在保留原有所有任务依赖关系的前提下，仅执行test_create_cluster_a -> setup_active_emr_cluster_a_id -> test_emr_trigger_job_a -> test_emr_trigger_job_b -> test_emr_trigger_job_c -> test_emr_trigger_job_d这条路径，同时允许故障时从任意指定步骤重新执行其他路径。

方案一：通过Airflow变量控制任务启停 #

1. 定义控制变量：在Airflow UI的「Admin > Variables」中创建布尔型变量RUN_ONLY_CLUSTER_A，默认设为True。
2. 添加短路控制任务：给test_create_cluster_b、test_create_cluster_c、test_create_cluster_d这三个起始路径的任务添加前置短路任务，根据变量值决定是否跳过后续任务：

from airflow.operators.python import ShortCircuitOperator
from airflow.models import Variable

# 控制Cluster B路径的短路任务
skip_cluster_b = ShortCircuitOperator(
    task_id="skip_cluster_b",
    python_callable=lambda: not Variable.get("RUN_ONLY_CLUSTER_A", default_var=False),
    dag=dag
)

# 调整依赖：短路任务 -> Cluster B起始任务
skip_cluster_b >> test_create_cluster_b >> setup_active_emr_cluster_b_id >> test_emr_trigger_job_b

# 同理处理Cluster C路径
skip_cluster_c = ShortCircuitOperator(
    task_id="skip_cluster_c",
    python_callable=lambda: not Variable.get("RUN_ONLY_CLUSTER_A", default_var=False),
    dag=dag
)
skip_cluster_c >> test_create_cluster_c >> setup_active_emr_cluster_c_id >> test_emr_trigger_job_c

# 同理处理Cluster D路径
skip_cluster_d = ShortCircuitOperator(
    task_id="skip_cluster_d",
    python_callable=lambda: not Variable.get("RUN_ONLY_CLUSTER_A", default_var=False),
    dag=dag
)
skip_cluster_d >> test_create_cluster_d >> setup_active_emr_cluster_d_id >> test_emr_trigger_job_d

3. 切换执行模式：当需要执行所有路径时，将RUN_ONLY_CLUSTER_A设为False即可，原有依赖关系完全保留，不影响故障后的重跑操作。

方案二：触发DAG时传入配置参数（更灵活） #

无需创建全局变量，每次触发DAG时通过MWAA的触发配置传入参数，控制仅执行指定路径：

1. 修改起始任务逻辑：为test_create_cluster_b、test_create_cluster_c、test_create_cluster_d添加前置短路任务，读取DAG运行时的配置：

from airflow.operators.python import ShortCircuitOperator

def should_run_other_clusters(**context):
    dag_conf = context.get("dag_run").conf or {}
    # 如果配置中指定run_only_cluster_a为true，则跳过其他路径
    return not dag_conf.get("run_only_cluster_a", False)

# Cluster B的短路控制
skip_cluster_b = ShortCircuitOperator(
    task_id="skip_cluster_b",
    python_callable=should_run_other_clusters,
    provide_context=True,
    dag=dag
)
skip_cluster_b >> test_create_cluster_b >> setup_active_emr_cluster_b_id >> test_emr_trigger_job_b

# 同理复制逻辑处理Cluster C、D的短路任务

2. 触发DAG时传入配置：在MWAA触发DAG时，填写配置JSON：

{"run_only_cluster_a": true}

这种方式无需修改全局变量，每次触发可灵活选择执行路径，原有依赖关系不受影响。

关键注意事项 #

• 两种方案都只是跳过其他路径的起始任务，而非删除依赖关系，因此故障时依然可以手动触发任意中间任务（比如setup_active_emr_cluster_b_id），依赖链会正常执行后续任务。
• 所有操作完全兼容MWAA的Airflow运行环境，无需额外配置。

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