snowflake-data-engineering

Snowflake データエンジニアリング

このスキルでは、SQL の規約、パイプラインアーキテクチャ（Dynamic Tables、Streams、Tasks、Snowpipe）、パフォーマンスチューニング、Snowflake でのコスト管理とアクセス管理について説明します。

Snowflake パイプライン構築のワークフロー

1. 生データを取り込む — 外部ステージからファイルを継続的にロードするには Snowpipe（AUTO_INGEST = TRUE）を使用し、SDK 経由で低レイテンシの行レベル取り込みを行うには Snowpipe Streaming を使用します。
2. 変換方法を選ぶ — ほとんどのパイプラインでは宣言型の Dynamic Tables を優先します。手続き型ロジックやストアドプロシージャの呼び出しが必要な場合に限り、Streams + Tasks を使用します。
3. 半構造化データをモデル化する — 生の JSON／Avro／Parquet を VARIANT として取り込み、可能な限り早い段階で型付きのリレーショナル列にフラット化します。
4. パイプラインの各段階を連結する — Dynamic Tables を積み重ねる（または Streams から Tasks にデータを渡す）ことで、各段階で対象範囲を生データからクリーンデータ、集計データへと絞り込みます。
5. パフォーマンスを調整する — クラスタリングキーや Search Optimization は、クエリパターン上必要な場合にのみ追加し、コストの帰属を追跡できるようクエリにタグを付けます。
6. アクセス制御を設定する — 機能別ロール（ローダー、トランスフォーマー、アナリスト）を使った最小権限の RBAC と、機密データ向けのマスキングポリシー／行アクセス・ポリシーを適用します。
7. コストと鮮度を監視する — WAREHOUSE_METERING_HISTORY と QUERY_HISTORY を追跡し、Resource Monitors を設定して、TARGET_LAG が実際の鮮度要件に合っていることを確認します。

SQL と半構造化データ

• JSON、Avro、Parquet、ORC データには VARIANT、OBJECT、ARRAY 型を使用します。
• コロン記法でネストされたフィールドにアクセスし、明示的にキャストします：src:customer.name::STRING、src:price::NUMBER(10,2)、src:created_at::TIMESTAMP_NTZ。
• LATERAL FLATTEN で配列をフラット化します：

SELECT f.value:name::STRING AS name
FROM my_table, LATERAL FLATTEN(input => src:items) f;

• 日付、文字列として格納された数値、または配列を含む半構造化データは、可能な限りリレーショナル列にフラット化します。データを VARIANT 内に保持すると、Snowflake の自動サブカラム化の効果が得られません。
• 同じ理由から、同じ VARIANT フィールド内で型を混在させないでください。
• JSON の null は文字列 "null" として格納され、SQL の NULL とは異なることに注意してください。同じものとして扱うには、ロード時に STRIP_NULL_VALUES => TRUE を使用します。

SQL コーディング規約

• すべての識別子に snake_case を使用し、引用符付き識別子は避けます。
• 可読性を高めるため、深くネストされたサブクエリより CTE を優先します。
• 冪等な DDL には CREATE OR REPLACE を使用します。
• バルクロードには COPY INTO を使用し、1 行ずつの INSERT は決して使用しません。
• アップサートには MERGE を使用します：

MERGE INTO target t USING source s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.name = s.name
WHEN NOT MATCHED THEN INSERT (id, name) VALUES (s.id, s.name);

• ストアドプロシージャでは、SQL 文の中でセッション変数／ローカル変数を参照する際に、変数名の前に : を付けます：

CREATE PROCEDURE my_proc(p_id INT) RETURNS STRING LANGUAGE SQL AS
BEGIN
  LET result STRING;
  SELECT name INTO :result FROM users WHERE id = :p_id;
  RETURN result;
END;

パフォーマンスの最適化

• クラスタリングキーは、頻繁に WHERE／JOIN／GROUP BY で使用される列を持つ、非常に大きなテーブル（数 TB 以上）にのみ追加します：

ALTER TABLE large_events CLUSTER BY (event_date, region);

• カーディナリティの高い列でのポイント検索や、部分文字列／正規表現検索には Search Optimization Service を使用します：

ALTER TABLE logs ADD SEARCH OPTIMIZATION ON EQUALITY(sender_ip), SUBSTRING(error_message);

• 単一テーブルに対するコストの高い集計を事前計算するには、Materialized Views を使用します。
• 同一のクエリを再実行する代わりに、RESULT_SCAN(LAST_QUERY_ID()) を使って以前の結果を再利用します。
• コストの帰属を追跡できるようクエリにタグを付けます：ALTER SESSION SET QUERY_TAG = 'etl_daily_load';
• ワイドテーブルに対して SELECT * を実行しないでください。列指向のプルーニング効果が失われます。

データパイプライン

適切なプリミティブを選びます：

アプローチ	用途
Dynamic Tables	宣言型：クエリを定義すると、Snowflake がリフレッシュを管理します。ほとんどのパイプラインでのデフォルトの選択肢です。
Streams + Tasks	手続き型の CDC とスケジューリングに使用します。手続き型ロジックやストアドプロシージャの呼び出しが必要な場合に使用します。
Snowpipe	S3／GCS／Azure からの継続的なファイルロードに使用します。
Snowpipe Streaming	SDK（Java、Python）経由の低レイテンシな行レベル取り込みに使用します。

Dynamic Tables

CREATE OR REPLACE DYNAMIC TABLE cleaned_events
  TARGET_LAG = '5 minutes'
  WAREHOUSE = transform_wh
  AS
  SELECT event_id, event_type, user_id, event_data:page::STRING AS page, event_timestamp
  FROM raw_events
  WHERE event_type IS NOT NULL;

-- Chain for multi-step pipelines
CREATE OR REPLACE DYNAMIC TABLE user_sessions
  TARGET_LAG = '10 minutes'
  WAREHOUSE = transform_wh
  AS
  SELECT user_id, MIN(event_timestamp) AS session_start, MAX(event_timestamp) AS session_end,
         COUNT(*) AS event_count
  FROM cleaned_events GROUP BY user_id;

TARGET_LAG は鮮度の目標値を設定します。REFRESH_MODE は AUTO、FULL、または INCREMENTAL に設定できます。ALTER DYNAMIC TABLE ... SET TARGET_LAG / REFRESH / SUSPEND / RESUME でライフサイクルを管理します。

ストリーム（CDC）

CREATE OR REPLACE STREAM raw_events_stream ON TABLE raw_events;

ストリームには、METADATA$ACTION、METADATA$ISUPDATE、METADATA$ROW_ID の各列が追加されます。挿入のみのソースでは、オーバーヘッドを抑えるために APPEND_ONLY = TRUE を設定してください。

タスク（スケジュール実行／トリガー実行）

CREATE OR REPLACE TASK process_events
  WAREHOUSE = transform_wh
  SCHEDULE = 'USING CRON 0 */1 * * * America/Los_Angeles'
  WHEN SYSTEM$STREAM_HAS_DATA('raw_events_stream')
  AS
  INSERT INTO cleaned_events
  SELECT event_id, event_type, user_id, event_timestamp
  FROM raw_events_stream WHERE event_type IS NOT NULL;

CREATE TASK child_task ... AFTER parent_task ... を使用して、タスク DAG を構築します。タスクはデフォルトで SUSPENDED 状態で作成されます。ALTER TASK ... RESUME を忘れると、何も実行されません。

Snowpipe

CREATE OR REPLACE PIPE my_pipe AUTO_INGEST = TRUE AS
  COPY INTO raw_events FROM @my_external_stage FILE_FORMAT = (TYPE = 'JSON');

一般的なエンドツーエンドのパターンは、Snowpipe で生データを取り込み、Dynamic Tables のチェーンに供給する方法です。

Time Travel とデータ保護

• Time Travel を使用して過去のデータをクエリします（デフォルトでは 1 日間、Enterprise 以上では最大 90 日間）。

SELECT * FROM my_table AT(TIMESTAMP => '2026-01-15 10:00:00'::TIMESTAMP);
SELECT * FROM my_table BEFORE(STATEMENT => '<query_id>');

• UNDROP TABLE/SCHEMA/DATABASE を使用して、削除したオブジェクトを復元します。
• ストレージを重複させずに開発／テスト環境やバックアップを作成するには、ゼロコピークローンを使用します。CREATE TABLE clone CLONE source;、CREATE SCHEMA dev CLONE prod;。

Snowflake Postgres

• Snowflake は、ワイヤ互換性を完全に備えたマネージド PostgreSQL（v16/17/18）を提供しています。CREATE POSTGRES INSTANCE my_instance COMPUTE_FAMILY='STANDARD_S' STORAGE_SIZE_GB=50;
• pg_lake 拡張機能を使用して OLTP と分析を連携します。この拡張機能により、Postgres と Snowflake の両方から読み取り可能な Iceberg テーブルが公開されます。
• ポイントインタイムリカバリには FORK を使用し、本番インスタンスでは HIGH_AVAILABILITY = TRUE を設定します。

ウェアハウスとコスト管理

• ウェアハウスのサイズは、生データの量ではなくクエリの複雑さに応じて決めます。まず X-Small から始め、必要な場合にのみスケールアップします。
• AUTO_SUSPEND = 60 と AUTO_RESUME = TRUE を設定します。また、負荷の高いジョブが対話型クエリを圧迫しないよう、ワークロードごとにウェアハウスを分けます。
• 同時実行性のスケーリングにはマルチクラスターウェアハウスを使用し、単一クエリの高速化には使用しません。
• Fail-safe のストレージコストを避けるため、ステージングデータには一時テーブルを使用します。
• SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY と WAREHOUSE_METERING_HISTORY で支出を監視し、Resource Monitor を設定してクレジット消費に上限を設けます。

アクセス制御

• オブジェクトへの権限付与にはデータベースロールを使用し、最小権限の RBAC を適用します。
• PII にはマスキングポリシーを使用し、マルチテナントの分離には行アクセスポリシーを使用します。
• パイプラインの各段階に応じて機能ロールを構成します。loader（生データを書き込む）、transformer（生データを読み取り、分析データを書き込む）、analyst（分析データのみを読み取る）。

データ共有と Iceberg

• ゼロコピーのアカウント間データ共有には CREATE SHARE を使用するか、外部とのデータ交換には Snowflake Marketplace を使用します。
• Spark、Flink、Trino との相互運用性のため、CREATE ICEBERG TABLE ... CATALOG='SNOWFLAKE' EXTERNAL_VOLUME='vol' BASE_LOCATION='path/'; を使用して Iceberg テーブルを作成します。

アンチパターン

• Dynamic Table で宣言的に表現できる単純な変換には、Streams と Tasks を使用しないでください。
• TARGET_LAG を実際に必要な鮮度より短く設定しないでください。計算コストに直接影響します。
• 作成後のタスクに RESUME を実行し忘れないでください。タスクは SUSPENDED 状態で開始します。
• 幅の広いテーブルに対して SELECT * を実行しないでください。また、マルチ TB テーブルでは、クラスターキーを追加する前にクラスタリング分析を必ず行ってください。
• 再利用可能なパイプラインコードでデータベース名やスキーマ名をハードコードしないでください。