1 篇博文 含有标签「Apache Airflow」

在用 Airflow REST API 重复触发同一个 DAG 做灰度验证时，请求返回了 4xx 但响应体里没有 dag_run_id，DAG 实际根本没有运行——而脚本却把它当成了成功。

在开发 AI 运营 时遇到此问题——基于大语言模型的智能分析，自动洞察市场趋势、用户行为、销售数据，提供精准运营策略。广告决策链路的灰度切换需要在 Airflow 上反复触发同一次分析做对照，结果部分触发悄无声息地失败了。

TL;DR​

Airflow 对每个 DAG 的 logical_date 有唯一约束（dag_run_id 同样必须唯一）。用相同的 logical_date 重复 POST /dags/{dag_id}/dagRuns，Airflow 会拒绝并返回 4xx，响应体里没有 dag_run_id。如果你只检查 HTTP 状态码、不检查返回的 dag_run_id，就会误以为触发成功。解法：每次触发用不同的 logical_date（和不同的 dag_run_id）。

问题现象​

为了对照测试，用固定日期 2026-01-01 连续触发同一个 DAG：

$ curl -s -X POST "$AIRFLOW/api/v2/dags/my_dag/dagRuns" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{
      "dag_run_id": "manual-run-1",
      "logical_date": "2026-01-01T00:00:00Z"
    }'
# 第一次：返回正常的 dag_run 对象，包含 dag_run_id ✅

$ curl -s -X POST "$AIRFLOW/api/v2/dags/my_dag/dagRuns" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{
      "dag_run_id": "manual-run-2",
      "logical_date": "2026-01-01T00:00:00Z"   # ⚠️ 同一个 logical_date
    }'
# 第二次：返回错误对象，没有 dag_run_id ❌
{
  "detail": "...",
  "status": 400,
  "title": "Bad Request",
  "type": "https://airflow.apache.org/docs/apache-airflow/2/stable-rest-api-ref.html#/default/Error"
}

如果调用方只判断「HTTP 是否 2xx」就停止解析，或者直接读 JSON 不校验 dag_run_id 字段，第二次失败就会被静默吞掉——日志里看不到异常，Airflow UI 里也找不到这次 run。

根因​

Airflow 用 dag_run_id 作为每次运行的主键，同时在 metadata 数据库的 dag_run 表上对 (dag_id, logical_date) 维护唯一性。logical_date 是调度的「逻辑时间」——调度器按它判断某个调度槽位是否已经跑过。一旦同一个 DAG 下已存在某 logical_date 的 run，再用相同值触发，Airflow 就会拒绝，避免重复执行。

问题在于这个失败是 HTTP 4xx + 错误 JSON，不是连接错误或 5xx。很多脚本只做 response.status_code == 200 的粗判断，或者拿到 JSON 后直接取字段而不校验是否存在 dag_run_id，于是把「拒绝创建」当成了「创建成功」。

解决方案​

核心：每次触发用不同的 logical_date（以及不同的 dag_run_id）。 灰度/回放场景下，给每次触发拼一个不重复的日期即可：

# 每次循环用不同的 logical_date（2026-01-01 / 02 / 03 …）
for i in 1 2 3; do
  curl -s -X POST "$AIRFLOW/api/v2/dags/my_dag/dagRuns" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d "{
      \"dag_run_id\": \"manual-run-$i\",
      \"logical_date\": \"2026-01-0${i}T00:00:00Z\"
    }"
done

更稳妥的是用递增时间戳，保证 logical_date 和 dag_run_id 永不重复。更重要的是：必须校验响应体里的 dag_run_id 字段，把它当成「触发真正成功」的唯一证据：

import requests

def trigger_dag(dag_id: str, logical_date: str, conf: dict | None = None) -> str:
    resp = requests.post(
        f"{AIRFLOW}/api/v2/dags/{dag_id}/dagRuns",
        headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"},
        json={"dag_run_id": f"manual-{logical_date}", "logical_date": logical_date, "conf": conf or {}},
    )
    # ❌ 不够：只看状态码，4xx 会被当异常但容易漏判
    # resp.raise_for_status()
    data = resp.json()
    # ✅ 正确：dag_run_id 存在才算真正创建成功
    if "dag_run_id" not in data:
        raise RuntimeError(f"Trigger failed: {resp.status_code} {data}")
    return data["dag_run_id"]

# 每次用不同 logical_date，重复触发安全
for i in range(1, 4):
    trigger_dag("my_dag", f"2026-01-0{i}T00:00:00Z")

dag_run_id 同样要保持唯一——它是主键，重复会被直接拒绝。用「前缀 + logical_date」组合是常见做法，既唯一又能在 UI 里一眼识别。

常见问题​

airflow trigger_dagrun 怎么通过 REST API 触发 DAG？​

POST /api/v2/dags/{dag_id}/dagRuns，请求体至少包含 dag_run_id 和 logical_date 两个字段（可选 conf 传参数）。这两个字段在同一个 DAG 下都必须唯一，否则 Airflow 返回 4xx。代码里推荐用 TriggerDagRunOperator，它内部也会生成唯一的 run id。

airflow 重复触发同一个 DAG 为什么失败？​

因为 Airflow 在 metadata 数据库的 dag_run 表上对 (dag_id, logical_date) 维护唯一约束，dag_run_id 本身也是主键。重复的 logical_date 或 dag_run_id 都会被拒绝并返回 4xx。要做回放或灰度对照，每次触发换一个新的 logical_date（或递增时间戳）即可。