airflow 当前版本（2.9）不支持在 web ui、rest api 或 cli 中强制用户输入 run_id；但可通过自定义 rest api 调用 + 前置校验的中间层方案，确保每次手动触发均使用业务唯一标识（如数据 dump id）作为 run_id，从而实现处理幂等性与可追溯性。
airflow 当前版本（2.9）不支持在 web ui、rest api 或 cli 中强制用户输入 run_id；但可通过自定义 rest api 调用 + 前置校验的中间层方案，确保每次手动触发均使用业务唯一标识（如数据 dump id）作为 run_id，从而实现处理幂等性与可追溯性。
在构建面向数据 dump 处理的 Airflow DAG 时，将业务维度的唯一 ID（例如 dump_20240515_abc123）直接映射为 DAG 运行的 run_id，是保障任务可审计、可重放、避免重复执行的关键设计。然而，Airflow 默认行为是：若未显式传入 run_id，系统将自动生成形如 manual__{timestamp} 的随机 ID，且 Web UI 和原生命令行工具均不提供必填 run_id 的表单或参数约束机制。
✅ 可行解决方案：基于 REST API 的受控触发中间层
推荐在 Airflow 前端（如内部运维平台、脚本或低代码表单）中封装一层调用逻辑，通过 Airflow POST /dags/{dag_id}/dagRuns 接口发起触发，并在调用前强制校验 dag_run_id 字段：
import requests
import json
def trigger_dag_with_required_run_id(dag_id: str, run_id: str, conf: dict = None):
if not run_id or not isinstance(run_id, str) or not run_id.strip():
raise ValueError("run_id is required and must be a non-empty string")
# 可选：校验 run_id 格式是否符合业务规范（如匹配 dump ID 模式）
import re
if not re.fullmatch(r'dump_\d{8}_[a-zA-Z0-9]{6,}', run_id):
raise ValueError("run_id must match pattern: dump_YYYYMMDD_<alphanumeric>")
url = f"http://your-airflow-webserver:8080/api/v1/dags/{dag_id}/dagRuns"
headers = {
"Content-Type": "application/json",
"Authorization": "Basic ..." # 使用 Basic Auth 或 JWT Token
}
payload = {
"dag_run_id": run_id,
"conf": conf or {}
}
response = requests.post(url, headers=headers, data=json.dumps(payload))
response.raise_for_status()
return response.json()
# 示例调用
trigger_dag_with_required_run_id(
dag_id="process_data_dump",
run_id="dump_20240515_x9f2m7",
conf={"dump_path": "/data/dumps/20240515/x9f2m7.json"}
)</alphanumeric>
⚠️ 注意事项与最佳实践
- 幂等性保障：Airflow 不允许同一 dag_id + run_id 重复提交（会返回 409 Conflict），因此该方案天然防止误重复触发；
- 权限控制：务必限制中间层服务对 Airflow REST API 的访问权限，仅允许可信来源调用；
- 错误反馈：前端应清晰提示用户 run_id 的命名规则与校验失败原因，提升操作体验；
- 兼容性提醒：截至 Airflow 2.9，官方尚未合并 issue #33945 等增强 run_id 约束的提案，未来升级需关注变更日志；
- 替代思路（轻量级）：若无法部署中间层，可在 DAG 内部 on_failure_callback 或首个 task 中主动校验 context['dag_run'].run_id 是否符合预期正则，不合规则直接 raise AirflowFailException，但此属运行时拦截，不如前置校验严谨。
综上，虽 Airflow 原生不支持强制 run_id 输入，但通过 REST API 封装 + 业务规则前置校验，可稳健实现「每个 dump ID 对应唯一、可控、可追溯的 DAG 运行实例」这一核心目标。
