DBT如何将多个查询结果加载至垂直（键值）表

该场景完全可以通过DBT实现，无需依赖官方未提供的多查询加载专属语法，核心思路是将所有产品对应的查询结果通过UNION ALL合并为单结果集，配合DBT增量模型配置即可完成垂直表写入，完全兼容你当前在Snowflake中执行的多INSERT写入逻辑。

1. 配置增量模型基础参数 #

在模型SQL文件开头添加DBT配置块，指定增量物化模式、去重主键即可，示例配置适配Snowflake环境：

{{ config(
    materialized='incremental',
    unique_key=['UNIQUEID', 'ACCOUNTID', 'PRODUCTID', 'KEY', 'DATE'],
    merge_update_columns = ['VALUE', 'PPD']
) }}

配置说明：

• materialized='incremental'：启用增量写入模式，和你之前使用的单查询增量填充逻辑一致
• unique_key：指定联合主键，匹配你业务场景下的记录唯一粒度，增量写入时会自动根据主键去重，避免重复插入数据
• merge_update_columns：如果主键已存在，自动更新VALUE、PPD字段；如果不需要更新历史数据、仅做追加写入，直接删除该配置项即可

2. 合并多产品查询逻辑 #

你之前通过多个独立INSERT语句写入不同产品数据，在DBT中只需要将每个产品对应的SELECT语句用UNION ALL拼接，保证所有SELECT块的字段顺序、字段类型和目标表完全对齐即可，原有业务关联、过滤逻辑不需要修改。示例结构如下：

-- 产品ID: 5898988asdfas
SELECT 
    UNIQUEID,
    ACCOUNTID,
    PRODUCTID,
    TARGET AS KEY,
    VALUE,
    PPD,
    CURRENT_DATE() as DATE
FROM accounttable t
INNER JOIN producttable t2 
    ON t.accountid = t2.accountid
WHERE t2.productid = '5898988asdfas'
    AND t.type = 'prodtype'

UNION ALL

-- 产品ID: 第二个待同步产品ID
SELECT 
    UNIQUEID,
    ACCOUNTID,
    PRODUCTID,
    TARGET AS KEY,
    VALUE,
    PPD,
    CURRENT_DATE() as DATE
FROM accounttable t
INNER JOIN producttable t2 
    ON t.accountid = t2.accountid
WHERE t2.productid = '第二个待同步产品ID'
    AND t.type = '该产品对应的type值'

-- 后续按相同格式追加所有需要同步的产品查询块即可

3. （可选）添加增量过滤提升运行效率 #

如果数据量较大，可以在每个SELECT块末尾添加增量过滤条件，增量运行时仅拉取最近更新的数据，减少全表扫描开销：

{% if is_incremental() %}
    -- 增量运行时仅拉取目标表最新日期前3天的新数据，可根据业务调整时间范围
    AND t.update_time >= (SELECT DATEADD(day, -3, MAX(DATE)) FROM {{ this }})
{% endif %}

该条件仅在增量运行时生效，全量刷新模型时不会触发。

如果需要同步的产品数量较多，手动维护大量UNION ALL块成本较高，可以通过Jinja循环自动生成所有查询逻辑，不需要重复编写相同结构的SQL。示例如下：

-- 维护产品配置列表，新增产品仅需在列表中添加对应配置项
{% set product_configs = [
    {"product_id": "5898988asdfas", "prod_type": "prodtype"},
    {"product_id": "第二个产品ID", "prod_type": "对应type值"}
] %}

{% for config in product_configs %}
SELECT 
    UNIQUEID,
    ACCOUNTID,
    PRODUCTID,
    TARGET AS KEY,
    VALUE,
    PPD,
    CURRENT_DATE() as DATE
FROM accounttable t
INNER JOIN producttable t2 
    ON t.accountid = t2.accountid
WHERE t2.productid = '{{ config.product_id }}'
    AND t.type = '{{ config.prod_type }}'
    {% if is_incremental() %}
        AND t.update_time >= (SELECT DATEADD(day, -3, MAX(DATE)) FROM {{ this }})
    {% endif %}
{% if not loop.last %} UNION ALL {% endif %}
{% endfor %}

内容的提问来源于stack exchange，提问作者Randy B.