能否在Airflow DAG中直接创建Variable变量

当然可以，Airflow内置的Variable.set()方法本身就支持upsert逻辑：当传入的key不存在时会自动创建全新的Variable条目，key已存在时才执行更新操作，你不需要额外调用create、save类的方法。

你给出的示例逻辑可以简化为如下可直接运行的代码：

from airflow.models import Variable

def create_new_client_config(client_name):
    data = get_client_details(client_name)
    # 自动判断key是否存在，不存在则创建，存在则更新
    Variable.set(key=f"{client_name}", value=data)

常见使用注意事项 #

• 如果存储的是字典、列表等结构化数据，取值时需要添加deserialize_json=True参数就能直接拿到原结构数据，示例：client_config = Variable.get(f"{client_name}", deserialize_json=True)
• 如果要存储密钥、密码等敏感信息，创建时添加encrypt=True参数即可实现加密存储，Airflow UI会自动隐藏对应值的明文：

  Variable.set(key=f"{client_name}_secret", value=sensitive_data, encrypt=True)
  

• 不要在DAG顶层全局代码中直接调用Variable的读写方法，该操作会在DAG解析阶段频繁请求元数据库，拖慢调度器性能，所有Variable操作都要放在Task的执行逻辑内部。

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