如何在父DAG中触发其他DAG任务？需用DAG2清理DAG1的TaskA

完全可以在Airflow代码内实现跨DAG操作，下面是两种直接落地的方案，满足你用dag2清理并重新运行dag1中TaskA的需求：

方案一：用PythonOperator调用Airflow核心API直接操作任务实例 #

通过Airflow的内部模型类，直接定位目标任务实例，修改其状态或清理记录后触发运行，无需修改目标DAG（dag1）的代码。

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import TaskInstance, DagRun
from airflow.utils.state import State
from datetime import datetime

def clean_and_re_run_task_a(**context):
    # 定义目标DAG和任务的标识
    target_dag_id = "dag1"
    target_task_id = "TaskA"

    # 获取目标DAG最近一次成功的运行实例（可根据需求指定特定execution_date）
    latest_dag_run = DagRun.find(dag_id=target_dag_id, state=State.SUCCESS)[-1]
    execution_date = latest_dag_run.execution_date

    # 定位到对应的TaskInstance
    task_instance = TaskInstance(
        task_id=target_task_id,
        dag_id=target_dag_id,
        execution_date=execution_date
    )

    # 清理任务：标记为失败（触发自动重试，需TaskA配置了重试策略）
    task_instance.set_state(State.FAILED)
    # 若需彻底清除任务记录后重新触发，可替换为：task_instance.clear()

    # 手动触发任务运行
    task_instance.run(ignore_ti_state=True)

with DAG(
    dag_id="dag2",
    schedule_interval=None,
    start_date=datetime(2023, 1, 1),
    catchup=False
) as dag:
    clean_task = PythonOperator(
        task_id="clean_dag1_task_a",
        python_callable=clean_and_re_run_task_a,
        provide_context=True
    )

方案二：结合TriggerDagRunOperator与目标DAG分支逻辑 #

如果允许修改dag1的代码，可以在dag1中添加分支判断，通过dag2传递触发信号，让dag1仅重新运行TaskA。

第一步：修改dag1，添加分支控制 #

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from datetime import datetime

def check_retrigger_signal(**context):
    # 读取dag_run的配置参数，判断是否仅运行TaskA
    run_conf = context["dag_run"].conf or {}
    if run_conf.get("retrigger_task_a"):
        return "TaskA"
    return "normal_flow_start"

def normal_flow_start(**context):
    # 原DAG的正常流程起始任务
    pass

with DAG(
    dag_id="dag1",
    schedule_interval="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False
) as dag:
    branch_check = BranchPythonOperator(
        task_id="check_retrigger",
        python_callable=check_retrigger_signal,
        provide_context=True
    )

    normal_start = PythonOperator(
        task_id="normal_flow_start",
        python_callable=normal_flow_start
    )

    TaskA = PythonOperator(
        task_id="TaskA",
        python_callable=lambda: print("Executing TaskA")
    )

    # 构建分支逻辑
    branch_check >> [normal_start, TaskA]
    # 原DAG的其他任务连接到normal_start...

第二步：在dag2中触发dag1并传递信号 #

from airflow import DAG
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from datetime import datetime

with DAG(
    dag_id="dag2",
    schedule_interval=None,
    start_date=datetime(2023, 1, 1),
    catchup=False
) as dag:
    trigger_task_a = TriggerDagRunOperator(
        task_id="trigger_dag1_task_a",
        trigger_dag_id="dag1",
        conf={"retrigger_task_a": True},  # 传递仅运行TaskA的信号
        wait_for_completion=True  # 可选：等待TaskA运行完成
    )

注意事项 #

• 确保Airflow执行用户拥有操作目标DAG任务实例的权限
• Airflow 2.x版本中，模型类的导入路径与1.x一致，但部分API细节可能有调整，需对应版本文档验证
• 指定execution_date时需精准，避免误操作其他时间的任务实例

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