Airflow使用TriggerDagRunOperator多次触发DAG出现dag_id唯一键冲突如何解决

问题根因 #

报错的核心原因是两个TriggerDagRunOperator实际传入的execution_date完全相同，触发同一个目标DAG时触发了Airflow元数据库中dag_run表的dag_id+execution_date联合主键唯一性约束。
你误以为两个配置的执行日期不同，但实际在天级调度场景下：

• {{ execution_date }}默认值就是调度周期的起始零点（例如2021-11-01 00:00:00）
• {{ ds }}是日期格式的字符串，Airflow解析时会自动补全时分秒为00:00:00
  两者最终解析出的执行时间完全一致，且两个触发任务并行执行时，即使配置了reset_dag_run = True也会因为竞态条件报错：两个任务同时校验目标DAG Run是否存在，同时发起创建请求，reset逻辑还未生效就触发了主键冲突。

可选解决方案 #

• 方案1：执行时间加微小偏移
  给第二个触发任务的执行时间加1秒的偏移，保证两个执行时间唯一，不影响业务侧对日期的判断：

  y = TriggerDagRunOperator(
      task_id="y_external_dag",
      trigger_dag_id="external_dag",
      python_callable=pass_args_for_y,
      # 加1秒偏移避免执行时间重复
      execution_date="{{ execution_date + macros.timedelta(seconds=1) }}",
      reset_dag_run = True
  )
  

• 方案2：自定义参数传递日期，自动生成执行时间
  如果不需要强制指定目标DAG的执行时间，可以移除execution_date参数，让Airflow自动生成唯一的执行时间，业务需要用到的日期参数放到conf字段中传递即可。
• 方案3：串行触发两个任务
  给两个触发任务配置依赖关系，让第一个任务执行完成后再触发第二个，保证reset_dag_run逻辑可以正常生效：

  x >> y
  

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