dbt增量模型中如何跨批次获取历史值计算时间差？

我正在开发一个dbt模型，通过LEAD()窗口函数按DEVICE_ID分区、INTEGRATION_TIME降序排序，计算每个设备的集成时间与上一次唤醒时间的天数差。由于源数据量大且持续增长，计划将该模型改为增量模式，但遇到了挑战：后续增量运行时，无法从历史数据中获取某设备的最后唤醒时间，以计算该设备第一条新记录的时间差（即DEVICE_ID同时存在于历史和新数据中的场景）。我目前仅想到创建sequenceid，获取DEVICE_ID的最大值来继续，但不需要完整解决方案，希望得到一些提示。

{{
  config(
    materialized='incremental',
    unique_key='DEVICE_ID',
    cluster_by=['DEVICE_ID']
  )
}}

with lead_wake_up_time as
(
    select
        raw.DEVICE_ID,
        raw.INTEGRATION_TIME,
        raw.NEXT_WAKEUP_TIME,
        LEAD(raw.NEXT_WAKEUP_TIME) OVER (PARTITION BY raw.DEVICE_ID ORDER BY raw.INTEGRATION_TIME DESC) as _LAST_WAKEUP_TIME,
        raw.LDTS
    from {{ ref("raw_data") }} as raw
    where
        {% if is_incremental() %}
            and raw.LDTS > (select max(LDTS) from {{ this }})
        {% endif %}
)

select
    DEVICE_ID,
    INTEGRATION_TIME,
    NEXT_WAKEUP_TIME,
    _LAST_WAKEUP_TIME,
    datediff(day, _LAST_WAKEUP_TIME, INTEGRATION_TIME) as TIME_DIFF_IN_DAYS,
    LDTS
from
  lead_wake_up_time

提示方向 #

• 增量运行时，先从当前模型（{{ this }}）生成临时CTE，存储每个DEVICE_ID对应的最新NEXT_WAKEUP_TIME（即该设备历史数据中的最后唤醒时间）
• 将增量拉取的原始数据与上述CTE左关联，对每个设备的第一条新记录，直接使用关联到的历史NEXT_WAKEUP_TIME作为_LAST_WAKEUP_TIME；同一设备的后续新记录，继续用LEAD()在新数据内部计算
• 用COALESCE()函数优先取历史关联的唤醒时间，再取窗口函数计算的结果，确保第一条新记录的时间差计算准确
• 不要依赖DEVICE_ID的最大值来关联，应该基于每个设备的最新INTEGRATION_TIME或LDTS匹配历史数据，避免逻辑错误

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