如何避免dbt中COPY INTO宏重复执行时的报错问题？

我有一个dbt宏用于将数据通过COPY INTO导出到Snowflake的S3外部阶段，代码如下：

{% macro my_macros() %}

  {%- if not execute or target.name != 'prod' -%}
    {{ return('') }}
  {%- endif %}

  {% set query = 'USE DATABASE MY_DB; USE SCHEMA MY_SCH; COPY INTO @MY_STAGE/my_table FROM (SELECT OBJECT_CONSTRUCT(*) from MY_TABLE) FILE_FORMAT =(TYPE = JSON COMPRESSION = NONE) OVERWRITE=FALSE;' %}
  {# dbt_utils.log_info(query) #}
  {%- do run_query(query) -%}
{% endmacro %}

设置OVERWRITE=FALSE是为了确保文件只被复制一次，但重复执行dbt run时会抛出错误：

001030 (22000): Files already existing at the unload destination: @MY_STAGE/my_table. Use overwrite option to force unloading.

以下是四种可行的解决方案：

方案1：在宏内捕获并忽略特定错误 #

通过Jinja的try-except块捕获目标文件已存在的特定错误，仅忽略该错误，其他异常正常抛出：

{% macro my_macros() %}
  {%- if not execute or target.name != 'prod' -%}
    {{ return('') }}
  {%- endif %}

  {% set query = 'USE DATABASE MY_DB; USE SCHEMA MY_SCH; COPY INTO @MY_STAGE/my_table FROM (SELECT OBJECT_CONSTRUCT(*) from MY_TABLE) FILE_FORMAT =(TYPE = JSON COMPRESSION = NONE) OVERWRITE=FALSE;' %}
  
  {% try %}
    {%- do run_query(query) -%}
  {% except as e %}
    {% if '001030' in e.message or 'Files already existing at the unload destination' in e.message %}
      {{ log("目标路径已有文件，跳过COPY INTO执行", info=True) }}
    {% else %}
      {{ exceptions.raise_compiler_error(e.message) }}
    {% endif %}
  {% endtry %}
{% endmacro %}

该方案精准控制错误范围，不会掩盖其他潜在问题。

方案2：先检查阶段文件存在性，再决定是否执行 #

在执行COPY INTO前，用LIST命令检查目标阶段是否已有文件，仅当文件不存在时才执行导出：

{% macro my_macros() %}
  {%- if not execute or target.name != 'prod' -%}
    {{ return('') }}
  {%- endif %}

  {% set check_query = 'LIST @MY_STAGE/my_table;' %}
  {% set results = run_query(check_query) %}
  
  {% if results.rows | length == 0 %}
    {% set copy_query = 'USE DATABASE MY_DB; USE SCHEMA MY_SCH; COPY INTO @MY_STAGE/my_table FROM (SELECT OBJECT_CONSTRUCT(*) from MY_TABLE) FILE_FORMAT =(TYPE = JSON COMPRESSION = NONE) OVERWRITE=FALSE;' %}
    {%- do run_query(copy_query) -%}
    {{ log("成功执行COPY INTO，导出数据到阶段", info=True) }}
  {% else %}
    {{ log("目标阶段已有文件，跳过导出操作", info=True) }}
  {% endif %}
{% endmacro %}

从根源避免错误触发，逻辑更直观。

方案3：仅屏蔽该宏/关联模型的错误 #

如果该宏是作为模型的post-hook执行的，可在模型配置中设置忽略执行失败：

# dbt_project.yaml
models:
  your_project_name:
    your_target_model:
      post-hook:
        - "{{ my_macros() }}"
      on-run-fail: "ignore"

注意：此配置仅针对该模型的钩子执行错误生效，不会影响其他任务的错误抛出。

方案4：配置宏每日仅执行一次 #

方法A：利用dbt运行记录控制 #

通过查询dbt的run_results表，判断该宏是否在最近24小时内成功执行过，若已执行则跳过：

{% macro my_macros() %}
  {%- if not execute or target.name != 'prod' -%}
    {{ return('') }}
  {%- endif %}

  {% set check_execution_query %}
    SELECT 1 
    FROM {{ ref('dbt_run_results') }}
    WHERE status = 'success'
      AND run_started_at >= DATEADD(day, -1, CURRENT_TIMESTAMP())
      AND model_name = 'my_macros'
  {% endset %}

  {% set has_run = run_query(check_execution_query) %}
  
  {% if has_run.rows | length == 0 %}
    {% set copy_query = 'USE DATABASE MY_DB; USE SCHEMA MY_SCH; COPY INTO @MY_STAGE/my_table FROM (SELECT OBJECT_CONSTRUCT(*) from MY_TABLE) FILE_FORMAT =(TYPE = JSON COMPRESSION = NONE) OVERWRITE=FALSE;' %}
    {%- do run_query(copy_query) -%}
    {{ log("今日首次执行COPY INTO，导出数据完成", info=True) }}
  {% else %}
    {{ log("今日已执行过该宏，跳过操作", info=True) }}
  {% endif %}
{% endmacro %}

需确保dbt_run_results表已创建（可通过dbt官方工具或自定义模型生成）。

方法B：外部调度工具控制 #

使用Airflow、Prefect等调度工具，配置每日仅触发一次包含该宏的dbt任务，从外部层面控制执行频率，无需修改宏代码。

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