Airflow：如何将 json / dict 格式的变量传递给操作员？答案

【问题标题】：Airflow: How to pass a variable in json / dict format to an operator?Airflow：如何将 json / dict 格式的变量传递给操作员？
【发布时间】：2023-01-04 01:44:42
【问题描述】：

例如，我需要知道如何将已注册的字典作为运算符参数中的变量传递以启动数据块笔记本。

就我而言，我尝试了一些方法但没有用。

我将此变量保存在 Airflow 中：

"dictionaries" : {
                     "dict1" : {
                                "a" : 1,
                                "b" : 2
                               }

和这段代码来尝试检索变量

dict = "dict1"
values = f"{{{{ var.json.dictionaries.{dict} }}}}"

稍后将其作为字典与运算符参数中的其他值一起传递

task1 = DatabricksRunNowOperator(
            task_id=f'Databricks_{dict1}',
            databricks_conn_id='databricks',
            job_id= 1111,
            notebook_params={"param1": "param1" , **values}

这是失败的，因为变量作为 str “TypeError: 'str' object is not a mapping” 出现，所以我尝试使用 json 库尝试转换格式但我没有成功，我收到错误消息“期望用双引号引起来的属性名称"

所以我怀疑它可能是用单引号检索变量，我也尝试过使用 replace 来尝试将它们更改为双引号，但这也没有用。

json.loads(values.replace("'","\""))

使用 jinja 来检索这个变量可能表现不一样，我对此有点迷茫，我最后试过

values = json.loads(f"{{{{ (var.json.dictionaries.{dict1}).replace('\'','\"') }}}}")
values = json.loads((f"{{{{ var.json.dictionaries.{dict1} }}}}").replace('\'','\"'))

也在运营商内部

notebook_params={"param1": "param1", **json.loads((f"{{{{ var.json.dictionaries.{dict1} }}}}").replace('\'','\"'))}

但得到相同的错误“json.decoder.JSONDecodeError：期望用双引号引起来的属性名称”

当然，我误解了恢复变量并对其进行转换的方式，如果有人可以帮助我解决这个问题，我将不胜感激。

问候

【问题讨论】：

标签： json python-3.x variables jinja2 jinja2-cli

【解决方案1】：

您必须保存变量的值，如下所示：

{
 "dictionaries" : {
                   "dict1" : {
                             "a" : 1,
                             "b" : 2
                            }
                   }
}

当您访问代码中的变量时，传递 deserialize_json=True。假设您的变量当时称为“json”，

from airflow.models import Variable
json_data = Variable.get('json', deserialize_json=True)
dict1 = json_data['dictionaries']['dict1']

【讨论】：