使用 Jinja 迭代 DBT 中的不同列值

Question

我是 DBT 和 Jinja 的新手，目前正在开发一个项目，我需要迭代不同的 ID 值，根据每个 ID 过滤表，然后将结果具体化为单独的表或将其导出为 CSV文件到S3。有人可以指导我如何在 DBT 中执行此操作吗？

{% set get_id_query %}
  select distinct id from  table_name limit 10
{% endset %}

{% set results = run_query(get_id_query) %}

{% if execute %}
 {# Return the first column #}
 {% set results_list = results.columns[0].values() %}
 {% else %}
 {% set results_list = [] %}
{% endif %}

{% for id in results_list %}
 select * from table_name
  WHERE id = '{{ id }}'
 limit 1 
{% endfor %}

Answer 1

我看到你已经有了一些代码。我想你还是需要知道如何使用它，对吗？

我看到在你的第一个问题中，你谈论了 dbt 配置块。如果您想使用默认配置（默认 dbt“表”具体化，或 dbt_project.yml 文件中定义的配置），则不需要此块。假设您想要更改默认配置并希望增量加载模型（以便我可以为您提供更完整的代码示例）。在这种情况下，dbt 模型可能如下所示：

型号：

{% set get_id_query %}
  select distinct id from table_name limit 10
{% endset %}

{% if execute %} {# So that your query doesn't run at compilation phase #}
 {# Return the first column #}
 {% set results_list = run_query(get_id_query).columns[0].values() %}
{% endif %}


{{
  config(
    materialized = 'incremental',
   )
}}

{% if results_list|length > 0 %}
 {% for id in results_list %}
  SELECT *
  FROM table_name
  WHERE id = '{{ id }}'
 {% if not loop.last %}UNION ALL {% endif %}
 {% endfor %}
{% else %}
 SELECT * FROM table_name LIMIT 0
{% endif %}

如果你想保留

LIMIT 1

的东西，如果可以的话，你必须应用一些

GROUP BY

或

DISTINCT

，或者在 CTE 内应用

ROW_NUMBER()

，然后对

row_number = 1

进行过滤。

我希望这有帮助！