如何将BigQuery存储过程集成到DBT并捕获其输出结果？

1. 封装存储过程调用的Macro #

先写一个Macro统一处理存储过程的调用逻辑，自动适配DBT的模型引用格式：

{% macro call_my_procedure(table_ref, column_name) %}
  CALL `project_id.my_procedure`({{ table_ref }}, '{{ column_name }}');
{% endmacro %}

使用时直接传DBT的模型引用（比如{{ ref('your_source_model') }}），DBT会自动解析成BigQuery能识别的完整标识符（项目.数据集.表名）。

2. 捕获临时表输出的两种方式 #

因为你的存储过程最后会返回临时表结果，核心要保证在同一个会话里完成调用和查询（BigQuery会话级临时表仅在当前会话可见），具体有两种实现方式：

方式一：单模型内直接调用+查询 #

新建一个DBT模型文件（比如models/processed_procedure_results.sql），内容如下：

-- 调用存储过程，传入DBT模型和目标列
{{ call_my_procedure(ref('your_source_model'), 'target_column') }}

-- 查询存储过程生成的临时表，注意表名要和存储过程里定义的完全一致
SELECT * FROM `your_temp_table_name`;

DBT会把这两个语句放在同一个作业里执行，自然能访问到会话级临时表。

方式二：Macro封装完整逻辑 #

如果临时表名需要动态指定，或者要复用查询逻辑，把调用和查询都封装进Macro：

{% macro get_procedure_results(table_ref, column_name, temp_table_name) %}
  {{ call_my_procedure(table_ref, column_name) }}
  SELECT * FROM `{{ temp_table_name }}`;
{% endmacro %}

然后在模型里直接调用：

{{ get_procedure_results(ref('your_source_model'), 'target_column', 'your_temp_table') }}

3. 关键注意事项 #

• 确认存储过程生成的是会话级临时表（BigQuery默认临时表前缀为#，或无前缀的会话临时表）；如果是全局临时表（前缀_global_），则无需担心会话问题，但要注意命名冲突。
• 确保执行DBT的服务账号拥有日志表的插入权限，以及临时表的查询权限。
• 测试时单独运行目标模型：dbt run --models processed_procedure_results，检查日志表记录和模型输出是否符合预期。

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