如何在DBT CORE增量模型的post_hook中发送POST请求？

这个需求完全可行，问题出在宏的实现逻辑和post_hook的调用方式上，以下是具体解决步骤：

1. 明确新记录的获取方式 #

增量模型运行后，新插入的记录可以通过两种可靠方式获取：

• 利用模型中的加载时间字段（比如loaded_at），查询目标表中本次运行新增的记录（即加载时间大于上次模型运行的最大加载时间）
• 如果用了快照或者dbt的增量更新逻辑，也可以通过dbt_valid_to字段过滤出最新的有效记录

2. 实现send_to_endpoint宏 #

宏需要完成「提取新ID」和「发送POST请求」两个核心动作，分两种场景给出实现：

Python宏（通用场景，需环境支持requests） #

{% macro send_to_endpoint(target_table, loaded_at_col) %}
    -- 1. 查询本次新增的记录ID
    {% set get_new_ids_sql %}
        SELECT id FROM {{ target_table }}
        {% if is_incremental() %}
            WHERE {{ loaded_at_col }} > (SELECT MAX({{ loaded_at_col }}) FROM {{ this }})
        {% endif %}
    {% endset %}

    {% set new_ids_result = run_query(get_new_ids_sql) %}
    {% if new_ids_result.rows | length > 0 %}
        -- 2. 整理ID列表为请求 payload
        {% set ids_list = new_ids_result.rows | map(attribute='id') | list %}
        {% set payload = {'ids': ids_list} %}

        -- 3. 发送POST请求
        {% set response = requests.post(
            url='https://your-backend-url.com/receive-ids',
            json=payload,
            headers={'Content-Type': 'application/json'}
        ) %}

        -- 4. 检查请求状态，失败则终止任务
        {% if response.status_code not in [200, 201] %}
            {% do exceptions.raise_compiler_error(f"发送ID失败：{response.status_code} - {response.text}") %}
        {% endif %}
    {% endif %}
{% endmacro %}

SQL宏（适配支持HTTP请求的数据仓库，如Snowflake） #

{% macro send_to_endpoint(target_table, loaded_at_col) %}
    -- 1. 聚合新记录ID为数组
    {% set get_ids_array_sql %}
        SELECT ARRAY_AGG(id) AS new_ids FROM {{ target_table }}
        {% if is_incremental() %}
            WHERE {{ loaded_at_col }} > (SELECT MAX({{ loaded_at_col }}) FROM {{ this }})
        {% endif %}
    {% endset %}

    {% set ids_result = run_query(get_ids_array_sql) %}
    {% set new_ids = ids_result.rows[0].new_ids %}

    -- 2. 发送POST请求（以Snowflake的SYSTEM$SEND_HTTP_REQUEST为例）
    {% if new_ids is not none %}
        CALL SYSTEM$SEND_HTTP_REQUEST(
            'https://your-backend-url.com/receive-ids',
            'POST',
            '{"ids": {{ new_ids | tojson }}}',
            PARSE_JSON('{"Content-Type": "application/json"}'),
            NULL
        );
    {% endif %}
{% endmacro %}

3. 正确配置模型的post_hook #

注意post_hook里不需要额外嵌套双引号，直接传递模型引用和字段名参数即可：

{{ config(
    materialized="incremental",
    post_hook="{{ send_to_endpoint(this, 'loaded_at') }}"
) }}

-- 你的增量模型SQL逻辑
SELECT
    id,
    name,
    -- 确保有加载时间字段，用于区分新记录
    CURRENT_TIMESTAMP AS loaded_at
FROM raw.your_source_table
{% if is_incremental() %}
    -- 增量过滤逻辑，只取上次运行后新增的记录
    WHERE loaded_at > (SELECT MAX(loaded_at) FROM {{ this }})
{% endif %}

关键注意事项 #

• 权限配置：确保DBT运行环境能访问后端接口（防火墙放行），数据仓库如果用SQL宏发送请求，需要对应权限（如Snowflake的EXECUTE权限）
• Python环境依赖：用Python宏的话，需要在DBT环境中安装requests库，同时在dbt_project.yml中配置宏路径
• 错误处理：宏中加入状态码检查，避免接口调用失败但DBT任务显示成功的情况
• 性能考虑：如果新记录数量极大，建议分批发送，避免单个请求 payload 过大

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