如何在 DBT 中编写增量模型？答案

【问题标题】：How do you write incremental models in DBT?如何在 DBT 中编写增量模型？
【发布时间】：2022-01-07 15:36:43
【问题描述】：

我正在尝试使用 dbt 加入和过滤两个大表。

SQL很简单，大致如下：

SELECT
   u.user_id, t.transaction_id
FROM users u
JOIN transactions t ON t.user_id = u.user_id
WHERE u.active = 1

目前我正在使用“表”具体化，但这相当浪费，因为表的基础表在每次运行中都是 99.99%。

但是，我从 DBT 文档中不明白如何将此模型设置为“增量”。

有什么想法吗？

PS。我在 SQL Server 上运行。

【问题讨论】：

你能分享你如何知道一行是否是新的吗？表中是否有primary_key 和date_last_updated？这是你需要的第一件事。也许transaction_id 是独一无二的？
我没有。不幸的是，我们从供应商那里获取这些数据，他们保持最新的方式是他们对我们的数据库执行任意写入/更新/删除。我可以连接到数据库日志来获取这些数据，但这会很痛苦。
@MYK 任何一个答案都对您有帮助吗？
感谢您的提示。我已经接受了你的回答。我还没有尝试过，但它看起来应该可以工作。
太棒了，谢谢！如果您遇到更多麻烦，请回来。

标签： dbt

【解决方案1】：

正如@anders-swanson 在他的评论中所写，如果transaction_id 绝对是唯一的，您可以将其设置为unique_key 并将您的模型具体化为增量表。

dbt's docs 解释如何做到这一点。使用您的示例，它可能是：

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

select
   u.user_id, t.transaction_id
from users u
join transactions t ON t.user_id = u.user_id
where u.active = 1

如果transaction_id 不是唯一的但transaction_id||user_id 是唯一的，您可以尝试在上游 dbt 模型中创建一个连接这些列的新列，然后将其分配为 unique_key：

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

select
   u.user_id, 
   t.transaction_id,
   u.user_id||t.transaction_id as pkey
from users u
join transactions t ON t.user_id = u.user_id
where u.active = 1

否则，您将不得不拉入一个列，该列要么是 a) 唯一的，要么 b) 具有可用于应用 is_incremental() 过滤器的有序质量（就像 @viacheslav-nefedov 所写的那样）。

【讨论】：

【解决方案2】：

如果您有日期字段，则可以使用它仅加载最后的数据。 IE。你有 transaction_date 列。

{{ config(
  materialized='incremental',
    as_columnstore=false, 
  pre_hook="""
    {% if is_incremental() %}
      delete from {{this}} 
      where transaction_date >= '{{ (modules.datetime.datetime.now() - modules.datetime.timedelta(2)).isoformat() }}'
    {% endif %}  
  """
)}}

SELECT
   u.user_id, t.transaction_id
FROM users u
JOIN transactions t ON t.user_id = u.user_id
WHERE u.active = 1

{% if is_incremental() %}
 and transaction_date >= '{{ (modules.datetime.datetime.now() - modules.datetime.timedelta(2)).isoformat() }}'
{% endif %}

第一次运行此模型时，它将忽略“if is_incremental()”下的所有代码。所有即将进行的运行，它将清理过去两天的事务并重新加载它们。

【讨论】：