如何在 DBT 中使用 Jinja 循环嵌套结构？

Question

我正在尝试在 DBT 中构建一个模型，该模型将一个结构体展平，该结构体的名称properties包含大约一百个结构体（例如property1、property2等），每个结构体都有 5 个不同的列，我想提取其中一个名为value. 我可以输入properties.propertyX.value100 次，但我想我可以尝试找到一种方法来循环遍历每个结构properties并使用 Jinja 在语句propertyX.value中获取SELECT，但我想我要么不熟悉语法或其限制，因为我不熟悉知道做什么。我试过了：

WITH t as (
    SELECT
        properties
    FROM
        src
)
SELECT
    {% for property in properties %}
    {{property}}.value
    {% endfor %}
    {%- if not loop.last %},{% endif -%}
FROM
    t

但我意识到我必须设置properties为变量，并且我真的不知道如何以引用结构中的各个属性的方式做到这一点properties。无论如何，我很迷失，如果有人可以帮助我，我将非常感激。

Answer 1

不确定我是否 100% 遵循了您的数据结构，但我们可以说它与此类似：

{
  "properties": {
    "property1": {
      "column1": "...",
      "column2": "...",
      "column3": "...",
      "value": "my value 1.0"
    },
    "property2": {
      "column1": "...",
      "column2": "...",
      "column3": "...",
      "value": "my value 2.0"
    },
    "propertyX": {
      "column1": "...",
      "column2": "...",
      "column3": "...",
      "value": "my value 3.0"
    }
  }
}

正如您所提到的，您需要使用它set来创建变量并能够操作数据。就我个人而言，我喜欢创建不同的变量来处理query statement、query result和query values。因此，按照这个策略，你会得到这样的结果：

{% set data_structure_query %}
    select properties from src
{% endset %}

{% set results = run_query(data_structure_query) %}

{% set properties = results.columns[0].values() %}

请注意，这results.columns[0].values()将带来查询第一列的数据，在本例中是properties.

获取.values()列的值作为元组，其中项目大多数时候定义为string. 因此，为了访问数据的属性，您必须将 json 字符串反序列化为 Python 对象，例如dict. 为此，您需要使用以下fromjson方法：

...

{% set properties = results.columns[0].values() %}

{% set properties_dict = fromjson(properties[0]) %}

...

假设您的查询仅返回 JSON 格式的一行，我指定properties[0]来访问结果查询的第一行。

在跳到下一步之前，重要的是要知道 dbt 有一个 jinja 变量，它会通知我们 dbt 何时处于“执行模式”。这是我们需要担心的事情，因为它可能会引发构建模型的问题。简而言之，任何依赖从数据库返回结果的 jinja 都会抛出错误。

在您的情况下，该results变量取决于需要在数据库中执行的值，这意味着如果您只是尝试运行模型，很可能会遇到Compilation Error. 为了避免这种情况，您需要添加一个if condition来检查 dbt 是否处于“执行模式”：

...

{% set results = run_query(data_structure_query) %}

{% if execute %}
    {% set properties = results.columns[0].values() %}
    {% set properties_dict = fromjson(properties[0]) %}
{% else %}
    {% set properties = [] %}
    {% set properties_dict = [] %}
{% endif %}

...

最后，您可以继续loop构建您的列：

select
{%- for property in properties_dict.properties %}
    {{ property }}.value
    {%- if not loop.last %},{% endif -%}
{%- endfor %}
from 
...

这将被编译为：

select
    property1.value,
    property2.value,
    propertyX.value
from
...

如果您想访问每列的值，则：

select
{%- for property in properties_dict.properties %}
    '{{ properties_dict.properties[property].value }}'
    {%- if not loop.last %},{% endif -%}
{%- endfor %}
from
...

它将被编译为：

select
    'my value',
    'my value 1.0',
    'my value 2.0'
from
...

可能值得查看您的数据库/仓库并检查是否有任何处理半结构化数据的内部函数。这也可以帮助你理解逻辑。例如，Snowflake 具有lateral flatten类似的行为，将属性拆分为多行。

出于调试目的，我建议compile您的模型并使用日志 ( {{ log('my message', info=True) }}) 来了解 dbt/jinja 如何处理数据。我提供的一些代码可能会根据您的查询的输出而改变。

一些有用的链接：

https://docs.getdbt.com/reference/dbt-jinja-functions/run_query

https://docs.getdbt.com/reference/dbt-jinja-functions/execute

https://docs.getdbt.com/reference/dbt-jinja-functions/fromjson/

https://docs.getdbt.com/tutorial/using-jinja