Snowflake Data Engineerキャリアガイド：クラウドデータウェアハウスの王座を制した技術の全て

┌──────────────────────────────────────────────┐
│         Cloud Services Layer                  │
│   (認証、メタデータ、クエリ最適化、トランザクション)  │
├──────────────────────────────────────────────┤
│         Compute Layer (Virtual Warehouses)     │
│   ┌──────┐  ┌──────┐  ┌──────┐               │
│   │ WH-1 │  │ WH-2 │  │ WH-3 │  独立スケール   │
│   │  XS   │  │  L   │  │  2XL │               │
│   └──────┘  └──────┘  └──────┘               │
├──────────────────────────────────────────────┤
│         Storage Layer                         │
│   (マイクロパーティション、カラムナ圧縮、S3/Blob/GCS) │
└──────────────────────────────────────────────┘

-- クラスタリングキーの設定
ALTER TABLE sales
  CLUSTER BY (sale_date, region);

-- クラスタリング状態の確認
SELECT SYSTEM$CLUSTERING_INFORMATION('sales', '(sale_date, region)');

-- 結果例:
-- {
--   "cluster_by_keys": "LINEAR(sale_date, region)",
--   "total_partition_count": 1024,
--   "total_constant_partition_count": 512,
--   "average_overlaps": 1.5,
--   "average_depth": 2.1
-- }

-- マルチクラスタウェアハウスの作成
CREATE WAREHOUSE analytics_wh
  WITH
    WAREHOUSE_SIZE = 'MEDIUM'
    MIN_CLUSTER_COUNT = 1
    MAX_CLUSTER_COUNT = 5
    SCALING_POLICY = 'STANDARD'
    AUTO_SUSPEND = 300
    AUTO_RESUME = TRUE;

-- 共有の作成
CREATE SHARE sales_share;

-- データベースとテーブルの共有
GRANT USAGE ON DATABASE analytics TO SHARE sales_share;
GRANT USAGE ON SCHEMA analytics.public TO SHARE sales_share;
GRANT SELECT ON TABLE analytics.public.sales TO SHARE sales_share;

-- コンシューマーアカウントの追加
ALTER SHARE sales_share ADD ACCOUNTS = partner_account;

-- 1時間前のデータを照会
SELECT * FROM orders
  AT(OFFSET => -3600);

-- 特定時点のデータを照会
SELECT * FROM orders
  AT(TIMESTAMP => '2025-03-20 14:30:00'::TIMESTAMP);

-- 特定クエリ実行直前の状態を照会
SELECT * FROM orders
  BEFORE(STATEMENT => '01abc-def-12345');

-- 削除されたテーブルの復元
UNDROP TABLE orders;

-- 過去時点へのテーブル復元
CREATE TABLE orders_restored CLONE orders
  AT(TIMESTAMP => '2025-03-20 10:00:00'::TIMESTAMP);

-- テーブルのクローン
CREATE TABLE orders_dev CLONE orders;

-- スキーマのクローン
CREATE SCHEMA dev_schema CLONE prod_schema;

-- データベースのクローン
CREATE DATABASE staging CLONE production;

-- Time Travelと組み合わせ
CREATE TABLE orders_backup CLONE orders
  AT(TIMESTAMP => '2025-03-20 10:00:00'::TIMESTAMP);

-- パイプの作成
CREATE PIPE sales_pipe
  AUTO_INGEST = TRUE
AS
  COPY INTO sales_raw
  FROM @my_s3_stage/sales/
  FILE_FORMAT = (TYPE = 'JSON')
  MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

-- パイプ状態の確認
SELECT SYSTEM$PIPE_STATUS('sales_pipe');

-- ロード履歴の確認
SELECT *
FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'sales_raw',
  START_TIME => DATEADD(hours, -24, CURRENT_TIMESTAMP())
));

# Snowpipe Streaming Python例
from snowflake.ingest import SimpleIngestManager, StagedFile

ingest_manager = SimpleIngestManager(
    account='myaccount',
    host='myaccount.snowflakecomputing.com',
    user='myuser',
    pipe='mydb.public.mypipe',
    private_key=private_key
)

staged_files = [StagedFile('data/file1.csv', None)]
resp = ingest_manager.ingest_files(staged_files)

-- Stream作成（CDC追跡）
CREATE STREAM orders_stream ON TABLE orders;

-- 変更内容の確認
SELECT * FROM orders_stream;
-- METADATA$ACTION: INSERT/DELETE
-- METADATA$ISUPDATE: TRUE/FALSE
-- METADATA$ROW_ID: 行識別子

-- Task作成（5分ごとに実行）
CREATE TASK process_orders
  WAREHOUSE = etl_wh
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
  MERGE INTO orders_analytics AS target
  USING orders_stream AS source
  ON target.order_id = source.order_id
  WHEN MATCHED AND source.METADATA$ACTION = 'DELETE'
    THEN DELETE
  WHEN MATCHED AND source.METADATA$ISUPDATE = TRUE
    THEN UPDATE SET target.amount = source.amount,
                    target.updated_at = CURRENT_TIMESTAMP()
  WHEN NOT MATCHED AND source.METADATA$ACTION = 'INSERT'
    THEN INSERT (order_id, amount, created_at)
         VALUES (source.order_id, source.amount, CURRENT_TIMESTAMP());

-- Taskの開始
ALTER TASK process_orders RESUME;

-- Dynamic Tableの作成
CREATE DYNAMIC TABLE daily_sales_summary
  TARGET_LAG = '1 hour'
  WAREHOUSE = transform_wh
AS
  SELECT
    DATE_TRUNC('day', sale_date) AS day,
    region,
    COUNT(*) AS total_orders,
    SUM(amount) AS total_revenue,
    AVG(amount) AS avg_order_value
  FROM raw_sales
  GROUP BY 1, 2;

-- Dynamic Tableチェーン（階層的変換）
CREATE DYNAMIC TABLE monthly_kpi
  TARGET_LAG = DOWNSTREAM
  WAREHOUSE = transform_wh
AS
  SELECT
    DATE_TRUNC('month', day) AS month,
    SUM(total_orders) AS monthly_orders,
    SUM(total_revenue) AS monthly_revenue
  FROM daily_sales_summary
  GROUP BY 1;

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, sum as sum_, avg, when

# セッション作成
session = Session.builder.configs({
    "account": "myaccount",
    "user": "myuser",
    "password": "mypass",
    "warehouse": "compute_wh",
    "database": "analytics",
    "schema": "public"
}).create()

# DataFrame作成と変換
df = session.table("orders")
result = (
    df.filter(col("status") == "completed")
      .group_by(col("region"))
      .agg(
          sum_(col("amount")).alias("total_revenue"),
          avg(col("amount")).alias("avg_order"),
          col("region")
      )
      .sort(col("total_revenue").desc())
)

# 結果を新しいテーブルに保存
result.write.mode("overwrite").save_as_table("regional_summary")

# Python UDF登録
from snowflake.snowpark.functions import udf
from snowflake.snowpark.types import StringType, IntegerType

@udf(name="sentiment_score", is_permanent=True,
     stage_location="@my_stage",
     replace=True)
def sentiment_score(text: str) -> int:
    """テキストの感情スコアを返す (-1, 0, 1)"""
    positive = ["good", "great", "excellent", "amazing"]
    negative = ["bad", "terrible", "awful", "poor"]
    text_lower = text.lower()
    if any(w in text_lower for w in positive):
        return 1
    elif any(w in text_lower for w in negative):
        return -1
    return 0

# UDFの使用
df = session.table("reviews")
df.select(
    col("review_text"),
    sentiment_score(col("review_text")).alias("sentiment")
).show()

from snowflake.snowpark import Session

def process_daily_etl(session: Session, target_date: str) -> str:
    """日次ETL処理プロシージャ"""
    # 1. Rawデータの読み込み
    raw = session.table("raw_events").filter(
        col("event_date") == target_date
    )

    # 2. 変換
    transformed = (
        raw.with_column("category",
            when(col("event_type") == "purchase", "revenue")
            .when(col("event_type") == "signup", "acquisition")
            .otherwise("engagement")
        )
        .group_by("category")
        .agg(sum_(col("value")).alias("total_value"))
    )

    # 3. 結果保存
    transformed.write.mode("overwrite").save_as_table(
        f"daily_summary_{target_date.replace('-', '')}"
    )

    return f"Processed {raw.count()} events for {target_date}"

# Stored Procedureの登録
session.sproc.register(
    func=process_daily_etl,
    name="daily_etl_proc",
    is_permanent=True,
    stage_location="@my_stage",
    replace=True
)

from snowflake.ml.modeling.preprocessing import (
    StandardScaler, OneHotEncoder, OrdinalEncoder
)
from snowflake.ml.modeling.pipeline import Pipeline
from snowflake.ml.modeling.xgboost import XGBClassifier
from snowflake.ml.registry import Registry

# パイプライン構成
pipeline = Pipeline(steps=[
    ("scaler", StandardScaler(
        input_cols=["age", "income"],
        output_cols=["age_scaled", "income_scaled"]
    )),
    ("encoder", OneHotEncoder(
        input_cols=["region"],
        output_cols=["region_encoded"]
    )),
    ("model", XGBClassifier(
        input_cols=["age_scaled", "income_scaled", "region_encoded"],
        label_cols=["churn"],
        output_cols=["predicted_churn"]
    ))
])

# 訓練
train_df = session.table("customer_features")
pipeline.fit(train_df)

# モデルレジストリに登録
registry = Registry(session=session)
model_version = registry.log_model(
    model_name="churn_predictor",
    version_name="v1",
    model=pipeline,
    sample_input_data=train_df.limit(10)
)

# 推論
predictions = model_version.run(
    session.table("new_customers"),
    function_name="predict"
)

-- COMPLETE：自由形式テキスト生成
SELECT SNOWFLAKE.CORTEX.COMPLETE(
    'mistral-large2',
    'Summarize the key benefits of cloud data warehousing in 3 bullet points.'
) AS response;

-- SUMMARIZE：テキスト要約
SELECT
    ticket_id,
    SNOWFLAKE.CORTEX.SUMMARIZE(description) AS summary
FROM support_tickets
WHERE created_at >= DATEADD(day, -7, CURRENT_DATE());

-- EXTRACT_ANSWER：質問に対する回答抽出
SELECT
    doc_id,
    SNOWFLAKE.CORTEX.EXTRACT_ANSWER(
        content,
        'What is the refund policy?'
    ) AS answer
FROM policy_documents;

-- SENTIMENT：感情分析（-1〜1）
SELECT
    review_id,
    review_text,
    SNOWFLAKE.CORTEX.SENTIMENT(review_text) AS score
FROM customer_reviews;

-- TRANSLATE：翻訳
SELECT SNOWFLAKE.CORTEX.TRANSLATE(
    'Cloud data warehousing enables scalable analytics.',
    'en',
    'ja'
) AS translated;

-- エンベディング生成
CREATE TABLE document_embeddings AS
SELECT
    doc_id,
    content,
    SNOWFLAKE.CORTEX.EMBED_TEXT_1024(
        'snowflake-arctic-embed-l-v2.0',
        content
    ) AS embedding
FROM documents;

-- ベクトル検索（コサイン類似度）
SELECT
    doc_id,
    content,
    VECTOR_COSINE_SIMILARITY(
        embedding,
        SNOWFLAKE.CORTEX.EMBED_TEXT_1024(
            'snowflake-arctic-embed-l-v2.0',
            'How to optimize query performance?'
        )
    ) AS similarity
FROM document_embeddings
ORDER BY similarity DESC
LIMIT 5;

-- Cortex Search Serviceの作成
CREATE CORTEX SEARCH SERVICE doc_search
  ON content
  ATTRIBUTES category, department
  WAREHOUSE = search_wh
  TARGET_LAG = '1 hour'
AS (
    SELECT content, category, department, doc_id
    FROM knowledge_base
);

-- ファインチューニング：カスタムモデル作成
SELECT SNOWFLAKE.CORTEX.FINETUNE(
    'CREATE',
    'my_custom_model',
    'mistral-7b',
    'SELECT prompt, completion FROM training_data',
    '{}'
);

-- Document AI：非構造化文書からデータ抽出
CREATE DOCUMENT AI MODEL invoice_extractor
  FROM @docs_stage
  WITH
    TRAINING_DATA = 'SELECT * FROM labeled_invoices'
    MODEL_TYPE = 'extraction';

-- 文書からフィールドを抽出
SELECT
    file_name,
    invoice_extractor!PREDICT(
        GET_PRESIGNED_URL(@docs_stage, file_name),
        'invoice_number'
    ) AS invoice_num,
    invoice_extractor!PREDICT(
        GET_PRESIGNED_URL(@docs_stage, file_name),
        'total_amount'
    ) AS total
FROM directory(@docs_stage);

-- 外部Icebergカタログ統合
CREATE CATALOG INTEGRATION glue_catalog
  CATALOG_SOURCE = GLUE
  CATALOG_NAMESPACE = 'my_namespace'
  TABLE_FORMAT = ICEBERG
  GLUE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/snowflake-glue'
  GLUE_CATALOG_ID = '123456789012'
  GLUE_REGION = 'us-east-1'
  ENABLED = TRUE;

-- Icebergテーブル作成（Snowflakeマネージド）
CREATE ICEBERG TABLE events (
    event_id STRING,
    event_type STRING,
    user_id STRING,
    event_data VARIANT,
    event_time TIMESTAMP_NTZ
)
  CATALOG = 'SNOWFLAKE'
  EXTERNAL_VOLUME = 'my_s3_volume'
  BASE_LOCATION = 'events/'
  AS SELECT * FROM raw_events;

# dbt_project.yml
name: 'analytics'
version: '1.0.0'

profile: 'snowflake_profile'

models:
  analytics:
    staging:
      +materialized: view
      +schema: staging
    intermediate:
      +materialized: ephemeral
    marts:
      +materialized: table
      +schema: marts

-- models/marts/fct_orders.sql
-- dbt incremental model

WITH source AS (
    SELECT
        order_id,
        customer_id,
        order_date,
        amount,
        status,
        _loaded_at
    FROM raw.orders
    WHERE 1=1
    -- incremental条件
    AND _loaded_at >= COALESCE(
        (SELECT MAX(_loaded_at) FROM analytics.marts.fct_orders),
        '1900-01-01'
    )
)

SELECT
    order_id,
    customer_id,
    order_date,
    amount,
    status,
    CURRENT_TIMESTAMP() AS processed_at,
    _loaded_at
FROM source

# models/marts/schema.yml
version: 2

models:
  - name: fct_orders
    description: '注文ファクトテーブル'
    columns:
      - name: order_id
        description: '注文固有ID'
        tests:
          - unique
          - not_null
      - name: customer_id
        description: '顧客ID'
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: amount
        description: '注文金額'
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 1000000

-- Snowflake専用マクロ：merge戦略
-- models/marts/dim_customers.sql

-- configでmerge戦略を使用
-- materialized='incremental'
-- incremental_strategy='merge'
-- unique_key='customer_id'

SELECT
    customer_id,
    customer_name,
    email,
    segment,
    lifetime_value,
    CURRENT_TIMESTAMP() AS updated_at
FROM staging.stg_customers

-- dbt snapshot (SCD Type 2)
-- snapshots/scd_customers.sql

-- snapshot設定
-- target_schema='snapshots'
-- unique_key='customer_id'
-- strategy='timestamp'
-- updated_at='updated_at'

SELECT
    customer_id,
    customer_name,
    email,
    segment,
    updated_at
FROM raw.customers

-- Resource Monitorの設定
CREATE RESOURCE MONITOR monthly_limit
  WITH CREDIT_QUOTA = 1000
  FREQUENCY = MONTHLY
  START_TIMESTAMP = IMMEDIATELY
  TRIGGERS
    ON 75 PERCENT DO NOTIFY
    ON 90 PERCENT DO NOTIFY
    ON 100 PERCENT DO SUSPEND;

ALTER WAREHOUSE analytics_wh SET RESOURCE_MONITOR = monthly_limit;

-- Query Profileでボトルネック分析
SELECT
    query_id,
    query_text,
    execution_status,
    total_elapsed_time / 1000 AS elapsed_sec,
    bytes_scanned / (1024*1024*1024) AS gb_scanned,
    partitions_scanned,
    partitions_total,
    ROUND(partitions_scanned / NULLIF(partitions_total, 0) * 100, 2)
      AS pct_scanned
FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY(
    DATEADD('hours', -24, CURRENT_TIMESTAMP()),
    CURRENT_TIMESTAMP()
))
WHERE execution_status = 'SUCCESS'
ORDER BY total_elapsed_time DESC
LIMIT 20;

-- 日次クレジット使用量の追跡
SELECT
    TO_DATE(start_time) AS usage_date,
    warehouse_name,
    SUM(credits_used) AS total_credits,
    SUM(credits_used) * 3.00 AS estimated_cost_usd
FROM SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_METERING_HISTORY
WHERE start_time >= DATEADD(month, -1, CURRENT_TIMESTAMP())
GROUP BY 1, 2
ORDER BY total_credits DESC;

-- 最もコストの高いクエリTop 20
SELECT
    query_id,
    user_name,
    warehouse_name,
    execution_status,
    total_elapsed_time / 1000 AS elapsed_sec,
    credits_used_cloud_services,
    bytes_scanned / (1024*1024*1024) AS gb_scanned,
    LEFT(query_text, 200) AS query_preview
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP())
ORDER BY credits_used_cloud_services DESC NULLS LAST
LIMIT 20;

-- VARIANTクエリ例
SELECT
    raw:customer.name::STRING AS customer_name,
    f.value:product_id::INTEGER AS product_id,
    f.value:quantity::INTEGER AS quantity
FROM orders,
    LATERAL FLATTEN(input => raw:items) f;