已自定义Run ID与执行日期仍触发DagRunAlreadyExists异常

问题场景 #

为MWAA的DAG指定自定义Run ID和执行日期后，在一秒内发起多次触发请求时，仍抛出DagRunAlreadyExists异常，错误日志显示同一执行日期和Run ID已存在。

核心问题分析 #

1. 执行日期格式错误 #

原代码中执行日期的格式化字符串存在低级错误：

# 错误：%m 代表月份，而非分钟
execution_date = datetime.utcnow().strftime("%Y-%m-%dT%H:%m:%S.%f")

这导致同一小时内所有请求的执行日期分钟部分被替换为当前月份（如10月则分钟固定为10），即使时间不同，执行日期的核心部分也会重复，触发Airflow的唯一性校验。

2. Run ID生成的潜在风险 #

• 使用datetime.now()（本地时间）生成时间戳，若本地时区与UTC不同步，可能导致跨时区的重复时间戳
• random模块非线程安全，高并发场景下可能生成重复的随机字符串
• 时间戳仅到秒级，同一秒内的请求可能生成重复的时间前缀

解决方案 #

1. 修复执行日期格式 #

将分钟的格式符从%m改为%M，并保留微秒级精度，同时使用UTC时间确保一致性：

from datetime import datetime
# 正确格式：%M 代表分钟，保留微秒并添加UTC时区标识
execution_date = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ")

2. 增强Run ID的唯一性 #

修改Run ID生成逻辑，统一使用UTC微秒时间戳+短UUID，确保全局唯一性：

def get_unique_key():
    from datetime import datetime
    import shortuuid
    
    # UTC微秒时间戳，确保同一秒内也有差异
    utc_ts = datetime.utcnow().strftime("%Y%m%d%H%M%S%f")
    short_uuid = shortuuid.ShortUUID().random(length=12)
    return f"{short_uuid}{utc_ts}"

3. 避免CLI命令的字符串转义问题 #

使用shlex.quote处理配置参数，避免特殊字符导致命令执行失败：

import json
import shlex

conf = json.dumps({"foo": "bar"})
quoted_conf = shlex.quote(conf)
command = f"dags trigger {workflow_id} -c {quoted_conf} -r {dag_run_id} -e {execution_date}"

4. 并发场景下的重试机制 #

极端并发场景下仍可能出现碰撞，添加重试逻辑捕获异常后重新触发：

import subprocess
import time

def trigger_dag_safe(workflow_id, conf, max_retries=3):
    for _ in range(max_retries):
        try:
            execution_date = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ")
            dag_run_id = get_unique_key()
            quoted_conf = shlex.quote(json.dumps(conf))
            command = f"dags trigger {workflow_id} -c {quoted_conf} -r {dag_run_id} -e {execution_date}"
            subprocess.run(command, shell=True, check=True, capture_output=True, text=True)
            return
        except subprocess.CalledProcessError as e:
            if "DagRunAlreadyExists" in e.stderr:
                time.sleep(0.1)
                continue
            raise
    raise RuntimeError(f"触发DAG失败，已重试{max_retries}次")

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