Airflow通过DAG配置变量触发唯一约束冲突报错排查

嘿，我来帮你搞定这个Airflow的报错问题！

首先咱们先理清楚错误根源：你在DAG里配置变量的逻辑，每次运行时都尝试插入sql_path这个变量，但它已经存在于Airflow的元数据库里了——Airflow的变量key是全局唯一的，数据库里有个variable_key_key的唯一约束，不允许重复插入同一个key的变量，所以就抛出了这个完整性错误。

下面给你几个实用的解决方案，按适配你重部署需求的优先级排序：

1. 用Airflow变量导入文件（最推荐） #

既然你的目标是快速搭建环境，完全没必要在DAG里写变量创建逻辑。Airflow支持用YAML/JSON文件批量导入变量，而且这个操作是幂等的（重复执行不会报错），完美适配部署脚本。

• 先创建一个airflow_variables.yaml文件，把所有需要的变量写进去：

  sql_path: "/your/actual/sql/file/path"
  db_connection_string: "postgresql://user:pass@host:port/db"
  # 其他需要的变量...
  

• 然后用Airflow命令导入：

  airflow variables import airflow_variables.yaml
  

  你可以把这个命令加到你的部署脚本里，每次重部署时运行一遍就行，完全不用担心重复问题。

2. 在DAG里使用replace=True参数 #

如果必须在DAG里动态设置变量，记得给Variable.set()加上replace=True参数，这样当变量存在时会直接覆盖，而不是尝试插入新的：

from airflow.models import Variable

# 存在就覆盖，不存在就创建，不会触发重复错误
Variable.set(key="sql_path", value="/your/actual/path", replace=True)

3. 先检查变量是否存在再创建 #

如果你不想覆盖现有变量的值，只想在变量不存在时才创建，可以用异常捕获来判断：

from airflow.models import Variable

try:
    # 尝试获取变量，不存在会抛出KeyError
    Variable.get("sql_path")
except KeyError:
    # 变量不存在时才创建
    Variable.set(key="sql_path", value="/your/actual/path")

4. 调整DAG变量代码的位置 #

最后要注意：如果你的变量创建代码放在DAG的顶层（也就是没有嵌套在Operator的execute方法里），那么每次Airflow解析DAG时（默认每隔30秒一次）都会执行这段代码，很容易反复触发插入操作。建议把这类初始化逻辑移到部署脚本里，而不是留在DAG中。

总结一下：这个问题的核心就是重复插入同一key的变量，只要让你的变量配置操作变成幂等的，就能彻底解决啦！

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