snowflake-snowpark-dbt

Snowflake Snowpark Python 与 dbt

本技能介绍如何使用 Snowpark Python（Snowflake 的服务器端 Python API）以及通过 dbt-snowflake 适配器使用 dbt 构建生产级数据转换流水线。

Snowpark 或 dbt 转换工作流

1. Snowpark：打开会话 — 使用限定于环境的凭据创建 Session，并明确指定角色、仓库、数据库和架构。
2. Snowpark：使用 DataFrame API 表达转换 — 对于可复用的流水线代码，优先使用 .filter()、.select()、.group_by().agg() 和 .join()，而不是原始 SQL 字符串；DataFrame 会延迟求值，仅在调用 .collect()/.show() 或执行写入操作时运行。
3. Snowpark：将计算下推到服务器端 — 对逐行逻辑使用标量 UDF，对 ML 推理使用向量化（pandas）UDF，当一行输入会生成多行输出时使用 UDTF，并使用存储过程进行多步骤服务器端编排。
4. dbt：分层建模 — 暂存模型（stg_*）用于重命名和类型转换；数据集市模型在暂存模型之上表达业务逻辑。
5. dbt：选择物化方式 — 简单逻辑使用 view，仅在读取频繁时使用 table，大型事实表使用 incremental，需要近实时新鲜度时使用 dynamic_table。
6. dbt：定义源和测试 — 在 _sources.yml 中声明源并设置新鲜度阈值；为关键列添加 unique/not_null 测试。
7. dbt：选择性运行 — 迭代期间使用 dbt run --select model+（模型及其下游模型）或 +model（模型及其上游模型），而不是运行整个项目。
8. dbt：构建并验证 — 合并前运行 dbt build（按依赖顺序运行并测试），并运行 dbt docs generate 以保持文档最新。

Snowpark Python

Snowpark 在 Snowflake 仓库内的服务器端运行 Python — 数据不会离开 Snowflake。核心抽象包括：Session、DataFrame、UDF、UDTF、UDAF 和存储过程。

Session

import os
from snowflake.snowpark import Session

session = Session.builder.configs({
    "account": os.environ["SNOWFLAKE_ACCOUNT"],
    "user": os.environ["SNOWFLAKE_USER"],
    "password": os.environ["SNOWFLAKE_PASSWORD"],
    "role": "my_role", "warehouse": "my_wh", "database": "my_db", "schema": "my_schema",
}).create()

绝不要硬编码凭据 — 始终从环境变量或密钥管理器读取凭据。

DataFrame API

DataFrame 会延迟求值：Snowpark 会构建查询计划，并且仅在调用 collect()/show() 或执行写入操作时运行。

df = session.table("customers")
df_filtered = df.filter(df["region"] == "US").select("name", "email", "revenue")
df_agg = df.group_by("region").agg(sum("revenue").alias("total_revenue"))
df_agg.show()

主要操作：.filter()、.select()、.group_by().agg()、.join()、.sort()、.with_column()、.drop()、.distinct()、.limit()、.union_all()、.flatten()、.write.save_as_table()。

标量 UDF

from snowflake.snowpark.functions import udf

@udf(name="normalize_email", replace=True)
def normalize_email(email: str) -> str:
    return email.strip().lower() if email else None

向量化 UDF

对于 ML 推理，向量化（pandas）UDF 比标量 UDF 快 10-100 倍，因为它们会批量处理多行，而不是针对每行分别调用 Python。

import pandas as pd
from snowflake.snowpark.functions import udf

@udf(name="predict_score", packages=["scikit-learn", "pandas"], replace=True)
def predict_score(features: pd.Series) -> pd.Series:
    import pickle, sys
    model = pickle.load(open(sys.path[0] + "/model.pkl", "rb"))
    return pd.Series(model.predict(features.values.reshape(-1, 1)))

UDTF（每行输入返回多行）

from snowflake.snowpark.types import StructType, StructField, StringType

class Tokenizer:
    def process(self, text: str):
        for token in text.split():
            yield (token,)

tokenize = session.udtf.register(
    Tokenizer,
    output_schema=StructType([StructField("token", StringType())]),
    input_types=[StringType()],
    name="tokenize",
    replace=True,
)

存储过程

from snowflake.snowpark import Session
from snowflake.snowpark.functions import sproc

@sproc(name="daily_etl", replace=True, packages=["snowflake-snowpark-python"])
def daily_etl(session: Session) -> str:
    raw = session.table("raw_events")
    cleaned = raw.filter(raw["event_type"].is_not_null())
    cleaned.write.mode("overwrite").save_as_table("cleaned_events")
    return f"Processed {cleaned.count()} rows"

包和文件访问

• 使用 session.add_packages("pandas", "scikit-learn==1.3.0", "xgboost") 添加第三方包 — 为生产环境中的 UDF 和存储过程固定版本。
• 使用 session.add_import("@my_stage/model.pkl") 附加静态文件（例如，pickle 格式的模型）。
• 若要在不将数据传输到客户端的情况下使用 Snowflake 上的 pandas，请搭配 Snowpark 插件使用 modin.pandas：import modin.pandas as pd; import snowflake.snowpark.modin.plugin; df = pd.read_snowflake("my_table")。

使用 Snowflake 适配器的 dbt

使用 pip install dbt-snowflake 安装。

profiles.yml

my_project:
  target: dev
  outputs:
    dev:
      type: snowflake
      account: myaccount
      user: myuser
      password: "{{ env_var('SNOWFLAKE_PASSWORD') }}"
      role: transformer
      database: analytics
      warehouse: transforming
      schema: public
      threads: 4

绝不要将真实凭据提交到 profiles.yml — 始终使用 env_var() 获取密钥。

物化方式

可用的物化方式：view、table、incremental、ephemeral、dynamic_table。简单逻辑默认使用 view；仅当模型读取频繁到足以抵消存储成本时才使用 table；大型事实表使用 incremental，需要近实时新鲜度时使用 dynamic_table。

dbt 中的动态表

{{ config(materialized='dynamic_table', snowflake_warehouse='transforming', target_lag='1 hour') }}
SELECT customer_id, SUM(amount) AS lifetime_value FROM {{ ref('stg_orders') }} GROUP BY 1

增量模型

{{
  config(
    materialized='incremental',
    unique_key='event_id',
    incremental_strategy='merge',
    on_schema_change='sync_all_columns'
  )
}}
SELECT * FROM {{ ref('stg_events') }}
{% if is_incremental() %}
  WHERE event_timestamp > (SELECT MAX(event_timestamp) FROM {{ this }})
{% endif %}

始终使用 {% if is_incremental() %} 保护 {{ this }} — 无条件引用它会导致首次（完整）运行失败，因为此时目标表尚不存在。

Snowflake 特定配置

• cluster_by=['col1', 'col2'] — 聚类，仅适用于大型表（通常大于 1TB）。
• transient=true — 不提供 Fail-safe，存储成本较低；用于暂存模型。
• query_tag='finance_daily' — 用于成本跟踪的工作负载归属标记。
• copy_grants=true — 在执行 CREATE OR REPLACE 时保留访问权限。
• snowflake_warehouse='lg_wh' — 为繁重转换单独指定模型使用的仓库。
• secure=true — 创建安全视图，适用于公开敏感列的模型。

配置了新鲜度检查的数据源

sources:
  - name: raw
    database: raw_db
    schema: jaffle_shop
    tables:
      - name: customers
        loaded_at_field: _loaded_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}

测试

models:
  - name: stg_customers
    columns:
      - name: customer_id
        tests: [unique, not_null]

关键命令

• dbt run、dbt test、dbt build（按依赖顺序运行并测试）、dbt compile
• dbt run --select my_model+ — 运行模型及其所有下游模型
• dbt run --select +my_model — 运行模型及其所有上游模型
• dbt source freshness — 根据配置的阈值检查数据源是否过时
• dbt docs generate && dbt docs serve — 生成并预览文档

自定义架构命名

{% macro generate_schema_name(custom_schema_name, node) %}
  {% if custom_schema_name %}{{ custom_schema_name | trim }}{% else %}{{ target.schema }}{% endif %}
{% endmacro %}

最佳实践

• ML 推理优先使用向量化（pandas）UDF，而非标量 UDF。
• 在生产环境的 UDF 和存储过程里固定软件包版本。
• 在可复用的 Python 流水线中，优先使用 Snowpark DataFrame API，而非原始 SQL 字符串。
• 使用暂存模型（stg_*）重命名并转换数据类型；将业务逻辑放在集市模型中。
• 事实表使用 incremental 物化；对近实时需求使用 dynamic_table。
• 在增量模型上设置 on_schema_change='sync_all_columns'，以安全应对上游架构变更。
• 使用 copy_grants=true，避免重建时权限频繁变更，并为模型添加标签以便选择性执行。
• 为 dbt 运行和分析人员的交互式查询使用独立的仓库。

反模式

• 不要将大型 Snowpark DataFrame .collect() 到客户端；应在服务端完成处理。
• 不要在 Snowpark 中使用 Python 循环遍历行；应使用 DataFrame 操作或向量化 UDF。
• 在增量模型中使用 {{ this }} 时，必须加上 {% if is_incremental() %} 保护条件。
• 不要为较小的表（约 1TB 以下）设置 cluster_by；其开销大于收益。
• 不要默认将每个模型都设为 materialized='table'；视图在查询之前不会产生费用。
• 不要在 dbt 模型中硬编码数据库或架构名称；应使用 {{ ref() }} 和 {{ source() }}。