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

-- Multi-cluster 웨어하우스 생성
CREATE WAREHOUSE analytics_wh
  WITH
    WAREHOUSE_SIZE = 'MEDIUM'
    MIN_CLUSTER_COUNT = 1
    MAX_CLUSTER_COUNT = 5
    SCALING_POLICY = 'STANDARD'   -- ECONOMY도 가능
    AUTO_SUSPEND = 300             -- 5분 후 자동 정지
    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',
    'ko'
) 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로 병목 분석
-- Snowflake 웹 UI의 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  -- On-demand 기준
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;