Airflow 2中TriggerDagRunOperator触发的run_id显示为manual_的修改求助

方法1：手动指定run_id参数 #

TriggerDagRunOperator支持直接传入自定义run_id，你可以在实例化操作符时，生成带scheduled_或triggered_前缀的唯一标识：

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

trigger_task = TriggerDagRunOperator(
    task_id="trigger_target_dag",
    trigger_dag_id="你的目标DAG ID",
    # 用当前时间戳生成triggered前缀的run_id
    run_id=f"triggered_{int(datetime.now().timestamp())}",
    # 若要基于调度执行日期生成，用Jinja模板语法：
    # run_id=f"scheduled_{{{{ execution_date.strftime('%Y%m%d%H%M%S') }}}}",
    wait_for_completion=False,
    dag=dag
)

方法2：重写TriggerDagRunOperator的execute方法 #

如果需要全局统一替换前缀，可以自定义操作符类，覆盖原有的run_id生成逻辑：

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

class CustomTriggerDagRunOperator(TriggerDagRunOperator):
    def execute(self, context):
        if not self.run_id:
            # 生成带triggered_前缀的唯一run_id
            self.run_id = f"triggered_{int(datetime.now().timestamp())}"
            # 也可以替换Airflow默认生成的manual_前缀：
            # original_run_id = super().generate_run_id(context)
            # self.run_id = original_run_id.replace("manual_", "triggered_")
        super().execute(context)

之后用这个自定义操作符触发DAG，所有生成的run_id都会自动使用指定前缀。

方法3：用PythonOperator调用trigger_dagrun函数 #

通过PythonOperator直接调用Airflow底层触发函数，完全掌控run_id的生成逻辑：

from airflow.operators.python import PythonOperator
from airflow.api.common.experimental.trigger_dag import trigger_dagrun

def trigger_target_dag(**context):
    # 基于执行日期生成scheduled前缀的run_id
    run_id = f"scheduled_{context['execution_date'].strftime('%Y%m%d%H%M%S')}"
    trigger_dagrun(
        dag_id="你的目标DAG ID",
        run_id=run_id,
        conf=context.get("conf"),
        execution_date=context.get("execution_date")
    )

trigger_task = PythonOperator(
    task_id="trigger_target_dag",
    python_callable=trigger_target_dag,
    provide_context=True,
    dag=dag
)

注意事项 #

• 确保run_id的唯一性，Airflow不允许同一DAG存在重复的run_id，时间戳或执行日期是可靠的唯一标识来源。
• 若基于调度触发，优先使用execution_date生成run_id，避免因任务重试导致的时间戳重复问题。

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