dbt微批增量策略配置后仍生成重复数据求助

问题概述 #

配置dbt微批增量策略时，已指定unique_keys='_id'，但首次全量执行时，所有数据被按批次重复处理3次，最终生成18条重复行；指定时间范围执行dbt run时，仍会插入时间范围外的数据。

示例种子数据 #

_id,event_time,value
1,2025-08-04 09:00:00,A
2,2025-08-04 10:00:00,B
3,2025-08-05 08:00:00,C
4,2025-08-05 09:00:00,D
5,2025-08-06 07:30:00,E
6,2025-08-06 08:00:00,F

当前模型代码 #

{{
    config(
        materialized='incremental',
        incremental_strategy='microbatch',
        event_time='event_time',
        begin='2025-08-04',
        batch_size='day',
        unique_keys='_id',
        concurrent_batches=false,
        lookback=2
    )
}}

WITH source_data AS (
    SELECT
        _id,
        event_time,
        value
    FROM
        {{ ref('sample_microbatch_seed') }}
),

deduplicated_source AS (
    SELECT
        *,
        ROW_NUMBER() OVER (PARTITION BY _id ORDER BY event_time DESC) as row_num
    FROM source_data
)

SELECT
    _id,
    event_time,
    value
FROM deduplicated_source
WHERE row_num = 1

问题现象 #

首次执行dbt run --select my_model_name后，目标表出现大量重复行，每条原始数据被插入3次：

_id,event_time,value
1,2025-08-04 09:00:00,A
1,2025-08-04 09:00:00,A
1,2025-08-04 09:00:00,A
2,2025-08-04 10:00:00,B
2,2025-08-04 10:00:00,B
2,2025-08-04 10:00:00,B
…
6,2025-08-06 08:00:00,F
6,2025-08-06 08:00:00,F
6,2025-08-06 08:00:00,F

已尝试无效操作 #

• 设置full_refresh=false无法阻止重复行生成
• 指定--event-time-start '2025-08-04' --event-time-end '2025-08-05'执行，仍会插入时间范围外的行

解决方案 #

1. 给源数据添加批次时间范围过滤 #

当前代码未对源数据做时间范围限制，导致每个微批都会读取全量种子数据，即使配置了unique_keys，每个批次仍会尝试插入所有数据（未触发更新逻辑）。需在source_data中加入增量判断与时间过滤：

WITH source_data AS (
    SELECT
        _id,
        event_time,
        value
    FROM
        {{ ref('sample_microbatch_seed') }}
    {% if is_incremental() %}
        -- 仅读取当前微批时间范围内的数据
        WHERE event_time >= '{{ var("microbatch_start") }}'::timestamp
          AND event_time < '{{ var("microbatch_end") }}'::timestamp
    {% endif %}
),

var("microbatch_start")和var("microbatch_end")是dbt微批策略自动生成的变量，对应每个批次的时间边界，确保每个批次仅处理对应时间段的数据。

2. 调整lookback参数 #

当前配置lookback=2，意味着每个微批会额外包含前2个batch_size（天）的时间范围数据，导致单个批次读取到超出预期的历史数据。如果不需要回溯历史数据，将lookback设为0：

{{
    config(
        ...
        lookback=0,
        ...
    )
}}

若业务需要回溯，根据实际需求设置合理值，避免不必要的全量数据读取。

3. 验证unique_keys的更新逻辑 #

微批策略中，unique_keys用于判断目标表中是否已存在该行：

• 若存在则执行更新
• 若不存在则执行插入
  确保目标表的unique_keys字段（_id）无空值，且源数据中_id是唯一标识，避免因数据问题导致更新逻辑失效。

4. 首次全量运行的正确方式 #

首次运行时，dbt会按begin到当前时间的范围生成所有批次，若想避免多批次重复处理，可通过--event-time-start和--event-time-end指定首次运行的时间范围，同时配合上述时间过滤逻辑，确保仅处理指定范围的数据。

内容的提问来源于stack exchange，提问作者Timothy Liew