基于BigQuery的增量DBT每日变更日志模型内存溢出问题求助

问题背景 #

需要基于包含create/update/delete操作的变更日志，在BigQuery上构建DBT增量模型，生成每日各ID的数据快照，核心规则：

• 每日取ID的最新有效数据（当日多次更新取最后一次）
• 被delete的ID在删除当日及之后不再展示
• 无变更的日期沿用前一日数据

原始变更日志结构：

ID  TIMESTAMP        DATA     OPERATION
id1 2023-01-01 13:40 data1_v1 create
id1 2023-01-01 15:00 data1_v2 update
id1 2023-01-03 00:02 data1_v3 update
id1 2023-01-04 05:04 data1_v3 delete
id2 2023-01-01 XX:XX data2_v1 create
id2 2023-01-02 XX:XX data2_v2 update
id2 2023-01-04 XX:XX data2_v3 update

期望生成的每日快照：

ID  DATE       DATA    
id1 2023-01-01 data1_v2
id1 2023-01-02 data1_v2
id1 2023-01-03 data1_v3
id2 2023-01-01 data2_v1
id2 2023-01-02 data2_v2
id2 2023-01-03 data2_v2
id2 2023-01-04 data2_v3

当前实现的问题 #

原代码触发内存溢出的核心原因：

• 窗口函数未按document_id分区，导致BigQuery对全量数据执行全局排序，数据量较大时超出内存限制
• 全量日期与变更日志的left join产生大量冗余数据，且未处理delete操作的过滤逻辑

优化方案 #

1. 预处理变更日志：过滤delete操作，提取每个ID每日的最后一条有效记录，减少后续计算的数据量
2. 精准日期关联：生成日期序列后，仅保留ID存在的日期区间，避免无效的ID-日期组合
3. 分区窗口函数：按document_id分区执行窗口函数，仅对单个ID的日期序列排序，降低内存占用
4. 增量逻辑优化：增量运行时仅处理新增日期及对应变更数据，缩小计算范围

优化后的DBT模型代码 #

{{ config(
    materialized='incremental',
    unique_key=['document_id', 'date']
) }}

with
-- 生成需要覆盖的日期范围（全量/增量模式自适应）
date_range as (
    select day as date
    from unnest(generate_date_array(
        {% if is_incremental() %}
            (select max(date) from {{ this }}) + interval 1 day
        {% else %}
            date_sub(current_date(), interval 6 month)
        {% endif %}
        , current_date()
    )) day
),

-- 预处理变更日志：过滤delete，取每个ID每日最后一条记录
processed_changelog as (
    select
        document_id,
        date(timestamp) as date,
        data,
        timestamp,
        row_number() over (partition by document_id, date(timestamp) order by timestamp desc) as rn
    from {{ ref("base_firestore_export__event_raw_changelog") }}
    where operation != 'delete'
    {% if is_incremental() %}
        and date(timestamp) >= (select min(date) from date_range)
    {% endif %}
),

latest_daily_changes as (
    select
        document_id,
        date,
        data
    from processed_changelog
    where rn = 1
),

-- 生成ID与有效日期的组合（仅保留ID存在的日期区间）
id_date_combinations as (
    select distinct
        l.document_id,
        d.date
    from latest_daily_changes l
    cross join date_range d
    where d.date >= (select min(date) from latest_daily_changes where document_id = l.document_id)
),

-- 填充无变更日期的数据
final_snapshot as (
    select
        dc.document_id,
        dc.date,
        last_value(ldc.data ignore nulls) over (
            partition by dc.document_id
            order by dc.date
            rows between unbounded preceding and current row
        ) as data
    from id_date_combinations dc
    left join latest_daily_changes ldc
        on dc.document_id = ldc.document_id
        and dc.date = ldc.date
)

select * from final_snapshot

关键优化点说明 #

• 数据量压缩：预处理阶段过滤冗余变更记录，将数据量降至每日每个ID仅一条记录
• 内存负载降低：按ID分区的窗口函数避免全局排序，内存占用仅为原方案的极小部分
• 无效数据排除：仅生成ID存在的日期组合，减少不必要的计算
• 增量效率提升：增量模式下仅处理新增日期及对应变更，大幅缩短运行时间

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