GCP环境下Airflow Python Operator内更新Variable失败求助

原因分析 #

1. 执行上下文与配置差异：

   • 外部执行代码时，使用的是本地Airflow配置，变量默认存储在Airflow的元数据库（如PostgreSQL/SQLite）中，因此能正常找到并更新变量。
   • Python Operator中的代码在Airflow Worker节点上运行，而Worker的Airflow配置可能启用了Google Cloud Secret Manager作为变量存储后端。此时Worker会去GCSM中查找名为inv_test_var的密钥，但该密钥并未在GCSM中创建，因此抛出NotFound错误。

2. 权限问题：
   即使GCSM中存在对应密钥，Airflow Worker使用的服务账号可能没有Secret Manager Editor或更高权限，导致无法执行更新操作。

解决方法 #

• 统一变量存储后端配置：
  检查Airflow全局配置文件airflow.cfg中的variables_backend项：

  • 如果不需要使用GCSM作为变量后端，将其修改为默认的元数据库存储：

    variables_backend = airflow.models.variable.Variable
    

  • 如果要统一使用GCSM，需要将原本存在于元数据库的inv_test_var变量迁移到GCSM中：在GCSM控制台创建名为inv_test_var的密钥，值设置为目标内容。

• 配置Worker服务账号权限：
  确保Airflow Worker使用的Google Cloud服务账号拥有Secret Manager Editor（或Secret Manager Admin）角色，使其具备创建、更新GCSM密钥的权限。

• 代码层面临时指定存储后端（不推荐）：
  若需临时在Operator中使用元数据库存储变量，可在代码中临时指定后端（不建议全局混用，易造成数据不一致）：

  from airflow.models.variable import Variable as MetastoreVariable
  
  def py_function():
      MetastoreVariable.update('inv_test_var','2023-06-28 04:20:00')
  

内容的提问来源于stack exchange，提问作者Mani Shankar.S