配置与优化 dbt 增量模型-未完纪

本文介绍在 dbt 中配置和优化增量模型。

Snowflake 列大小变更

原文提示，Snowflake 计划于 2026 年 9 月增大字符串与二进制类型的默认列大小。低于 v1.10.6 的 dbt-snowflake，在该变化部署后，可能无法构建某些增量模型。

如果适配器低于 v1.10.6，或 dbt 平台尚未迁移到 release track，同时模型包含定义了 collation 的字符串列，并设置 on_schema_change='sync_all_columns'，就可能受影响。

可以运行以下命令查找相关模型：

dbt ls -s config.materialized:incremental,config.on_schema_change:sync_all_columns --resource-type model

返回“No nodes selected!”则无需操作。如果返回模型，例如“Found 1000 models, 644 macros”，且这些模型有未指定宽度的字符串列，则可能受影响，应升级到包含修复的版本：

dbt v1：dbt-snowflake v1.10.6 或更高，升级方法见 v1 安装说明。
dbt 平台：任意 release track，包括 v1 Latest、Compatible、Extended、Fallback。
dbt v2：v2.0.0。

这样可在保留所需 collation 设置的同时，安全处理 schema 变化。

增量模型是什么

增量模型在数据仓库中物化为表。首次运行时转换全部源数据建表；之后只转换你指定筛选的行，再写入已存在的目标表。

通常筛选的是上次运行后新增或更新的数据，因此每次运行逐步构建模型。减少转换数据量，能够显著缩短时间、改善仓库性能并降低计算成本。

配置增量物化

与其他内置物化一样，使用 select 定义模型，并在 config 中指定：

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

select ...

还需要告诉 dbt 如何在增量运行时筛选行，以及模型的唯一键，如果有的话。

is_incremental() 宏

只有以下条件全部满足时返回 True：模型已作为表存在；没有传入 full-refresh；当前模型配置 materialized=’incremental’。

无论宏返回 True 还是 False，模型 SQL 都必须合法。

筛选增量行

将筛选所需行的有效 SQL 包在 is_incremental() 条件中。

通常需要获取上次运行之后新增的行。最可靠的时间依据，是目标表中最新时间戳。使用 {{ this }} 可以方便查询目标表。

如果还要捕获更新的记录，需要定义 unique_key，避免把修改后的记录作为重复行插入。条件应筛选上次运行后创建或修改的行。

例如，某列需要昂贵计算的模型可这样增量构建。文件：models/stg_events.sql。

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

select
    *,
    my_slow_function(my_column)

from {{ ref('app_data_events') }}

{% if is_incremental() %}

  -- this filter will only be applied on an incremental run
  -- (uses >= to include records whose timestamp occurred since the last run of this model)
  -- (If event_time is NULL or the table is truncated, the condition will always be true and load all records)
where event_time >= (select coalesce(max(event_time),'1900-01-01') from {{ this }} )

{% endif %}

优化：复杂模型使用 CTE 时，注意 is_incremental() 的位置。在某些仓库中，越早筛选，查询越快。

incremental_predicates

这是面向数据量很大、值得投入额外优化的高级配置，接受有效 SQL 表达式列表，dbt 不检查其 SQL 语法。

Snowflake 上的 YAML 配置示例：


models:
  - name: my_incremental_model
    config:
      materialized: incremental
      unique_key: id
      # this will affect how the data is stored on disk, and indexed to limit scans
      cluster_by: ['session_start']  
      incremental_strategy: merge
      # this limits the scan of the existing table to the last 7 days of data
      incremental_predicates: ["DBT_INTERNAL_DEST.session_start > dateadd(day, -7, current_date)"]
      # `incremental_predicates` accepts a list of SQL statements. 
      # `DBT_INTERNAL_DEST` and `DBT_INTERNAL_SOURCE` are the standard aliases for the target table and temporary table, respectively, during an incremental run using the merge strategy.

同样配置也可以写在模型内：

-- in models/my_incremental_model.sql

{{
  config(
    materialized = 'incremental',
    unique_key = 'id',
    cluster_by = ['session_start'],  
    incremental_strategy = 'merge',
    incremental_predicates = [
      "DBT_INTERNAL_DEST.session_start > dateadd(day, -7, current_date)"
    ]
  )
}}

...

dbt.log 中会生成类似 merge：

merge into <existing_table> DBT_INTERNAL_DEST
    from <temp_table_with_new_records> DBT_INTERNAL_SOURCE
    on
        -- unique key
        DBT_INTERNAL_DEST.id = DBT_INTERNAL_SOURCE.id
        and
        -- custom predicate: limits data scan in the "old" data / existing table
        DBT_INTERNAL_DEST.session_start > dateadd(day, -7, current_date)
    when matched then update ...
    when not matched then insert ...

还应在增量模型 SQL 内限制上游表扫描范围，减少处理和转换的“新”数据：

with large_source_table as (

    select * from {{ ref('large_source_table') }}
    {% if is_incremental() %}
        where session_start >= dateadd(day, -3, current_date)
    {% endif %}

),

...

定义唯一键

可选 unique_key 允许更新已有行，而不是只追加。如果相同键出现新信息，可以替换旧信息；重复行也可以被忽略。更多行为，例如只更新特定列，见具体增量策略配置。

未设置 unique_key 时，多数适配器只追加模型 SQL 返回的全部行，不考虑是否重复。

unique_key 指定模型的粒度，可以是单列或多列组合，必须标识唯一行。可在模型顶部 config 中设置，值为单列名称字符串，或列名列表。

多列示例为 ['col1', 'col2']。这些列不应有 null，否则可能匹配失败并产生重复。可以使用 coalesce 填补，或通过 dbt_utils.generate_surrogate_key 生成单列代理键。

提示：多列唯一键推荐写成 unique_key = ['user_id', 'session_number']，而不是 concat 字符串表达式。列表更通用，dbt 能按数据库特性正确生成 SQL。仍须保证各列无空值，否则可能运行失败；另一选择是单列代理键。

设置后，每条新数据的行为是：

新旧数据都有同一键：更新或替换旧行，具体机制取决于数据库、增量策略及配置。
旧数据没有该键：插入整行。

如果现有目标表或新增数据中，一个键对应多行，可能导致模型失败。排查时，应确认两边的键都真正唯一。

说明：delete+insert、merge 等策略会使用 unique_key，但 insert_overwrite 按分区操作，不使用它。

唯一键示例

假设根据事件流计算日活用户数 DAU。新数据到达时，需要重新计算上次运行所在日，以及之后各天。文件：models/staging/fct_daily_active_users.sql。

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

select
    date_trunc('day', event_at) as date_day,
    count(distinct user_id) as daily_active_users

from {{ ref('app_data_events') }}


{% if is_incremental() %}

  -- this filter will only be applied on an incremental run
  -- (uses >= to include records arriving later on the same day as the last run of this model)
  where date_day >= (select coalesce(max(date_day), '1900-01-01') from {{ this }})

{% endif %}

group by 1

没有 unique_key，同一天每次运行都会新增一行。设置它后，会更新已有日期行。

重新构建增量模型

模型逻辑变化后，新数据采用的转换可能与目标表内历史数据不同，此时应重建。

使用 –full-refresh，会先删除现有目标表，再根据全部历史数据重新构建：

$ dbt run --full-refresh --select my_incremental_model+

末尾的 + 也会运行所有依赖此模型的下游模型，其中的增量模型同样全部刷新。

还可以在项目或资源级别用 full_refresh 指定始终或从不全量刷新。显式 true／false 的配置优先于命令行是否传入 –full-refresh。详细用法见 dbt run 文档。

增量模型的列发生变化

可选 on_schema_change 让 dbt 在 schema 变化时继续运行增量模型，减少全量刷新和查询成本。

dbt_project.yml：

models:
  +on_schema_change: "sync_all_columns"

模型内配置：

{{
    config(
        materialized='incremental',
        unique_key='date_day',
        on_schema_change='fail'
    )
}}

可选值：

ignore：默认，行为见下文。
fail：源与目标 schema 不一致时报错。
append_new_columns：追加新列，但不删除目标表中已不再出现在新数据里的列。
sync_all_columns：添加新列、删除缺失列，并处理类型变化。在 BigQuery 上，类型变化需要全表扫描，应权衡成本。

注意：这些选项都不会为新增列回填历史值。如需回填，可手动更新或执行 –full-refresh。

目前只跟踪顶层列变化，不跟踪嵌套列。例如 BigQuery 上增删改嵌套字段，不会触发 schema 变化处理。

默认行为

默认 on_schema_change: ignore。新增列后执行 dbt run，目标表不会出现该列；删除列后执行，则运行失败。

因此，当增量逻辑变化时，应对该模型以及下游模型执行 full-refresh。

原文：Configure incremental models。作者／维护者：dbt 文档团队。本文为原文的中文译文；代码保留原文内容。

文章版权归作者所有，未经允许请勿转载。

THE END

技术译文

配置与优化 dbt 增量模型