snowflake-snowpark-dbt

Snowflake Snowpark Python と dbt

このスキルでは、Snowpark Python（Snowflake のサーバーサイド Python API）および dbt-snowflake アダプターを使って、本番環境向けのデータ変換パイプラインを構築する方法を説明します。

Snowpark または dbt による変換のワークフロー

1. Snowpark：セッションを開く — 環境ごとに設定された認証情報から Session を構築し、ロール、ウェアハウス、データベース、スキーマを明示的に指定します。
2. Snowpark：DataFrame API で変換を表現する — 再利用可能なパイプラインコードでは、生の SQL 文字列より .filter()、.select()、.group_by().agg()、.join() を優先します。DataFrame は遅延評価され、.collect()／.show()／書き込み操作でのみ実行されます。
3. Snowpark：計算をサーバーサイドで実行する — 行単位のロジックにはスカラー UDF、ML 推論にはベクトル化（pandas）UDF、1 つの入力行から複数の出力行を生成する場合には 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（依存関係の順序に従って run と test を実行）を実行し、dbt docs generate でドキュメントを最新の状態に保ちます。

Snowpark Python

Snowpark は Snowflake ウェアハウス内で Python をサーバーサイド実行します。データが Snowflake の外部に出ることはありません。主な抽象化は Session、DataFrame、UDF、UDTF、UDAF、ストアドプロシージャです。

セッション

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

ベクトル化（pandas）UDF は、行ごとに Python を呼び出すのではなく、行をまとめて処理するため、ML 推論ではスカラー UDF より 10～100 倍高速です。

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 とストアドプロシージャでは、バージョンを固定してください。
• 静的ファイル（例：pickle 化されたモデル）を session.add_import("@my_stage/model.pkl") でアタッチします。
• クライアントへのデータ移動なしで pandas-on-Snowflake を使用するには、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 の Dynamic Tables

{{ 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 %}

{{ this }} は必ず {% if is_incremental() %} で囲んでください。無条件に参照すると、対象テーブルがまだ存在しない初回の（全件）実行でエラーになります。

Snowflake 固有の設定

• cluster_by=['col1', 'col2'] — クラスタリング。大規模テーブルでのみ使用します（一般に 1 TB 超）。
• transient=true — フェイルセーフがなく、ストレージコストを抑えられます。ステージングモデルに使用します。
• query_tag='finance_daily' — コスト追跡のためのワークロード帰属情報です。
• copy_grants=true — CREATE OR REPLACE の実行後もアクセス権限を維持します。
• snowflake_warehouse='lg_wh' — 負荷の高い変換に使用する、モデルごとのウェアハウスの上書き設定です。
• secure=true — 機密列を公開するモデル向けのセキュアビューです。

Freshness チェック付きのソース

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 推論には、スカラー UDF よりベクトル化（pandas）UDF を優先します。
• 本番環境の UDF とストアドプロシージャでは、パッケージのバージョンを固定します。
• 再利用可能な Python パイプラインでは、生の SQL 文字列より Snowpark DataFrame API を使用します。
• ステージングモデル（stg_*）で名前の変更と型変換を行い、ビジネスロジックはマートモデルに記述します。
• ファクトテーブルには incremental マテリアライゼーションを、ニアリアルタイムの要件には dynamic_table を使用します。
• インクリメンタルモデルに on_schema_change='sync_all_columns' を設定し、上流のスキーマ変更を安全に処理します。
• 再ビルド時の権限変更を避けるために copy_grants=true を使用し、選択的な実行のためにモデルにタグを付けます。
• dbt の実行には、アナリストの対話型クエリとは別のウェアハウスを使用します。

アンチパターン

• 大規模な Snowpark DataFrame に .collect() を使ってクライアント側へ取り込まないでください。処理はサーバー側で行います。
• Snowpark で行を処理するために Python のループを使わないでください。DataFrame 操作またはベクトル化 UDF を使用します。
• インクリメンタルモデルでは、{% if is_incremental() %} ガードなしに {{ this }} を参照しないでください。
• 小規模なテーブル（およそ 1 TB 未満）に cluster_by を設定しないでください。オーバーヘッドがメリットを上回ります。
• すべてのモデルのデフォルトを materialized='table' にしないでください。ビューはクエリされるまでコストがかかりません。
• dbt モデルでデータベース名やスキーマ名をハードコードしないでください。{{ ref() }} と {{ source() }} を使用します。