如何在DBT Snowflake环境中使用COPY INTO及创建空表？

一、创建指定结构的空表 #

直接通过DBT模型或宏执行CREATE TABLE语句即可。如果是固定结构的空表，可写在模型文件（如models/stg_my_table.sql）中：

{{ config(materialized='table') }}

CREATE OR REPLACE TABLE {{ target.schema }}.my_target_table (
    id INT,
    name STRING,
    load_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
);

执行dbt run --models stg_my_table即可创建表。如果需要动态创建，也可以封装成宏（见下文）。

二、自定义宏仅输出日志不执行SQL的修复 #

你的宏只调用了log函数打印信息，但没有触发SQL执行逻辑。DBT中执行SQL需要用run_query或execute函数，修复后的宏示例：

{% macro load_data_with_copy_into() %}
    {% set copy_sql %}
        COPY INTO {{ target.schema }}.my_target_table
        FROM @{{ target.schema }}.my_stage/my_data.csv
        FILE_FORMAT = (TYPE = 'CSV' SKIP_HEADER = 1 FIELD_OPTIONALLY_ENCLOSED_BY = '"')
    {% endset %}

    {{ log('Running COPY INTO command: ' ~ copy_sql, info=True) }}
    -- 关键：添加SQL执行语句
    {% do run_query(copy_sql) %}
{% endmacro %}

• run_query：用于执行SQL并返回结果，适合DML/DDL操作；
• execute：也可用于执行SQL，但无返回值，简单场景也能用。

正确调用宏的方式 #

如果是操作宏（不属于模型的独立宏），需要用run-operation命令执行：

dbt run-operation load_data_with_copy_into

如果要在模型中调用宏，直接在模型SQL里写{{ load_data_with_copy_into() }}即可，执行dbt run时会触发。

三、完整流程：空表创建+COPY INTO加载 #

将两个步骤整合到一个宏里，实现端到端数据加载：

{% macro full_data_load_pipeline() %}
    -- 1. 创建指定结构的空表
    {% set create_table_sql %}
        CREATE OR REPLACE TABLE {{ target.schema }}.my_target_table (
            id INT,
            name STRING,
            load_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
        )
    {% endset %}
    {{ log('Creating empty target table...', info=True) }}
    {% do run_query(create_table_sql) %}

    -- 2. 执行COPY INTO加载数据
    {% set copy_sql %}
        COPY INTO {{ target.schema }}.my_target_table (id, name)
        FROM @{{ target.schema }}.my_stage/my_data.csv
        FILE_FORMAT = (TYPE = 'CSV' SKIP_HEADER = 1)
        ON_ERROR = 'CONTINUE'
    {% endset %}
    {{ log('Starting data load with COPY INTO...', info=True) }}
    {% do run_query(copy_sql) %}

    {{ log('Data load pipeline completed successfully.', info=True) }}
{% endmacro %}

执行命令：

dbt run-operation full_data_load_pipeline

四、常见问题排查 #

1. SQL未执行：确认宏中是否调用了run_query/execute，仅log不会触发SQL执行；
2. 权限报错：检查DBT使用的Snowflake角色是否拥有：
   • 创建表的权限（CREATE TABLE on schema）；
   • 读取stage的权限（USAGE on stage + SELECT on stage）；
   • 执行COPY INTO的权限；
3. Stage/文件问题：验证stage路径是否正确，文件是否存在，文件格式定义是否与实际文件匹配；
4. DBT配置错误：检查profiles.yml中Snowflake的account、warehouse、database、schema等配置是否正确。

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