如何在DBT中实现Snowflake Task及执行COPY INTO命令

DBT本身不内置调度能力，通常通过以下两种方式实现每日自动刷新：

• Snowflake Task配合DBT：创建Snowflake Task，在Task中调用DBT运行命令（比如dbt run --models <目标模型名>）。可以把DBT命令封装成Snowflake存储过程，或者通过外部调度工具（如Airflow）触发DBT运行，再由Snowflake Task衔接后续操作。
• DBT Cloud原生调度：如果使用DBT Cloud，直接在平台作业调度模块设置每日运行计划，指定要执行的模型或完整作业流，DBT Cloud会自动按时间触发运行，无需手动维护Snowflake Task。
• 优化建议：把全量刷新逻辑改成DBT增量模型（通过{{ config(materialized='incremental') }}配置），只更新新增或变化的数据，提升运行效率。

通过自定义DBT模型即可直接实现，步骤如下：

1. 创建DBT模型文件（比如load_stage_data.sql），配置模型物化方式为table（按需也可选择incremental）：

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

-- 如需清空表后再加载，可添加TRUNCATE语句
TRUNCATE TABLE {{ this }};

COPY INTO {{ this }}
FROM @your_stage_name/path_to_data_files
FILE_FORMAT = (TYPE = CSV FIELD_DELIMITER = ',' FIELD_OPTIONALLY_ENCLOSED_BY = '"')
ON_ERROR = CONTINUE;

• {{ this }}是DBT内置变量，指代当前模型对应的Snowflake表，无需手动写死表名。
• 根据实际数据源格式调整FILE_FORMAT参数（比如JSON、Parquet等）。

2. 确保DBT使用的Snowflake角色拥有读取Stage和写入目标表的权限。
3. 运行dbt run --models load_stage_data即可触发数据加载，结合前面的调度方案实现每日自动运行。

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