用于重复任务的 DBT 宏答案

【问题标题】：DBT macro for repetitive task用于重复任务的 DBT 宏
【发布时间】：2022-06-13 22:26:21
【问题描述】：

我是 DBT 的初学者。我有一个要求，我创建了一个如下所示的增量模型。我需要为不同的系统执行相同的增量模型逻辑语句。

我需要传递 3 个变量或参数。即对于每次运行，ATTRIBUTE_NAME、VIEW_NAME、SYSTEM_NAME 都需要通过。对于下一次运行，所有 3 个参数都会不同。

但是，对于特定的 SYSTEM_NAME，VIEW_NAME 和 ATTRIBUTE_NAME 是固定的。

请帮助我使用宏执行 dbt 运行以满足此要求并传递不同的系统名称及其对应的视图名称和属性名称。目标是使用单个 dbt 运行语句并为所有 ATTRIBUTE_NAME、VIEW_NAME、SYSTEM_NAME 执行此模型。

目前，我已经为每个系统定义了变量并分别执行每次运行，如下所示在 CLI 中

例如

dbt run --vars '{"VIEW_NAME": CCC, "SYSTEM_NAME": BBBB, "ATTRIBUTE_NAME": AAAA}' -m incremental_modelname

dbt run --vars '{"VIEW_NAME": DDD, "SYSTEM_NAME": FFF, "ATTRIBUTE_NAME": HHH}' -m incremental_modelname

dbt run --vars '{"VIEW_NAME": EEE, "SYSTEM_NAME": GGG, "ATTRIBUTE_NAME": III}' -m incremental_modelname

可重复使用的增量模型：

{{ 
    config(
        materialized='incremental', 
        transient=false, 
        unique_key='composite_key',  
        post_hook="insert into table (col1, col2, col3)
                        select 
                            '{{ var('ATTRIBUTE_NAME') }}',
                            col2, 
                            col3 
                        from {{ this }}  a
                        join table b on a=b
                        where b.SYSTEM_NAME='{{ var('SYSTEM_NAME') }}';
                         commit;"
    ) 
}}


with name1 AS (
    select 
        *
    from {{ var('VIEW_NAME') }}
),

select 
    *
from name1

{% if is_incremental() %}

where (select timestamp_column from {{ var('VIEW_NAME') }}) > 
    (select max(timestamp_column) from {{ this }} where SYSTEM_NAME='{{ var("SYSTEM_NAME") }}')
                             
{% endif %}

【问题讨论】：

标签： dbt

【解决方案1】：

最简单的方法是：

创建一个包含系统名称、视图名称和属性名称的模型（甚至是种子）。

在您的代码中，添加一个 for 循环

 {% set query %}
 select system_name, view_name, attribute_name from model_name
 {% endset %}

 {% set results = run_query(query) %}

 {% for result in results %}

     /*
     Put your query here but reference the columns needed 
     results.columns[0].values()  = system_name
     results.columns[1].values()  = view_name
     results.columns[2].values()  = attribute_name
     */

【讨论】：