从DTL迁移至DBT实现的DLT流程：表写入配置难题

针对你遇到的多进程共享目标表、不能删除表且需保留原更新+审计日志写入的需求，以下是具体实现方案：

核心思路 #

放弃DBT默认的全量/增量模型替换逻辑，用临时中间结果+Post-Hook执行原生SQL的方式，直接复刻原DTL的update/insert + returning逻辑，既不破坏目标表结构，也能和其他进程并行写入。

1. 处理client_dim的更新+审计日志写入 #

将你的视图修改为生成待更新的client_id列表，通过Post-Hook执行原生update并把返回结果插入审计表：

{{ config(
    materialized='ephemeral', -- 不持久化中间表，仅作为临时子查询
    post_hook=[
        """
        update data_warehouse.client_dim
        set is_deleted = true,
            deleted_at = current_timestamp
        where client_id in (select client_id from {{ this }})
          and source_id = 1
          and is_deleted = false -- 保证幂等，避免重复更新
        returning 'delete' as type, client_id as key
        into data_warehouse.audit_matching_log (type, key)
        """
    ]
) }}

-- 生成待更新的client_id列表
select
    client_id
from {{ ref('stg__data_warehouse__client_dim') }}
where client_id in (select client_id from {{ ref('int__log_client__deletes') }})
  and client_dim_source_id = 1

2. 处理插入操作（类似逻辑） #

如果是需要插入新记录到client_dim并写入审计日志，用同样的模式：

{{ config(
    materialized='ephemeral',
    post_hook=[
        """
        insert into data_warehouse.client_dim (client_id, source_id, is_deleted, deleted_at, [其他列])
        select 
            client_id,
            1 as source_id,
            false as is_deleted,
            null as deleted_at,
            [其他列]
        from {{ this }}
        returning 'insert' as type, client_id as key
        into data_warehouse.audit_matching_log (type, key)
        """
    ]
) }}

-- 生成待插入的记录（按需过滤已存在的记录）
select
    client_id,
    [其他列]
from {{ ref('stg__your_source_table') }}
where not exists (
    select 1 from data_warehouse.client_dim 
    where client_id = {{ ref('stg__your_source_table') }}.client_id
)

关键说明 #

• ephemeral材质：仅作为临时逻辑层，不会在数据库创建持久化表，避免额外存储开销。
• Post-Hook原生SQL：直接执行原DTL的更新/插入逻辑，完全保留returning写入审计表的行为，且不会删除或重建目标表。
• 幂等性保证：更新语句中加入is_deleted = false，避免重复执行时修改已标记删除的记录。
• 并发兼容性：数据库的行级锁机制会保证仅锁定待更新/插入的行，不影响其他进程对目标表的写入操作。

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