如何将 dbt 与 Snowflake 配合使用：一份全面指南

随着现代数据团队采用日益复杂的分析技术栈，数据工程工作流正逐步向建立标准实践的方向转变。像 dbt（Data Build Tool）这样的工具是一种开源框架，将软件工程的最佳实践（如版本控制、测试与文档）引入到分析工作流中。

dbt 与 Snowflake（云原生数据仓库）的组合，已成为分析工程工作流的基石。本指南将探讨将 dbt 与 Snowflake 集成的基础概念、搭建方法与进阶实施策略，帮助数据团队构建更易维护、更安全且性能更佳的数据管道。

如果您刚开始接触这些工具，建议先学习我们的Snowflake 基础技能路径以及dbt 入门课程。

dbt 与 Snowflake：基础概念与优势

先聊聊 dbt 与 Snowflake 如何协同工作。本节将介绍 dbt 的能力、其与 Snowflake 的兼容性，以及两者集成带来的益处。

什么是 dbt（Data Build Tool）？

dbt 是一款开源框架，使数据团队能够在数据仓库中直接使用 SQL（结合 Jinja 模板）进行数据转换、测试与文档编写。

dbt 本身不是一种编程语言；它充当编译器，将模块化的 SQL 模型转换为可在仓库（如 Snowflake）中运行的可执行 SQL 查询。

通过引入版本控制、自动化测试和文档，dbt 将成熟的软件工程实践带入分析工程，使其作为 ELT 工作流一部分的转换管道更可靠、可维护且具备可扩展性。

其核心组件包括编译器与运行器。功能如下：

• 编译器：dbt 用户以 Jinja 语言编写代码，编译为原生 SQL。
• 运行器：将编译后的代码对接数据仓库（如 Snowflake）执行以获取结果。

为保证 dbt 正常运行，还需要一些底层组件来协同配合：

• 模型：核心的 select 语句，以 .sql 文件保存。
• 种子（Seeds）：.csv 文件，dbt 可将其加载到数据仓库，作为模型的参考数据。
• 源（Sources）：仓库中需要读取的数据来源。
• 数据测试（Data Tests）：用于根据规则校验数据的 .sql 文件。
• 文档（Documentation）：dbt 支持文档站点，可展示模型的依赖树。

关于 dbt 的更多细节，建议参考这门优秀的dbt 入门课程。

dbt 与 Snowflake 集成概览

结合 dbt 的基于 Jinja 的模板与 Snowflake 强大的数据类型（如 Streams），我们可以在转换工作流上做出许多灵活设计。

例如，您可能只想转换新加载的数据，或针对分层表与生产表采用不同的转换，以便进行数据健康检查。

二者正迅速成为常见搭配，随着越来越多公司采用 dbt，能帮助数据工程师在数据管道中获得更强的可控性与灵活性。

使用 dbt 的优势

dbt 的优势很多：传统软件开发方法论、自动化测试与效率提升。

代码驱动开发

我们可以像对待传统代码一样对待数据与 SQL，从而获得模块化、可测试与可版本控制的能力。借助模板与 Python 风格的流程控制语句，dbt 带来强大功能。

自动化测试

使用 dbt 内置的 data_test 配置，我们可以定期对数据运行测试，无需构建繁琐的 SQL 管道。事实上，dbt 已内置了一些强大的基础测试，我们只需在配置文件中轻松启用即可。

效率提升

dbt 内含文档生成器与管道可视化工具。新成员可以更快上手，现有成员也能更深入理解数据流。这有助于缩短熟悉期并提升全员对工作流的整体理解。

使用 Snowflake 的优势

Snowflake 的云平台优势众多。其架构将存储与计算资源分离，减少资源冲突、控制成本。它支持零拷贝克隆，使用户可在不增加存储的情况下克隆数据库。

此外，它支持一些复杂的表类型，如动态表与 Streams，可用于增量刷新逻辑。关于 Snowflake 平台的更多细节，请参阅这篇介绍Snowflake 如何运作的文章。

最重要的是，这些特性与 dbt 高度互补。我们可以利用增量刷新来让 dbt 基于新数据触发转换；也可利用 Snowflake 存储与计算分离的特性，让 dbt 执行更高强度的计算任务而不影响数据的上传与存储能力。

此外，Snowflake 具备快速扩展能力与按需付费成本模型。随着 dbt 脚本复杂度上升，我们可以轻松扩大 Snowflake 计算仓库以应对需求。同时，借助 Snowflake 的监控能力来精简仓库规模，并用 dbt 优化查询性能。

dbt 与 Snowflake 的关键协同

结合使用时，Snowflake 与 dbt 相得益彰。Snowflake 的持续调优有助于降低云端成本；dbt 可利用如聚簇键等分区方法与 Snowflake 的缓存以降低数据开销。

Snowflake 本身不易通过 Git 等工具直接管理。由于 dbt 基于与其他编程语言相同的原则，我们可以使用基于 Git 的工作流（如 CI/CD 管道）更轻松地管理开发过程。这有助于对管道进行版本控制，而如果完全在 Snowflake 中实现则更具挑战。

为更便于开发者协作，dbt 还提供 UI，将所有 dbt 作业集中展示，帮助团队共同跟踪流程并确保其正确运行。此外，dbt 近期推出了 Studio IDE，便于结合 Git 与数据库进行测试工作流集成。一些团队也可能选择使用 Snowflake 的 UI 做全面集成。

许多数据团队已经开始同时使用 dbt 与 Snowflake，以最大化数据库的价值。继续学习 dbt，请查看这篇面向数据工程师的 dbt 动手入门，并通过这门课程了解Snowflake 数据建模如何运作。

来自 Snowflake Quickstarts：借助 dbt Core 与 Snowflake 加速

dbt 与 Snowflake 的实施与配置

我们来讲解如何配置 Snowflake 与 dbt 以实现正确集成，同时涵盖项目结构设计的一些最佳实践。

前置条件与搭建

我们将从 Snowflake 数据库的设置开始，然后介绍 dbt 的设置。

Snowflake 前置条件

您需要一个专用的数据库、架构与虚拟仓库来存放 dbt 数据。除此之外，您还需要创建一个 dbt 专用的用户角色供 dbt 访问。在设置 dbt 时，您将提供该用户的凭据，以便其与 Snowflake 通信。本节将给出一些基本指引；关于如何管理 Snowflake 数据库与用户角色的更多细节，请学习这门Snowflake 入门课程。

1. 创建专用数据库、架构与虚拟仓库

	CREATE DATABASE analytics_db;
	CREATE SCHEMA analytics_db.transformations;
	CREATE WAREHOUSE dbt_wh
	  WITH WAREHOUSE_SIZE = 'XSMALL'
	  AUTO_SUSPEND = 60
	  AUTO_RESUME = TRUE
	  INITIALLY_SUSPENDED = TRUE;
	技术支持

2. 创建专用的 dbt 角色与用户。这对于允许 dbt 访问您的 Snowflake 服务器至关重要

	CREATE ROLE dbt_role;
	GRANT USAGE ON WAREHOUSE dbt_wh TO ROLE dbt_role;
	GRANT USAGE ON DATABASE analytics_db TO ROLE dbt_role;
	GRANT USAGE, CREATE SCHEMA ON DATABASE analytics_db TO ROLE dbt_role;
	GRANT ALL PRIVILEGES ON SCHEMA analytics_db.transformations TO ROLE dbt_role;

	CREATE USER dbt_user PASSWORD='StrongPassword123'
	  DEFAULT_ROLE = dbt_role
	  DEFAULT_WAREHOUSE = dbt_wh
	  DEFAULT_NAMESPACE = analytics_db.transformations;
	GRANT ROLE dbt_role TO USER dbt_user;
	技术支持

3. 最后，查找您的 Snowflake 账户名。我们在稍后设置 dbt 时会用到。这可以在 Snowflake 的设置与账户详情中找到。更多细节请参阅Snowflake 文档。

dbt 配置

要运行 dbt，我们需要使用 pip 安装它，并配置 profiles.yml 文件。

1. 在您的 Python 环境中安装 dbt。您需要同时安装 dbt-core 与 dbt-snowflake 包

pip install dbt-core dbt-snowflake技术支持

2. 配置 profiles.yml 文件，并保存到 ~/.dbt/profiles.yml

snowflake-db:
  target:dev
  outputs:
    dev:
      type: snowflake
      account: [account_id from earlier]
      user: [dbt_user]
      password: [dbt_pw]
      role: [dbt_role]
      warehouse:[dbt_warehouse]
      database: [dbt_database]
      schema: [dbt_schema]
      threads: 4
      client_session_keep_alive: False
      query_tag: [anything]技术支持

3. 初始化一个 dbt 项目

dbt init --profile snowflake-db技术支持

4. （备用初始化方式）如果您不想预先配置 YAML 文件，也可以通过以下命令进入 dbt 的初始化步骤。

dbt init snowflake-profile-name技术支持

这会引导您逐步完成初始化，并在需要时生成 profiles.yml 文件！

5. 为确保连接正确，我们可以使用 debug 功能：

dbt debug --connection技术支持

该命令会检查您的连接设置是否有效，如无效将抛出错误。

需要注意的是：与其将用户名与密码直接写在 profiles.yml 文件中，不如使用环境变量。例如，可将 dbt_password 保存到 SNOWFLAKE_PASSWORD 环境变量中；然后在 profiles.yml 中写 password: “{{env_var(‘SNOWFLAKE_PASSWORD’)}}”。这样可以防止他人从配置文件中看到凭据，增加一层安全性。

版本控制设置

连接 Git 有助于最大化 dbt 配置的价值。若在企业环境中实施，具体步骤可能有所不同，但总体思路相同：在 GitHub 或 GitLab 上创建代码库，然后将您的 dbt 账户连接到 GitHub/GitLab 账户。完成后即可 git clone 您的代码库！关于将 Git 连接至 dbt 的更多信息，请查阅dbt 的 Git 文档以适配您的平台。

项目结构设计

一致化、模板化的项目结构能为新项目提供清晰的归置位置，加快启动速度。

组织模型

将项目按 staging、intermediate 与 marts 分文件夹组织。在每个层级内，可按部门细分并维护各自的模型与 SQL。

	models/
	│   ├── intermediate
	│   │   └── finance
	│   │       ├── _int_finance__models.yml
	│   │       └── int_payments_pivoted_to_orders.sql
	│   ├── marts
	│   │   ├── finance
	│   │   │   ├── _finance__models.yml
	│   │   │   ├── orders.sql
	│   │   │   └── payments.sql
	│   │   └── marketing
	│   │       ├── _marketing__models.yml
	│   │       └── customers.sql
	│   ├── staging
	│   │   ├── jaffle_shop
	│   │   │   ├── _jaffle_shop__docs.md
	│   │   │   ├── _jaffle_shop__models.yml
	│   │   │   ├── _jaffle_shop__sources.yml
	│   │   │   ├── base
	│   │   │   │   ├── base_jaffle_shop__customers.sql
	│   │   │   │   └── base_jaffle_shop__deleted_customers.sql
	│   │   │   ├── stg_jaffle_shop__customers.sql
	│   │   │   └── stg_jaffle_shop__orders.sql
	│   │   └── stripe
	│   │       ├── _stripe__models.yml
	│   │       ├── _stripe__sources.yml
	│   │       └── stg_stripe__payments.sql
	│   └── utilities
	│       └── all_dates.sql/技术支持

models 下各文件夹的用途如下：

• Staging：清洗并重命名的原始数据。
• Intermediate：包含较重逻辑的连接或转换。
• Marts：面向业务的最终表。

对于 staging、intermediate 与 marts 的模型配置文件，您可能希望设置 +materialized 以定义 SQL 在 Snowflake 中的生成方式。可选择视图、表或增量数据集：

• View：默认，轻量。
• Table：具备缓存，但占用更多存储。
• Incremental：适用于大规模且频繁追加的数据集。

配置示例如下：

# The following dbt_project.yml configures a project that looks like this:
# .
# └── models
#     ├── csvs
#     │   ├── employees.sql
#     │   └── goals.sql
#     └── events
#         ├── stg_event_log.sql
#         └── stg_event_sessions.sql

name: my_project
version: 1.0.0
config-version: 2

models:
  my_project:
    events:
      # materialize all models in models/events as tables
      +materialized: table
    csvs:
      # this is redundant, and does not need to be set
      +materialized: view技术支持

高级优化与宏

将 dbt 与 Snowflake 集成的一大优势是可同时获得两端的优化。对 Snowflake 而言，我们可以通过多种方式处理查询以加速运行；对 dbt 而言，我们可以通过宏与物化策略来优化代码。

性能调优技巧

我们可以利用一些 Snowflake 独特功能来提升查询性能。

查询优化

在 Snowflake 中，两种重要的优化方式是谓词下推与聚簇键。

谓词下推的思想是改变过滤的执行时机，以减少 Snowflake 需要读取的数据量。比如，我们的查询有两层过滤：一层是安全过滤，一层是分类过滤。

安全过滤会根据安全角色定义可见的数据范围；分类过滤则在查询的 WHERE 子句中。Snowflake 会选择先执行能读取最少数据的过滤条件。

为辅助谓词下推，我们可以考虑使用聚簇键。虽然 Snowflake 在表聚簇方面表现不错，但当数据达到 TB 级时，维持合理分区对 Snowflake 来说也会变得困难。

通过定义聚簇键，我们可以帮助 Snowflake 按照常见的查询方式来分区。例如，我们经常同时筛选两个字段，如日期与物业类型。若每种物业类型对应的日期数量足够小，那么重新分区就更合理，数据定位也更高效。

将聚簇键与 Snowflake 的谓词下推优化结合使用，可以显著提升在大表中的查询效率。

仓库扩缩容

在 Snowflake 这样的按需计费系统中，资源使用非常重要！请遵循以下简单原则：

• 使用 Snowflake 的 auto-suspend 在管道长时间离线时自动挂起仓库
• 在间隔一段时间后再次访问仓库时，auto-resume 可按需自动恢复仓库
• 开发环境使用 XS 或 S 级仓库，生产环境按需扩容

这些简单的仓库设置，在长期内能让您的 Snowflake 使用更高效。

动态表 vs 增量模型

在 dbt 端，我们可以利用动态表与增量模型来简化更新逻辑。

动态表是 Snowflake 的一种表类型，可按设定的滞后时间自动刷新。dbt 能将动态表作为物化类型进行配置，并以类似于在 Snowflake 中的方式进行设置。

models:
  <resource-path>:
    +materialized: dynamic_table
    +on_configuration_change: apply | continue | fail
    +target_lag: downstream | <time-delta>
    +snowflake_warehouse: <warehouse-name>
    +refresh_mode: AUTO | FULL | INCREMENTAL
    +initialize: ON_CREATE | ON_SCHEDULE技术支持

这对于构建不需要复杂 SQL 的简易数据传输管道非常方便。最棒的是，我们可以按需连接这些动态表并获取最新数据。

另一种灵活构建表的方法是使用增量模型。在编写 SQL 时，我们基于某列筛选新数据；然后利用 is_incremental() 宏，让 dbt 在将模型配置为增量物化时，仅应用该筛选条件。

示例 SQL 如下，摘自dbt 文档：

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

select
    *,
    my_slow_function(my_column)

from {{ ref('app_data_events') }}

{% if is_incremental() %}

  -- this filter will only be applied on an incremental run
  -- (uses >= to include records whose timestamp occurred since the last run of this model)
  -- (If event_time is NULL or the table is truncated, the condition will always be true and load all records)
where event_time >= (select coalesce(max(event_time),'1900-01-01') from {{ this }} )

{% endif %}技术支持

用宏复用逻辑

前面我们提到了“宏”。宏是在 dbt 中可复用的代码片段，可用于重复的计算或高频出现的转换。

Snowflake 专用宏

我们使用 Jinja 模板来创建宏。一个常见用例是在处理数据前克隆数据库。

	{% macro clone_tables(table_to_clone) -%}

	–- shows all tables within a schema
	    {% set all_tables_query %}
	        show tables in schema {{ clone_tables }}
	    {% endset %}

	-- take the set results and use the run_query macro
	    {% set results = run_query(all_tables_query) %}

	    {{ "create or replace schema " ~ generate_schema_name(var("custom_tables_list")) ~ ";" }}

	--execute the cloning function
	    {% if execute %}
	        {% for result_row in results %}
	            {{ log("create table " ~ generate_schema_name(var("custom_tables_list")) ~ "." ~ result_row[1] ~ " clone " ~ clone_tables~ "." ~ result_row[1] ~ ";") }}
	            {{ "create table " ~ generate_schema_name(var("custom_tables_list")) ~ "." ~ result_row[1] ~ " clone " ~ clone_tables~ "." ~ result_row[1] ~ ";" }}
	        {% endfor %}
	    {% endif %}
	{%- endmacro %}技术支持

该模板接收指定架构，生成该架构下的表清单，并将其克隆到 dbt 配置中指定的仓库/架构。随后，您可以在 dbt 代码或启动 dbt 时调用该宏。

想进一步练习 dbt 中的 Jinja，强烈推荐这份基于 dbt 构建电商数据模型的案例研究，它涵盖复杂的 Jinja 模板并提供高级动手实践！

用 dbt 屏蔽 PII：

在处理客户信息时，可能会接触到可识别个人身份信息（PII），需要在数据库中进行隐藏。借助 Snowflake，这很简单！我们可以使用 CREATE MASKING POLICY 告知 Snowflake 在查询时需要屏蔽哪些列与数据类型。这些策略可确保在查询敏感信息时不会泄露 PII。

随后，我们可以结合 dbt 的 tests 与自定义 SQL 来验证表是否正确屏蔽。例如，若定义的屏蔽策略仅允许 ANALYST 角色查看 PII，如下所示：

	CREATE OR REPLACE MASKING POLICY email_mask AS (val string) returns string ->
	  CASE
	    WHEN current_role() IN ('ANALYST') THEN VAL
	    ELSE '*********'
	  END;技术支持

那么，使用 dbt_account 用户角色查询数据时应返回 '*********' 这样的字符串。

若我们的测试像这样简单：

	SELECT *
	FROM schema.table
	WHERE email <> '*********'技术支持

如果该测试返回结果，则应视为失败。任何邮件地址都不应返回为非屏蔽字符串。接下来应检查 Snowflake 上的屏蔽策略，并持续测试直至正常。

安全与治理

如往常一样，安全是云访问的关键支柱。本节将介绍安全与数据保护的一些基础内容。

访问控制框架

Snowflake 的主要安全框架是基于角色的访问控制（RBAC）与服务账号的使用。

基于角色的访问控制（RBAC）

定义 RBAC 权限使我们可以对一大类用户统一应用安全设置，而无需逐个分配，便于维护。

如前所述，我们使用 CREATE ROLE 创建角色。授予权限使用 GRANT <privilege> … TO ROLE，撤销权限则用 REVOKE <privilege> … FROM ROLE。

以下是一些最佳实践：

• 坚持最小权限原则。例如，仅为 dbt_reader 赋予读权限，为 dbt_writer 赋予建表权限，而删除权限仅授予 dbt_admin 等管理员账号。
• 可考虑按环境（开发、分层、生产）划分角色访问范围。

服务账号

服务账号是无需人工交互的机器账号。这类账号没有用户名与密码，而是使用公钥与网络限制来降低风险。

创建服务账号可能如下所示：

CREATE USER dbt_service
RSA_PUBLIC_KEY = <keysring>技术支持

这样，您拥有一个由管理员统一管理的账号，无需担心个人用户问题。建议定期轮换密钥以防泄露。同时，使用 Snowflake 的 LOGIN_HISTORY 密切关注网络活动。

数据保护

保护您的数据与登录信息。虽然示例中我们曾硬编码凭据，但实际中这并非良好做法。请使用 AWS Secrets Manager 或 GitHub Secrets 等密钥管理工具来保存重要凭据。

它们允许您在线存储凭据，仅对拥有合适 RSA/安全密钥的用户可见。任何人都看不到真实值，且通过其自动化管理可轻松轮换密码/密钥。

确保所有连接均已妥善加固，且用户通过合规的企业网络访问数据。这通常通过如 GlobalProtect 的 VPN 实现。由于我们连接的是 Snowflake，端到端加密大多由云端处理——这也是使用 Snowflake 的一大优势！

最后，我们还要审计活动。Snowflake 通过 QUERY_HISTORY 与 QUERY_HISTORY_BY_USER 等表记录查询日志。我们可以在这些日志中监控使用情况、错误与安全滥用。借助 dbt，我们还可以通过版本控制管道跟踪新运行、变更与项目编辑。

版本控制与代码模板有助于维持变更跟踪与合规性。通过强制的 Pull Request，可在问题成为隐患前轻松发现并处理变更引入的问题。

监控与成本管理

聊聊如何省钱——在 dbt 中当然可以。监控模型性能至关重要，随着表规模增大，曾经流畅运行的模型也可能变慢、性能下降。

性能监控

在 dbt 中，最佳监控方式是使用内置包如 dbt-snowflake-monitoring，或借助 Looker、Datafold 等外部工具进行可视化与数据对比。

dbt-snowfake-monitoring 的用法

开始使用 dbt-snowflake-monitoring 非常简单！在 dbt 的 packages.yml 中添加如下内容：

packages:
  - package: get-select/dbt_snowflake_monitoring
    version: [">=5.0.0", "<6.0.0"] # We'll never make a breaking change without creating a new major version.技术支持

然后，在每个项目的 YAML 文件中添加如下内容：

dispatch:
  - macro_namespace: dbt
    search_order:
      - <YOUR_PROJECT_NAME>
      - dbt_snowflake_monitoring
      - dbt
query-comment:
  comment: '{{ dbt_snowflake_monitoring.get_query_comment(node) }}'
  append: true # Snowflake removes prefixed comments.技术支持

现在，您就可以使用该包来监控仓库使用、查询时长与失败模型等！它会利用 dbt 用户现有的访问权限，将包含有用信息的表写入 Snowflake。例如，若想查看每月仓库支出，可使用如下 SQL：

select
    date_trunc(month, date)::date as month,
    warehouse_name,
    sum(spend_net_cloud_services) as spend
from daily_spend
where service in ('Compute', 'Cloud Services')
group by 1, 2技术支持

外部工具

Looker 或 Tableau 等外部工具可直接与 dbt 集成，实现数据与性能的便捷可视化。对于告警，dbt 可以发送邮件或 Slack 通知。

对于更深入的对比，Datafold 等外部技术栈可提供变更对数据质量影响的更深洞察。Datafold 会在每次代码变更时自动进行数据差异对比，使团队成员更易审阅您的代码与其对数据的影响。这类自动回归测试让我们无需不断构建新检查与工具，也能维持高质量数据。

成本优化技巧

我们无法避免运行 ETL 流程，但可以提前规划，尽量降低计算与存储成本的影响。得益于 Snowflake 的分离式设计，我们可以分别优化仓库与存储，以实现更灵活的成本优化。

就仓库而言，最佳方式是尽量减少空闲时间。尽可能分散自动化管道的运行，以降低空闲时间、计算需求峰值与资源竞争。

定期使用 WAREHOUSE_LOAD_HISTORY 与 QUERY_HISTORY 表进行监控，了解资源使用情况。最佳实践是从 XS 开始；若查询过慢，再根据团队 SLA 进行扩容。

就存储而言，最佳实践是移除不必要的表。虽然 Snowflake 本身不像 AWS 与 Google 那样有冷存储分层，但您可以选择将数据从 Snowflake 迁移到 AWS/Google 的冷存储层以降低成本。

数据保留策略可清晰规定很少使用的表的时间线以及何时迁移至冷存储。

对于频繁增长的大表，要密切监控。随着规模增大，Snowflake 自动高效分区的能力会下降。利用上文提到的自动聚簇，帮助 Snowflake 更好地为数据分区。更好的分区意味着更少的存储读取时间，从而节省成本。

CI/CD 与自动化

得益于 dbt 原生集成版本控制，实现 CI/CD 与自动化管道轻而易举。我们将介绍一些自动化编排选项与测试最佳实践。

管道编排

在 dbt Cloud 中，我们可以使用 Snowflake Tasks 来触发 dbt 项目的定时运行。其工作方式是：先通过 Git 代码库在 Snowflake 中创建一个 dbt 项目，然后创建一个任务执行该 dbt 项目。

例如，我们可以如下创建一个 dbt 项目：

CREATE DBT PROJECT sales_db.dbt_projects_schema.sales_model
  FROM '@sales_db.integrations_schema.sales_dbt_git_stage/branches/main'
  COMMENT = 'generates sales data models';技术支持

然后用以下代码运行该项目：

CREATE OR ALTER TASK sales_db.dbt_projects_schema.run_dbt_project
  WAREHOUSE = sales_warehouse
  SCHEDULE = '6 hours'
AS
  EXECUTE DBT PROJECT sales_db.dbt_projects_schema.sales_model args='run --target prod';技术支持

实际上，它非常完备，我们甚至可以随后运行测试，确保项目执行正确：

CREATE OR ALTER TASK sales_db.dbt_projects_schema.test_dbt_project
        WAREHOUSE = sales_warehouse
        AFTER run_dbt_project
AS
  EXECUTE DBT PROJECT sales_db.dbt_projects_schema.test_dbt_project args='test --target prod';技术支持

关于调度，我们可以设置固定时间（如 60 MINUTES 与 12 HOURS），也可使用 Cron，如 USING CRON 0 9 * * * UTC。

不过我们的模型很少保持静态，经常需要迭代。手动构建管道的每个环节很耗时。这时就需要 GitHub Actions。我们常在 CI/CD 管道中使用 GitHub Actions 来自动化测试与构建代码。

GitHub Actions 通过仓库内 .github/workflows 文件夹中的 YAML 文件定义。我们可以设置在发起 Pull Request 时触发，以运行 dbt 测试。

name: dbt pull test

# CRON job to run dbt at midnight UTC(!) everyday
on:
  pull_request:

types:
openedreopened
# Setting some Env variables to work  with profiles.yml
# This should be your snowflake secrets
env:
  DBT_PROFILE_TARGET: prod
  DBT_PROFILE_USER: ${{ secrets.DBT_PROFILE_USER }}
  DBT_PROFILE_PASSWORD: ${{ secrets.DBT_PROFILE_PASSWORD }}

jobs:
  dbt_run:
    name: dbt testing on pull request
    runs-on: ubuntu-latest
    timeout-minutes: 90
    
    # Steps of the workflow:
    steps:
      - name: Setup Python environment
	uses: actions/setup-python@v4
	with:
	  python-version: "3.11"

      - name: Install dependencies
	run: |
          python -m pip install --upgrade pip
	  python -m pip install -r requirements.txt
	
      - name: Install dbt packages
	run: dbt deps
        # optionally use this parameter
        # to set a main directory of dbt project:
        # working-directory: ./my_dbt_project
	    		
      - name: Run tests
	run: dbt test
        # working-directory: ./my_dbt_project技术支持

妥善设置自动化能让您的工作轻松许多，也便于其他开发者查看已测试的模型，从而通过减少手动测试来简化开发流程。

测试框架

dbt 自带强大的测试套件，包含一些即开即用的通用数据测试，可直接在您的模型上配置，并允许自定义 SQL 逻辑。通用测试针对模型中的特定列，检查这些列是否通过相应测试。内置测试包括：

• unique：该列的每个值都应唯一
• not_null：该列不应有空值
• accepted_values：检查值是否在可接受值列表中
• relationships：检查该列的值是否存在于另一相关表中

以下是这些关系测试的示例（来自 dbt 测试文档）：

version: 2

models:
  - name: orders
    columns:
      - name: order_id
        data_tests:
	# makes sure this column is unique and has no nulls
          - unique
          - not_null
      - name: status
        data_tests:
	#makes sure this column only has the below values
          - accepted_values:
              values: ['placed', 'shipped', 'completed', 'returned']
      - name: customer_id
        data_tests:
	# makes sure that the customer ids in this column are ids in the customers table
          - relationships:
              to: ref('customers')
              field: id技术支持

我们还可以使用自定义 SQL 创建更具体与通用的测试。目标是让测试返回失败行，也就是说只要有结果返回，测试就视为失败。这些自定义 SQL 文件放在 tests 目录中。

例如，若我们想知道是否有任何销售额为负，可以使用如下语句：

	select
	    order_id,
	    sum(amount) as total_amount
	from {{ ref('sales') }}
	group by 1
	having total_amount < 0技术支持

随后在测试文件夹内的 schema.yml 中引用该测试：

version: 2
data_tests:
  - name: assert_sales_amount_is_positive
    description: >
    Sales should always be positive and are not inclusive of refunds
技术支持

如果我们希望在更多表与列上使用此类检查，可以将其做成通用测试。示例如下：

{% test negative_values(model, group,sum_column) %}

    select {{group}},
    sum({{sum_column}}) as total_amt
    from {{ model }}
    group by 1
    having total_amt < 0
{% endtest %}技术支持

现在，我们就可以像使用 dbt 自带的其他通用测试那样，在模型的 yaml 文件里使用该测试了！您可以检查行程中是否存在负乘客数、负利润等，可能性几乎无穷无尽。

为让测试更顺畅与简单，我们可以利用 Snowflake 的零拷贝克隆。使用 Snowflake 的 CREATE TABLE <object> CLONE <source_object>，即可在开发/测试环境创建零拷贝克隆。

接着，所有测试与 CI 都可对克隆对象进行。其好处是无需额外存储空间即可拥有安全的测试环境；且当需要更多测试空间时，可通过即时克隆轻松创建更多测试环境。

新兴趋势与未来展望

让我们展望 dbt 与 Snowflake 的未来改进与趋势。

原生集成进展

dbt 正在持续改进其核心产品与 AI 的集成。在 Snowflake 中，我们可以使用 Snowsight 作为 UI 界面来管理 dbt 工作空间。通过更无缝地融入 Snowflake 环境，我们可以减少在外部编辑 dbt 模型的需求，转而在 Snowflake UI 中集中团队的编辑与协作工作流。

为加速开发，dbt 提供了一个名为 dbt Copilot 的 AI 助手。可通过 dbt Studio IDE、Canvas 或 Insights 访问，利用自然语言处理来加速模型开发中耗时的环节，如撰写文档与创建测试。

dbt Copilot 甚至可用于帮助新分析师入职，为其提供项目与模型的简要概述。

请同时利用这两类工具，加速投产、缩短开发时间，让团队有更多精力应对组织中更困难、更具挑战的数据问题。

机器学习集成

机器学习管道需要持续干净且良好设计的数据。dbt 是构建馈送机器学习模型的数据转换管道的理想工具。我们可以利用 dbt 的灵活性与强大能力，自动化数据转换并无缝集成至机器学习管道。

例如，我们可以将数据直接转换到机器学习仓库。然后结合Snowflake 的 Cortex AI获取更深入洞察。若有多个模型需要相似的转换，dbt 的模板机制可简化数据清洗方式，并提供更高一致性与更易管治的数据。若想进一步了解 Snowflake 的 AI 工具，请阅读这份关于 Snowflake Arctic 的指南，它是 Snowflake 的大语言模型（LLM）。

容器化执行

得益于 Snowflake 的原生 Python 支持与 Snowpark for Python API，我们可以在不迁移数据出 Snowflake 的前提下本地运行代码。更进一步的是使用 Snowpark Containers（在撰写时仅适用于 AWS 与 Azure 区域）来完全隔离地执行 dbt 代码。

为此，我们将 dbt 模型与环境打包为容器，并存放于 Snowflake 的镜像仓库中。

随后即可在 Snowpark 中便捷使用该容器化的 dbt 项目。其主要好处？更易利用 Snowflake 的资源进行复杂的基于 Python 的数据转换，并在更大的 Snowflake 环境中集成 dbt 转换。

若您想进一步了解 Snowpark，请查看这份介绍Snowflake Snowpark 细节的资料。

结语

将 dbt 与 Snowflake 集成，可帮助数据团队构建模块化、受治理且可扩展的转换管道。借助自动化测试、基于 Git 的 CI/CD 工作流与无缝扩展能力，该技术栈非常适合现代数据运营。

为保持竞争力，数据团队应优先关注：

• 更好的数据治理，以维护数据生命周期
• 提升成本可见性与持续性能监控，以合理定尺仓库
• 容器化以保证可复现性，并统一分析与机器学习的工作流

随着两款工具持续发展，dbt + Snowflake 的集成只会愈发强大。若您希望进一步学习 dbt 或 Snowflake，请参考以下资源：

• Top 26 个 dbt 面试问题
• Top 32 个 Snowflake 面试问题
• SQL 助理数据工程师
• Snowflake 架构深度解析