DBT Artifacts上传模型失败：如何限制未执行模型的重复统计插入？

问题根源 #

默认的dbt_artifacts.upload_results会上传项目中所有模型的统计数据，不管这些模型是否在本次dbt运行中执行过。随着项目里的模型越来越多，插入SQL的长度会逐渐超过BQ标准查询1024K字符的上限，最终触发报错。

解决方案 #

1. 只上传本次实际执行的模型数据

直接在on-run-end钩子中对results变量做过滤，只保留本次运行中实际执行（成功、失败或警告状态）的模型结果：

on-run-end:
  - "{{ dbt_artifacts.upload_results([result for result in results if result.status in ('success', 'failed', 'warn')]) }}"

这个方法最直接，每次只上传本次运行涉及的模型数据，能大幅缩小插入SQL的长度，快速解决查询过大的问题。

2. 只上传有变更的模型（精准过滤）

如果想进一步缩小上传范围，只同步最近有变更的模型数据，可以自定义宏实现：
首先在项目中创建一个自定义宏（比如放在macros/目录下）：

{% macro upload_only_changed_models(results) %}
  {% set filtered_results = [] %}
  {% for result in results %}
    {% set model = result.node %}
    -- 这里可以根据实际情况调整变更判断逻辑，比如对比模型的modified_at或meta字段
    {% if model.get('modified_at') is not none and model.modified_at >= run_started_at - modules.datetime.timedelta(days=1) %}
      {% do filtered_results.append(result) %}
    {% endif %}
  {% endfor %}
  {{ dbt_artifacts.upload_results(filtered_results) }}
{% endmacro %}

然后修改on-run-end配置调用这个宏：

on-run-end:
  - "{{ upload_only_changed_models(results) }}"

注意：不同dbt版本获取模型变更时间的方式可能有差异，你可以根据自己的项目配置调整判断逻辑，比如用模型meta字段里的自定义修改时间。

3. 分批上传（兜底方案）

如果过滤后数据量还是超出BQ限制，可以把结果拆分成多个批次上传：
创建分批上传的自定义宏：

{% macro upload_results_in_batches(results, batch_size=50) %}
  {% for i in range(0, results|length, batch_size) %}
    {% set batch = results[i:i+batch_size] %}
    {{ dbt_artifacts.upload_results(batch) }}
  {% endfor %}
{% endmacro %}

然后配置调用：

on-run-end:
  - "{{ upload_results_in_batches(results, batch_size=50) }}"

这个方法把大的插入查询拆分成多个小查询，每个查询的长度都控制在BQ的限制内，适合模型数量极多的场景。

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