Data Warehouses: Snowflake vs BigQuery

Overview of Modern Data Warehouses

Evolution of Data Warehousing

text

Traditional DW → Cloud DW → Modern DW → Lakehouse
├─ On-premise     ├─ Redshift     ├─ Snowflake     ├─ Databricks
├─ Expensive      ├─ BigQuery     ├─ BigQuery      ├─ Snowflake
├─ Rigid schema   ├─ Managed      ├─ Serverless    ├─ BigQuery
└─ Limited scale  └─ Scalable     └─ Separation of   └─ Unity Catalog
                                       compute/storage

Core Concepts

• ELT vs ETL: Load first, transform later

• Separation of Compute & Storage: Scale independently

• Massively Parallel Processing (MPP): Distributed query execution

• Columnar Storage: Optimized for analytics

• Zero-copy cloning: Instant, space-efficient copies

Snowflake: The Data Cloud

Architecture Overview

text

┌─────────────────────────────────────────────────────┐
│                   Snowflake Architecture             │
├─────────────────────────────────────────────────────┤
│            Cloud Services Layer                     │
│  • Authentication • Metadata • Query Optimization   │
│  • Access Control • Transaction Management          │
├─────────────────────────────────────────────────────┤
│            Compute Layer (Virtual Warehouses)       │
│  • X-Small to 6X-Large clusters                     │
│  • Per-second billing                               │
│  • Auto-suspend/resume                              │
├─────────────────────────────────────────────────────┤
│            Storage Layer (Cloud Storage)            │
│  • AWS S3/Azure Blob/GCP Cloud Storage              │
│  • Columnar format (Parquet-like)                   │
│  • Automatic micro-partitioning                     │
└─────────────────────────────────────────────────────┘

Key Features

1. Virtual Warehouses (Compute)

sql

-- Create and manage virtual warehouses
CREATE WAREHOUSE analytics_wh
  WITH 
  WAREHOUSE_SIZE = 'X-LARGE'
  AUTO_SUSPEND = 300  -- Suspend after 5 minutes idle
  AUTO_RESUME = TRUE
  MIN_CLUSTER_COUNT = 1
  MAX_CLUSTER_COUNT = 10  -- Multi-cluster warehouse
  SCALING_POLICY = 'STANDARD';

-- Scale warehouse dynamically
ALTER WAREHOUSE analytics_wh SET WAREHOUSE_SIZE = '2X-LARGE';
ALTER WAREHOUSE analytics_wh SUSPEND;
ALTER WAREHOUSE analytics_wh RESUME;

2. Zero-Copy Cloning & Time Travel

sql

-- Instant cloning (no data duplication)
CREATE TABLE sales_clone CLONE sales_table;
CREATE SCHEMA analytics_dev CLONE analytics_prod;

-- Time Travel (up to 90 days)
SELECT * FROM sales_table
AT(TIMESTAMP => '2024-01-15 10:00:00'::timestamp);

-- Restore dropped table
UNDROP TABLE sales_table;

-- Query historical data
SELECT * FROM sales_table
BEFORE(STATEMENT => '8e5d0ca9-005e-44b6-b858-a6f50d72e22c');

3. Data Sharing & Marketplace

sql

-- Secure Data Sharing (no data movement)
CREATE SHARE sales_share;
GRANT USAGE ON DATABASE sales_db TO SHARE sales_share;
GRANT USAGE ON SCHEMA sales_db.public TO SHARE sales_share;
GRANT SELECT ON TABLE sales_db.public.sales TO SHARE sales_share;

-- Share with external account
ALTER SHARE sales_share ADD ACCOUNTS = xy12345;

-- Consume shared data
CREATE DATABASE external_sales FROM SHARE other_account.sales_share;

Snowflake SQL Features

sql

-- Window Functions with advanced options
SELECT 
    customer_id,
    order_date,
    amount,
    SUM(amount) OVER (
        PARTITION BY customer_id 
        ORDER BY order_date 
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS cumulative_amount,
    RANK() OVER (PARTITION BY customer_id ORDER BY amount DESC) AS rank,
    LAG(amount, 1) OVER (PARTITION BY customer_id ORDER BY order_date) AS prev_amount
FROM orders;

-- JSON & Semi-structured Data
CREATE TABLE user_profiles (
    user_id NUMBER,
    profile VARIANT  -- Stores JSON
);

INSERT INTO user_profiles 
SELECT 
    123,
    PARSE_JSON('{
        "name": "John Doe",
        "email": "john@example.com",
        "preferences": {
            "theme": "dark",
            "notifications": true
        },
        "tags": ["premium", "active"]
    }');

-- Query JSON data
SELECT 
    user_id,
    profile:name::STRING AS name,
    profile:email::STRING AS email,
    profile:preferences.theme::STRING AS theme,
    VALUE::STRING AS tag
FROM user_profiles,
LATERAL FLATTEN(input => profile:tags);

-- UDFs and Stored Procedures
CREATE OR REPLACE FUNCTION calculate_discount(price FLOAT, tier VARCHAR)
RETURNS FLOAT
LANGUAGE JAVASCRIPT
AS $$
    if (tier === 'premium') {
        return price * 0.8;
    } else if (tier === 'regular') {
        return price * 0.9;
    } else {
        return price;
    }
$$;

-- External functions (call external APIs)
CREATE EXTERNAL FUNCTION get_weather(lat FLOAT, long FLOAT)
RETURNS VARIANT
API_INTEGRATION = weather_api
AS 'https://api.weather.com/v1/forecast';

Snowflake Data Engineering Patterns

1. Incremental Load with Streams & Tasks

sql

-- Create stream for change data capture
CREATE OR REPLACE STREAM sales_stream ON TABLE raw_sales;

-- Create task for incremental processing
CREATE OR REPLACE TASK process_incremental_sales
    WAREHOUSE = transform_wh
    SCHEDULE = '5 MINUTE'
    WHEN SYSTEM$STREAM_HAS_DATA('sales_stream')
AS
BEGIN
    -- Merge changes into target table
    MERGE INTO processed_sales AS target
    USING (
        SELECT 
            sales_id,
            customer_id,
            amount,
            sale_date,
            metadata$action,
            metadata$isupdate
        FROM sales_stream
    ) AS source
    ON target.sales_id = source.sales_id
    WHEN MATCHED AND source.metadata$action = 'DELETE' THEN
        DELETE
    WHEN MATCHED AND source.metadata$action = 'INSERT' THEN
        UPDATE SET 
            target.amount = source.amount,
            target.last_updated = CURRENT_TIMESTAMP()
    WHEN NOT MATCHED AND source.metadata$action = 'INSERT' THEN
        INSERT (sales_id, customer_id, amount, sale_date)
        VALUES (source.sales_id, source.customer_id, 
                source.amount, source.sale_date);
    
    -- Log processing
    INSERT INTO task_log (task_name, rows_processed, run_time)
    SELECT 
        'process_incremental_sales',
        COUNT(*),
        CURRENT_TIMESTAMP()
    FROM sales_stream;
END;

-- Resume task
ALTER TASK process_incremental_sales RESUME;

2. Data Pipeline with External Stages

sql

-- Create external stage for cloud storage
CREATE OR REPLACE STAGE sales_stage
    URL = 's3://my-bucket/sales-data/'
    CREDENTIALS = (AWS_KEY_ID='...' AWS_SECRET_KEY='...')
    FILE_FORMAT = (TYPE = 'PARQUET');

-- Create file format for CSV
CREATE OR REPLACE FILE FORMAT csv_format
    TYPE = CSV
    FIELD_DELIMITER = ','
    SKIP_HEADER = 1
    NULL_IF = ('NULL', 'null')
    EMPTY_FIELD_AS_NULL = TRUE;

-- Load data from stage
COPY INTO sales_raw
FROM @sales_stage
PATTERN = '.*sales.*[.]parquet'
FILE_FORMAT = (TYPE = 'PARQUET')
ON_ERROR = 'CONTINUE';

-- Automate with Snowpipe (continuous ingestion)
CREATE OR REPLACE PIPE sales_pipe
    AUTO_INGEST = TRUE
    AS
    COPY INTO sales_raw
    FROM @sales_stage
    FILE_FORMAT = (TYPE = 'PARQUET');

-- Query staged data directly (no load)
SELECT *
FROM @sales_stage
(FILE_FORMAT => 'parquet_format')
LIMIT 100;

3. Data Quality Framework

sql

-- Create data quality view
CREATE OR REPLACE VIEW data_quality_metrics AS
WITH base_metrics AS (
    SELECT 
        'sales' AS table_name,
        COUNT(*) AS total_rows,
        COUNT(DISTINCT sales_id) AS unique_keys,
        SUM(CASE WHEN sales_id IS NULL THEN 1 ELSE 0 END) AS null_keys,
        SUM(CASE WHEN amount < 0 THEN 1 ELSE 0 END) AS negative_amounts,
        SUM(CASE WHEN sale_date > CURRENT_DATE() THEN 1 ELSE 0 END) AS future_dates
    FROM sales
),
completeness AS (
    SELECT 
        table_name,
        (total_rows - null_keys) * 100.0 / total_rows AS completeness_pct
    FROM base_metrics
),
validity AS (
    SELECT 
        table_name,
        (total_rows - negative_amounts - future_dates) * 100.0 / total_rows AS validity_pct
    FROM base_metrics
)
SELECT 
    b.table_name,
    b.total_rows,
    b.unique_keys,
    b.null_keys,
    b.negative_amounts,
    b.future_dates,
    c.completeness_pct,
    v.validity_pct,
    CASE 
        WHEN c.completeness_pct > 95 AND v.validity_pct > 95 THEN 'PASS'
        ELSE 'FAIL'
    END AS quality_status
FROM base_metrics b
JOIN completeness c ON b.table_name = c.table_name
JOIN validity v ON b.table_name = v.table_name;

-- Automated quality alerts
CREATE OR REPLACE TASK data_quality_check
    WAREHOUSE = monitoring_wh
    SCHEDULE = '1 HOUR'
AS
BEGIN
    DECLARE
        fail_count INTEGER;
    BEGIN
        SELECT COUNT(*)
        INTO fail_count
        FROM data_quality_metrics
        WHERE quality_status = 'FAIL';
        
        IF (fail_count > 0) THEN
            -- Call external alerting function
            CALL send_alert('Data quality check failed: ' || fail_count || ' tables');
            
            -- Log to audit table
            INSERT INTO quality_audit 
            SELECT *, CURRENT_TIMESTAMP()
            FROM data_quality_metrics
            WHERE quality_status = 'FAIL';
        END IF;
    END;
END;

Google BigQuery: Serverless Analytics

Architecture Overview

text

┌─────────────────────────────────────────────────────┐
│                BigQuery Architecture                 │
├─────────────────────────────────────────────────────┤
│           Serverless Query Engine                    │
│  • Dremel execution engine                          │
│  • Automatic scaling                                │
│  • Batch & streaming                                │
│  • BI Engine (in-memory acceleration)               │
├─────────────────────────────────────────────────────┤
│           Columnar Storage (Colossus)               │
│  • Capacitor format                                 │
│  • Automatic partitioning & clustering              │
│  • Long-term storage tier                           │
├─────────────────────────────────────────────────────┤
│           Federated Queries                         │
│  • BigQuery ML                                      │
│  • BigQuery GIS                                     │
│  • External data sources (GCS, Sheets, etc.)        │
└─────────────────────────────────────────────────────┘

Key Features

1. Serverless & Automatic Scaling

sql

-- No infrastructure management needed
-- Automatic scaling based on query complexity

-- Partitioned tables (automatic or manual)
CREATE TABLE sales.sales_data
PARTITION BY DATE(sale_date)
CLUSTER BY customer_id, product_id
AS
SELECT * FROM staging.sales_raw;

-- Query partitioned data (prunes partitions automatically)
SELECT * FROM sales.sales_data
WHERE sale_date BETWEEN '2024-01-01' AND '2024-01-31';

-- Automatic re-clustering (no maintenance needed)

2. BigQuery ML (Machine Learning)

sql

-- Create model directly in SQL
CREATE OR REPLACE MODEL sales.forecast_model
OPTIONS(
    MODEL_TYPE = 'ARIMA_PLUS',
    TIME_SERIES_TIMESTAMP_COL = 'sale_date',
    TIME_SERIES_DATA_COL = 'daily_sales',
    HOLIDAY_REGION = 'US'
) AS
SELECT 
    sale_date,
    SUM(amount) AS daily_sales
FROM sales.sales_data
WHERE sale_date >= '2020-01-01'
GROUP BY sale_date;

-- Make predictions
SELECT *
FROM ML.FORECAST(MODEL sales.forecast_model,
                 STRUCT(30 AS horizon));

-- Create classification model
CREATE OR REPLACE MODEL sales.churn_model
OPTIONS(
    MODEL_TYPE = 'BOOSTED_TREE_CLASSIFIER',
    INPUT_LABEL_COLS = ['churned'],
    AUTO_CLASS_WEIGHTS = TRUE
) AS
SELECT 
    customer_id,
    total_purchases,
    avg_order_value,
    days_since_last_purchase,
    churned
FROM sales.customer_features;

-- Evaluate model
SELECT *
FROM ML.EVALUATE(MODEL sales.churn_model);

-- Explain predictions
SELECT *
FROM ML.EXPLAIN_PREDICT(
    MODEL sales.churn_model,
    (SELECT * FROM new_customers),
    STRUCT(3 AS top_k_features)
);

3. BigQuery GIS (Geospatial)

sql

-- Geospatial data types
CREATE TABLE locations.properties (
    property_id STRING,
    address STRING,
    location GEOGRAPHY,
    price FLOAT64
);

-- Insert spatial data
INSERT INTO locations.properties
VALUES (
    'prop_001',
    '123 Main St',
    ST_GEOGPOINT(-122.4194, 37.7749),
    1000000.00
);

-- Spatial queries
SELECT 
    property_id,
    address,
    price,
    ST_DISTANCE(
        location,
        ST_GEOGPOINT(-122.408, 37.783)
    ) AS distance_from_office  -- in meters
FROM locations.properties
WHERE ST_DWITHIN(
    location,
    ST_GEOGPOINT(-122.408, 37.783),
    5000  -- 5km radius
)
ORDER BY distance_from_office;

-- Spatial aggregations
SELECT 
    zip_code,
    COUNT(*) AS property_count,
    AVG(price) AS avg_price,
    ST_CONVEXHULL(
        ST_UNION_AGG(location)
    ) AS coverage_area
FROM locations.properties
GROUP BY zip_code;

BigQuery Data Engineering Patterns

1. Incremental Loading with MERGE

sql

-- Slowly Changing Dimensions (SCD) Type 2
MERGE INTO dim_customers AS target
USING (
    SELECT 
        customer_id,
        customer_name,
        email,
        city,
        CURRENT_DATE() AS valid_from,
        DATE '9999-12-31' AS valid_to,
        TRUE AS is_current
    FROM staging.customers
    WHERE snapshot_date = CURRENT_DATE()
) AS source
ON target.customer_id = source.customer_id 
   AND target.is_current = TRUE
WHEN MATCHED AND (
    target.customer_name != source.customer_name
    OR target.email != source.email
    OR target.city != source.city
) THEN
    UPDATE SET 
        valid_to = CURRENT_DATE() - 1,
        is_current = FALSE
WHEN NOT MATCHED THEN
    INSERT (
        customer_id, customer_name, email, city,
        valid_from, valid_to, is_current
    )
    VALUES (
        source.customer_id, source.customer_name, 
        source.email, source.city,
        source.valid_from, source.valid_to,
        source.is_current
    );

-- Incremental fact table updates
MERGE INTO fact_sales AS target
USING (
    SELECT 
        sales_id,
        customer_id,
        product_id,
        sale_date,
        amount,
        quantity
    FROM staging.sales_incremental
    WHERE process_date = CURRENT_DATE()
) AS source
ON target.sales_id = source.sales_id
WHEN MATCHED THEN
    UPDATE SET 
        amount = source.amount,
        quantity = source.quantity,
        last_updated = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
    INSERT (
        sales_id, customer_id, product_id,
        sale_date, amount, quantity,
        created_at, last_updated
    )
    VALUES (
        source.sales_id, source.customer_id,
        source.product_id, source.sale_date,
        source.amount, source.quantity,
        CURRENT_TIMESTAMP(), CURRENT_TIMESTAMP()
    );

2. External Tables & Federation

sql

-- Query data directly from Google Cloud Storage
CREATE OR REPLACE EXTERNAL TABLE sales.external_sales_data
OPTIONS (
    format = 'PARQUET',
    uris = ['gs://my-bucket/sales/*.parquet'],
    hive_partition_uri_prefix = 'gs://my-bucket/sales/',
    uris = ['gs://my-bucket/sales/*']
);

-- Partitioned external table
CREATE OR REPLACE EXTERNAL TABLE sales.partitioned_sales
WITH PARTITION COLUMNS (
    year INTEGER,
    month INTEGER,
    day INTEGER
)
OPTIONS (
    format = 'PARQUET',
    uris = ['gs://my-bucket/sales/*/*/*/*.parquet'],
    hive_partition_uri_prefix = 'gs://my-bucket/sales/'
);

-- Query federated data sources
-- From Google Sheets
SELECT *
FROM EXTERNAL_QUERY(
    'connection_id',
    'SELECT * FROM spreadsheet_data'
);

-- From Cloud SQL
SELECT *
FROM EXTERNAL_QUERY(
    'us.connection_id',
    '''SELECT * FROM mysql_table 
       WHERE updated_at > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)'''
);

3. Materialized Views & BI Engine

sql

-- Create materialized view (auto-refreshed)
CREATE MATERIALIZED VIEW sales.daily_metrics
PARTITION BY metric_date
CLUSTER BY region, product_category
AS
SELECT 
    DATE(sale_date) AS metric_date,
    region,
    product_category,
    COUNT(DISTINCT customer_id) AS unique_customers,
    SUM(amount) AS total_revenue,
    SUM(CASE WHEN amount > 1000 THEN 1 ELSE 0 END) AS premium_sales,
    AVG(amount) AS avg_order_value
FROM sales.sales_data
GROUP BY 1, 2, 3;

-- Query materialized view (fast, pre-aggregated)
SELECT *
FROM sales.daily_metrics
WHERE metric_date = CURRENT_DATE() - 1;

-- Enable BI Engine (in-memory acceleration)
-- In Google Cloud Console or via bq command:
-- bq update --bi_reservation_size=2G project:location.reservation

Comparison: Snowflake vs BigQuery

Architecture Comparison

Performance Comparison

sql

-- SNOWFLAKE: Optimized for concurrent workloads
-- Pros: Predictable performance, workload isolation
-- Cons: Manual scaling decisions

-- BIGQUERY: Optimized for ad-hoc analytics
-- Pros: No capacity planning, automatic optimization
-- Cons: Less control over performance tuning

Cost Comparison

sql

-- SNOWFLAKE COST STRUCTURE
-- Compute: $2-4 per credit per hour (X-Small = 1 credit)
-- Storage: $23-40 per TB per month (compressed)
-- Example: X-Large (16 credits) running 8h/day = ~$1000/month
--         + 10TB storage = ~$300/month
-- Total: ~$1300/month

-- BIGQUERY COST STRUCTURE
-- On-demand: $5 per TB processed
-- Flat-rate: $2,000 per 500 slots per month
-- Storage: $20 per TB per month
-- Example: 10TB processed/month = $50
--          + 10TB storage = $200/month
-- Total: ~$250/month

-- BIG TIP: Use clustering and partitioning to reduce costs!

Integration Ecosystem

Snowflake Integrations:

python

# Python with Snowflake Connector
import snowflake.connector
import pandas as pd
import polars as pl

# Connect to Snowflake
ctx = snowflake.connector.connect(
    user='user',
    password='password',
    account='account.region',
    warehouse='transform_wh',
    database='sales_db',
    schema='public'
)

# Query with pandas
df_pandas = pd.read_sql('SELECT * FROM sales', ctx)

# Query with Polars (via pandas)
df_polars = pl.from_pandas(df_pandas)

# Use Snowpark (DataFrame API)
from snowflake.snowpark import Session
session = Session.builder.configs(connection_params).create()
df_snowpark = session.table('sales').filter(col('amount') > 1000)
df_pandas = df_snowpark.to_pandas()

# Load data from Polars
df = pl.read_parquet('data.parquet')
df.write_database(
    table_name='sales_staging',
    connection=engine_url,
    if_table_exists='replace'
)

BigQuery Integrations:

python

# Python with BigQuery Client
from google.cloud import bigquery
import pandas as pd
import polars as pl
from google.oauth2 import service_account

# Authenticate
credentials = service_account.Credentials.from_service_account_file(
    'key.json',
    scopes=['https://www.googleapis.com/auth/cloud-platform']
)
client = bigquery.Client(
    project='my-project',
    credentials=credentials
)

# Query to pandas
query = """
    SELECT * FROM `project.dataset.sales`
    WHERE amount > 1000
"""
df_pandas = client.query(query).to_dataframe()

# Query to Polars
df_polars = pl.from_pandas(df_pandas)

# Load data from Polars
df = pl.read_parquet('data.parquet')
job_config = bigquery.LoadJobConfig(
    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    schema=[
        bigquery.SchemaField('sale_id', 'STRING'),
        bigquery.SchemaField('amount', 'FLOAT64'),
    ]
)
job = client.load_table_from_dataframe(
    df.to_pandas(),
    'project.dataset.sales_staging',
    job_config=job_config
)
job.result()

# Use BigQuery Storage API (fast)
from google.cloud import bigquery_storage_v1
bqstorageclient = bigquery_storage_v1.BigQueryReadClient()
table = "projects/{}/datasets/{}/tables/{}".format(
    "my-project", "my_dataset", "my_table"
)
reader = bqstorageclient.read_rows(table)
df = reader.to_dataframe()  # Fast Arrow-based

Modern Data Engineering Patterns

1. Medallion Architecture Implementation

Snowflake Implementation:

sql

-- Bronze (Raw) Layer
CREATE OR REPLACE TABLE bronze.sales_raw (
    raw_data VARIANT,
    file_name STRING,
    load_timestamp TIMESTAMP_LTZ
);

-- Silver (Cleaned) Layer
CREATE OR REPLACE TABLE silver.sales_cleaned
CLUSTER BY (sale_date, customer_id)
AS
SELECT 
    raw_data:sale_id::STRING AS sale_id,
    raw_data:customer_id::STRING AS customer_id,
    raw_data:amount::FLOAT AS amount,
    raw_data:sale_date::DATE AS sale_date,
    raw_data:product_id::STRING AS product_id,
    CURRENT_TIMESTAMP() AS processed_at
FROM bronze.sales_raw
WHERE raw_data:sale_id IS NOT NULL
  AND raw_data:amount > 0;

-- Gold (Business) Layer
CREATE OR REPLACE TABLE gold.sales_aggregated
PARTITION BY sale_month
CLUSTER BY region, product_category
AS
SELECT 
    DATE_TRUNC('month', sale_date) AS sale_month,
    c.region,
    p.product_category,
    COUNT(DISTINCT s.customer_id) AS unique_customers,
    SUM(s.amount) AS total_revenue,
    SUM(CASE WHEN s.amount > 1000 THEN s.amount ELSE 0 END) AS premium_revenue,
    COUNT(*) AS total_transactions
FROM silver.sales_cleaned s
JOIN gold.customers c ON s.customer_id = c.customer_id
JOIN gold.products p ON s.product_id = p.product_id
GROUP BY 1, 2, 3;

BigQuery Implementation:

sql

-- Bronze Layer (External table)
CREATE OR REPLACE EXTERNAL TABLE bronze.sales_raw
OPTIONS (
    format = 'PARQUET',
    uris = ['gs://my-bucket/bronze/sales/*.parquet']
);

-- Silver Layer (Cleaned)
CREATE OR REPLACE TABLE silver.sales_cleaned
PARTITION BY DATE(sale_date)
CLUSTER BY customer_id, product_id
AS
SELECT 
    SAFE_CAST(raw_data.sale_id AS STRING) AS sale_id,
    SAFE_CAST(raw_data.customer_id AS STRING) AS customer_id,
    SAFE_CAST(raw_data.amount AS FLOAT64) AS amount,
    PARSE_DATE('%Y-%m-%d', raw_data.sale_date) AS sale_date,
    SAFE_CAST(raw_data.product_id AS STRING) AS product_id,
    CURRENT_TIMESTAMP() AS processed_at
FROM bronze.sales_raw
WHERE raw_data.sale_id IS NOT NULL
  AND SAFE_CAST(raw_data.amount AS FLOAT64) > 0;

-- Gold Layer (Business aggregates)
CREATE OR REPLACE TABLE gold.sales_metrics
PARTITION BY metric_date
CLUSTER BY region, product_category
AS
WITH daily_metrics AS (
    SELECT 
        DATE(s.sale_date) AS metric_date,
        c.region,
        p.product_category,
        COUNT(DISTINCT s.customer_id) AS dau,
        SUM(s.amount) AS daily_revenue,
        COUNT(*) AS daily_transactions
    FROM silver.sales_cleaned s
    JOIN gold.customers c USING(customer_id)
    JOIN gold.products p USING(product_id)
    GROUP BY 1, 2, 3
),
rolling_metrics AS (
    SELECT 
        metric_date,
        region,
        product_category,
        dau,
        daily_revenue,
        daily_transactions,
        AVG(dau) OVER (
            PARTITION BY region, product_category
            ORDER BY metric_date
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ) AS weekly_avg_dau,
        SUM(daily_revenue) OVER (
            PARTITION BY region, product_category
            ORDER BY metric_date
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS monthly_revenue
    FROM daily_metrics
)
SELECT * FROM rolling_metrics;

2. Data Vault 2.0 Implementation

Snowflake Data Vault:

sql

-- Hubs (Business Keys)
CREATE TABLE hub_customer (
    customer_hashkey VARCHAR(32) PRIMARY KEY,
    customer_id VARCHAR(50),
    load_date TIMESTAMP_NTZ,
    record_source VARCHAR(50)
);

-- Links (Relationships)
CREATE TABLE link_customer_order (
    link_hashkey VARCHAR(32) PRIMARY KEY,
    customer_hashkey VARCHAR(32),
    order_hashkey VARCHAR(32),
    load_date TIMESTAMP_NTZ,
    record_source VARCHAR(50)
);

-- Satellites (Descriptive Data)
CREATE TABLE sat_customer_details (
    customer_hashkey VARCHAR(32),
    hash_diff VARCHAR(32),
    customer_name VARCHAR(100),
    email VARCHAR(100),
    city VARCHAR(50),
    load_date TIMESTAMP_NTZ,
    load_end_date TIMESTAMP_NTZ,
    record_source VARCHAR(50),
    PRIMARY KEY (customer_hashkey, load_date)
);

-- Point-in-Time (PIT) Table
CREATE TABLE pit_customer AS
SELECT 
    h.customer_id,
    s.customer_name,
    s.email,
    s.city,
    s.load_date AS valid_from,
    LEAD(s.load_date, 1, '9999-12-31') 
        OVER (PARTITION BY h.customer_hashkey ORDER BY s.load_date) AS valid_to
FROM hub_customer h
JOIN sat_customer_details s ON h.customer_hashkey = s.customer_hashkey;

3. Real-time Streaming Pipelines

Snowflake Streaming:

sql

-- Create stream for CDC
CREATE STREAM sales_stream ON TABLE sales_raw;

-- Create task for real-time processing
CREATE TASK process_real_time_sales
    WAREHOUSE = streaming_wh
    SCHEDULE = '1 MINUTE'
AS
INSERT INTO sales_real_time
SELECT 
    sales_id,
    customer_id,
    amount,
    sale_timestamp,
    CURRENT_TIMESTAMP() AS processed_at
FROM sales_stream
WHERE metadata$action = 'INSERT';

-- Enable Snowpipe Streaming (Java/Python SDK)
/*
Python example:
from snowflake.ingest import SimpleIngestManager
ingest_manager = SimpleIngestManager(
    account='account',
    host='host.snowflakecomputing.com',
    user='user',
    pipe='sales_pipe',
    private_key=private_key
)
ingest_manager.ingest_files(['data.parquet'])
*/

BigQuery Streaming:

python

# BigQuery Streaming Insert
from google.cloud import bigquery
import datetime

client = bigquery.Client()

table_id = 'project.dataset.sales_real_time'

rows_to_insert = [
    {
        "sales_id": "sale_001",
        "customer_id": "cust_123",
        "amount": 100.50,
        "sale_timestamp": datetime.datetime.now().isoformat(),
        "processed_at": datetime.datetime.now().isoformat()
    }
]

errors = client.insert_rows_json(table_id, rows_to_insert)
if errors:
    print(f"Encountered errors: {errors}")

# Use Dataflow for high-volume streaming
"""
Apache Beam pipeline:
with beam.Pipeline() as p:
    (p
     | 'Read from PubSub' >> beam.io.ReadFromPubSub(
         subscription='projects/project/subscriptions/sales-sub')
     | 'Parse JSON' >> beam.Map(json.loads)
     | 'Transform' >> beam.Map(lambda x: {
         'sales_id': x['id'],
         'amount': float(x['amount']),
         'timestamp': x['timestamp']
     })
     | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
         'project:dataset.table',
         schema='sales_id:STRING,amount:FLOAT,timestamp:TIMESTAMP',
         write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
    )
"""

Security & Governance

Snowflake Security:

sql

-- Row Access Policies
CREATE OR REPLACE ROW ACCESS POLICY regional_access
AS (region VARCHAR) RETURNS BOOLEAN ->
    CURRENT_ROLE() IN ('ACCOUNTADMIN', 'SYSADMIN')
    OR CURRENT_ROLE() = region || '_ANALYST'
    OR IS_ROLE_IN_SESSION(region || '_ANALYST');

ALTER TABLE sales ADD ROW ACCESS POLICY regional_access ON (region);

-- Dynamic Data Masking
CREATE OR REPLACE MASKING POLICY email_mask AS (val STRING) RETURNS STRING ->
    CASE
        WHEN CURRENT_ROLE() IN ('ANALYST', 'DATA_SCIENTIST') THEN val
        ELSE REGEXP_REPLACE(val, '(.)(.*)(@.*)', '\\1***\\3')
    END;

ALTER TABLE users MODIFY COLUMN email 
SET MASKING POLICY email_mask;

-- Tag-based masking
CREATE TAG sensitivity_level 
    ALLOWED_VALUES 'public', 'internal', 'confidential', 'restricted';

ALTER TABLE users SET TAG sensitivity_level = 'confidential';

CREATE MASKING POLICY tag_based_mask AS (val STRING) RETURNS STRING ->
    CASE
        WHEN SYSTEM$GET_TAG_ON_CURRENT_COLUMN('sensitivity_level') = 'restricted'
             AND CURRENT_ROLE() NOT IN ('ADMIN') THEN '***'
        WHEN SYSTEM$GET_TAG_ON_CURRENT_COLUMN('sensitivity_level') = 'confidential'
             AND CURRENT_ROLE() NOT IN ('ADMIN', 'ANALYST') THEN '***'
        ELSE val
    END;

BigQuery Security:

sql

-- Column-level security (BigQuery Standard SQL)
CREATE VIEW sales_secure AS
SELECT 
    sales_id,
    sale_date,
    amount,
    CASE
        -- Mask email for non-admin users
        WHEN 'admin' IN UNNEST(SESSION_USER_EMAILS())
        THEN customer_email
        ELSE REGEXP_REPLACE(customer_email, r'^(.).*(.@.*)$', r'\1***\2')
    END AS customer_email,
    -- Show region only for authorized users
    IF('regional_analyst' IN UNNEST(SESSION_USER_ROLES()),
       region,
       NULL) AS region
FROM sales.sales_data;

-- Data Catalog Policy Tags
/*
In Google Cloud Console:
1. Create policy tag "Confidential"
2. Create taxonomy "Data_Classification"
3. Apply policy tag to sensitive columns
4. Set IAM policies on policy tag
*/

-- Authorized Views
CREATE VIEW sales_authorized
OPTIONS(
    privacy_policy= '{
        "aggregation_threshold_policy": {
            "threshold": 10
        }
    }'
) AS
SELECT 
    region,
    product_category,
    COUNT(*) AS transaction_count,
    SUM(amount) AS total_amount
FROM sales.sales_data
GROUP BY region, product_category;

-- Grant access to authorized view
GRANT `roles/bigquery.dataViewer`
ON VIEW sales_authorized
TO "group:analysts@company.com";

Performance Optimization

Snowflake Optimization:

sql

-- Query optimization
SELECT 
    -- Use qualified column names
    s.sales_id,
    c.customer_name,
    
    -- Avoid SELECT *
    s.amount,
    s.sale_date,
    
    -- Use WHERE on partition columns
    s.sale_date BETWEEN '2024-01-01' AND '2024-01-31',
    
    -- Use JOIN predicates
    c.customer_id = s.customer_id
    
FROM sales s
JOIN customers c 
    ON s.customer_id = c.customer_id
    AND s.region = c.region  -- Partition pruning
    
-- Cluster keys for large tables
CREATE TABLE large_table CLUSTER BY (date_id, customer_id);

-- Materialized views for frequent queries
CREATE MATERIALIZED VIEW mv_daily_sales AS
SELECT 
    DATE(sale_date) AS sale_day,
    product_id,
    SUM(amount) AS daily_sales
FROM sales
GROUP BY 1, 2;

-- Search optimization service
ALTER TABLE customers ADD SEARCH OPTIMIZATION ON (
    customer_name,
    email
);

-- Query acceleration service
ALTER SESSION SET USE_ACCELERATION = TRUE;

BigQuery Optimization:

sql

-- Use partitioning
CREATE TABLE sales_data
PARTITION BY DATE(sale_date)
CLUSTER BY customer_id, product_id
AS SELECT * FROM staging.sales;

-- Query partitioned data efficiently
SELECT * FROM sales_data
WHERE sale_date BETWEEN '2024-01-01' AND '2024-01-31';

-- Use clustering for large tables
CREATE TABLE large_table
PARTITION BY DATE(created_at)
CLUSTER BY user_id, event_type;

-- Approximate functions for large datasets
SELECT 
    APPROX_COUNT_DISTINCT(user_id) AS unique_users,
    APPROX_QUANTILES(amount, 4) AS amount_quartiles
FROM sales_data;

-- Use BI Engine for dashboard queries
/*
1. Enable BI Engine reservation
2. Create BI Engine optimized view
3. Route queries through BI Engine
*/

-- Optimize JOIN order
SELECT /*+ JOIN_ORDER(customers, orders, products) */
    c.customer_name,
    p.product_name,
    SUM(o.amount) AS total_spent
FROM customers c
JOIN orders o ON c.customer_id = o.customer_id
JOIN products p ON o.product_id = p.product_id
GROUP BY 1, 2;

Cost Optimization Strategies

Snowflake Cost Control:

sql

-- Monitor warehouse usage
SELECT 
    warehouse_name,
    SUM(credits_used) AS total_credits,
    AVG(credits_used) AS avg_credits_per_query,
    COUNT(*) AS query_count
FROM snowflake.account_usage.warehouse_metering_history
WHERE start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP())
GROUP BY warehouse_name
ORDER BY total_credits DESC;

-- Auto-suspend warehouses
ALTER WAREHOUSE analytics_wh 
SET AUTO_SUSPEND = 300;  -- 5 minutes

-- Multi-cluster warehouses for concurrency
ALTER WAREHOUSE etl_wh SET
    MIN_CLUSTER_COUNT = 1,
    MAX_CLUSTER_COUNT = 5,
    SCALING_POLICY = 'ECONOMY';

-- Use resource monitors
CREATE RESOURCE MONITOR etl_budget
    WITH CREDIT_QUOTA = 1000
    FREQUENCY = MONTHLY
    START_TIMESTAMP = IMMEDIATE
    TRIGGERS 
        ON 80 PERCENT DO NOTIFY
        ON 100 PERCENT DO SUSPEND;

ALTER WAREHOUSE etl_wh SET RESOURCE_MONITOR = etl_budget;

-- Query optimization to reduce costs
-- 1. Use result cache (same query within 24 hours)
-- 2. Filter early with WHERE
-- 3. Avoid unnecessary columns
-- 4. Use materialized views for repeated queries

BigQuery Cost Control:

sql

-- Monitor query costs
SELECT
    job_id,
    query,
    total_bytes_processed / POWER(1024, 3) AS gb_processed,
    total_bytes_processed * 5 / POWER(1024, 9) AS estimated_cost_usd,
    user_email
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
ORDER BY total_bytes_processed DESC
LIMIT 100;

-- Use partitioning and clustering
CREATE TABLE sales_partitioned
PARTITION BY DATE(sale_date)
CLUSTER BY customer_id, product_id
AS SELECT * FROM sales;

-- Query only necessary partitions
SELECT * FROM sales_partitioned
WHERE sale_date = CURRENT_DATE()  -- Only scans today's partition

-- Use flat-rate pricing for predictable workloads
/*
Reserve slots:
- 100 slots: $2,000/month
- 500 slots: $8,000/month
- Flex slots: $2/slot/hour
*/

-- Optimize storage costs
-- Convert to long-term storage (after 90 days, 50% cheaper)
SELECT
    table_name,
    active_bytes / POWER(1024, 3) AS active_gb,
    long_term_bytes / POWER(1024, 3) AS long_term_gb,
    total_rows
FROM `dataset`.INFORMATION_SCHEMA.TABLE_STORAGE;

-- Set up cost controls
/*
1. Budget alerts in Cloud Console
2. IAM constraints: constraints/compute.disableSerialPortAccess
3. Query quotas per user/project
4. BigQuery Reservations API for slot management
*/

Integration with Orchestration Tools

Airflow + Snowflake:

python

from airflow import DAG
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from datetime import datetime

default_args = {
    'owner': 'data_team',
    'start_date': datetime(2024, 1, 1),
}

with DAG('snowflake_pipeline', 
         schedule_interval='@daily',
         default_args=default_args) as dag:
    
    extract = SnowflakeOperator(
        task_id='extract_data',
        sql='''
            COPY INTO @sales_stage/sales_{{ ds }}.parquet
            FROM sales_raw
            WHERE sale_date = '{{ ds }}'
            FILE_FORMAT = (TYPE = PARQUET)
        ''',
        snowflake_conn_id='snowflake_default',
        warehouse='transform_wh',
        database='sales_db',
        schema='public',
    )
    
    transform = SnowflakeOperator(
        task_id='transform_data',
        sql='''
            CREATE OR REPLACE TABLE sales_processed AS
            SELECT 
                s.*,
                c.customer_name,
                c.region,
                p.product_category,
                s.amount * 1.1 AS amount_with_tax
            FROM sales_raw s
            LEFT JOIN customers c USING(customer_id)
            LEFT JOIN products p USING(product_id)
            WHERE s.sale_date = '{{ ds }}'
        ''',
    )
    
    load = SnowflakeOperator(
        task_id='load_to_dwh',
        sql='''
            MERGE INTO sales_dwh AS target
            USING sales_processed AS source
            ON target.sales_id = source.sales_id
            WHEN MATCHED THEN UPDATE SET 
                target.amount = source.amount,
                target.last_updated = CURRENT_TIMESTAMP()
            WHEN NOT MATCHED THEN INSERT 
                (sales_id, customer_id, amount, sale_date)
                VALUES (source.sales_id, source.customer_id, 
                        source.amount, source.sale_date);
        ''',
    )
    
    quality_check = SnowflakeOperator(
        task_id='data_quality_check',
        sql='''
            INSERT INTO quality_checks (check_date, check_type, result)
            SELECT 
                '{{ ds }}',
                'row_count',
                CASE 
                    WHEN COUNT(*) > 0 THEN 'PASS'
                    ELSE 'FAIL'
                END
            FROM sales_processed
            WHERE sale_date = '{{ ds }}';
            
            INSERT INTO quality_checks (check_date, check_type, result)
            SELECT 
                '{{ ds }}',
                'negative_amounts',
                CASE 
                    WHEN SUM(CASE WHEN amount < 0 THEN 1 ELSE 0 END) = 0 
                    THEN 'PASS' 
                    ELSE 'FAIL'
                END
            FROM sales_processed
            WHERE sale_date = '{{ ds }}';
        ''',
    )
    
    extract >> transform >> load >> quality_check

Dagster + BigQuery:

python

from dagster import Definitions, asset, OpExecutionContext
from dagster_gcp import BigQueryResource
from google.cloud import bigquery
import pandas as pd
import polars as pl

@asset
def extract_from_bigquery(context: OpExecutionContext, bigquery_resource: BigQueryResource) -> pl.DataFrame:
    """Extract data from BigQuery using Dagster resource"""
    client = bigquery_resource.get_client()
    
    query = """
        SELECT 
            customer_id,
            order_date,
            amount,
            product_id,
            region
        FROM `project.dataset.sales`
        WHERE order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
    """
    
    # Query to pandas
    df_pandas = client.query(query).to_dataframe()
    
    # Convert to Polars for processing
    df = pl.from_pandas(df_pandas)
    
    context.log.info(f"Extracted {len(df)} rows from BigQuery")
    
    return df

@asset
def transform_with_polars(context: OpExecutionContext, extract_from_bigquery: pl.DataFrame) -> pl.DataFrame:
    """Transform data using Polars"""
    df = extract_from_bigquery
    
    transformed = (
        df.lazy()
        .with_columns([
            pl.col('order_date').cast(pl.Date).alias('date'),
            (pl.col('amount') * 1.1).alias('amount_with_tax'),
            pl.col('region').cast(pl.Categorical).alias('region_cat'),
        ])
        .group_by(['date', 'region', 'product_id'])
        .agg([
            pl.col('amount').sum().alias('daily_revenue'),
            pl.col('customer_id').n_unique().alias('unique_customers'),
            pl.col('amount').mean().alias('avg_order_value'),
        ])
        .collect()
    )
    
    context.log.info(f"Transformed to {len(transformed)} aggregated rows")
    
    return transformed

@asset
def load_to_bigquery(
    context: OpExecutionContext, 
    bigquery_resource: BigQueryResource,
    transform_with_polars: pl.DataFrame
) -> str:
    """Load transformed data back to BigQuery"""
    client = bigquery_resource.get_client()
    
    # Convert Polars to pandas for BigQuery load
    df_pandas = transform_with_polars.to_pandas()
    
    # Define table reference
    table_ref = client.dataset('analytics').table('sales_aggregated')
    
    # Configure load job
    job_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        schema=[
            bigquery.SchemaField('date', 'DATE'),
            bigquery.SchemaField('region', 'STRING'),
            bigquery.SchemaField('product_id', 'STRING'),
            bigquery.SchemaField('daily_revenue', 'FLOAT64'),
            bigquery.SchemaField('unique_customers', 'INT64'),
            bigquery.SchemaField('avg_order_value', 'FLOAT64'),
        ]
    )
    
    # Load data
    job = client.load_table_from_dataframe(
        df_pandas, table_ref, job_config=job_config
    )
    job.result()  # Wait for completion
    
    context.log.info(f"Loaded {job.output_rows} rows to BigQuery")
    
    return f"Successfully loaded to {table_ref}"

# Define resources and assets
defs = Definitions(
    assets=[extract_from_bigquery, transform_with_polars, load_to_bigquery],
    resources={
        'bigquery_resource': BigQueryResource(
            project='my-gcp-project',
            location='US'
        )
    }
)

Future Trends & Considerations

Emerging Patterns:

1. Unified Storage Layer (Iceberg/Hudi/Delta)

sql

-- Snowflake with Iceberg Tables
CREATE ICEBERG TABLE sales_iceberg
    EXTERNAL_VOLUME = my_volume
    CATALOG = 'aws_glue'
    EXTERNAL_TABLE = 'sales_db.sales_table'
    INTEGRATION = 'aws_integration';

-- BigQuery with BigLake (Iceberg)
CREATE EXTERNAL TABLE sales_biglake
WITH CONNECTION `project.region.my_connection`
OPTIONS (
    format = 'ICEBERG',
    uris = ['gs://my-bucket/sales/*'],
    hive_partition_uri_prefix = 'gs://my-bucket/sales/'
);

2. Reverse ETL (Operational Analytics)

python

# Send data from warehouse back to operational systems
# Using tools like Hightouch, Census, or custom pipelines

# Example: Sync customer segments to CRM
SELECT 
    customer_id,
    CASE 
        WHEN total_spent > 10000 THEN 'VIP'
        WHEN total_spent > 1000 THEN 'Premium'
        ELSE 'Standard'
    END AS customer_tier,
    last_purchase_date,
    avg_order_value
FROM customer_metrics
WHERE updated_at >= CURRENT_TIMESTAMP() - INTERVAL '1 hour'

3. Data Mesh Implementation

sql

-- Domain-oriented data products
-- Each domain manages their own data as a product

-- Marketing domain
CREATE DATA PRODUCT marketing.customer_behavior AS
SELECT 
    customer_id,
    visit_frequency,
    preferred_categories,
    conversion_rate
FROM web_analytics
WHERE data_domain = 'marketing';

-- Finance domain
CREATE DATA PRODUCT finance.revenue_reporting AS
SELECT 
    date,
    product_line,
    region,
    revenue,
    cogs,
    margin
FROM sales_joined
WHERE data_domain = 'finance';

Recommendations by Use Case:

Choose Snowflake when:

• You need predictable performance with workload isolation

• Your team prefers traditional warehouse concepts

• You need strong data sharing capabilities

• You're in a multi-cloud environment

• You need fine-grained cost control per workload

Choose BigQuery when:

• You want fully serverless operation

• You're heavily invested in Google Cloud ecosystem

• You need built-in ML/AI capabilities

• You have unpredictable or ad-hoc query patterns

• You prefer pay-per-query pricing model

Hybrid Approach:

• Use Snowflake for structured, predictable ETL workloads

• Use BigQuery for exploratory analytics and ML

• Use both for disaster recovery and vendor diversification

Migration Considerations:

Snowflake to BigQuery Migration:

sql

-- Use BigQuery Data Transfer Service
-- Or export from Snowflake to GCS, then load to BigQuery

-- Export from Snowflake
COPY INTO 'gcs://my-bucket/snowflake-export/'
FROM sales_table
FILE_FORMAT = (TYPE = PARQUET);

-- Load to BigQuery
LOAD DATA INTO my_dataset.sales_table
FROM FILES (
    format = 'PARQUET',
    uris = ['gs://my-bucket/snowflake-export/*.parquet']
);

BigQuery to Snowflake Migration:

sql

-- Export from BigQuery to GCS
EXPORT DATA
OPTIONS(
    uri='gs://my-bucket/bigquery-export/*.parquet',
    format='PARQUET',
    overwrite=true
) AS
SELECT * FROM my_dataset.sales_table;

-- Load to Snowflake
COPY INTO sales_table
FROM 'gcs://my-bucket/bigquery-export/'
FILE_FORMAT = (TYPE = PARQUET)
STORAGE_INTEGRATION = gcs_integration;

Modern data warehouses have evolved from simple storage systems to comprehensive data platforms. Snowflake excels with its separation of compute/storage and enterprise features, while BigQuery leads in serverless operation and integrated ML capabilities.

Key Takeaways:

1. Snowflake is ideal for predictable workloads requiring control

2. BigQuery excels at ad-hoc analytics and integrated AI/ML

3. Both support modern data engineering patterns (medallion, data vault)

4. Cost optimization requires understanding each platform's pricing model

5. Security and governance features are mature in both platforms

6. Integration with orchestration tools is well-supported

The choice often depends on existing cloud investments, team skills, and specific workload patterns. Many organizations successfully use both platforms for different use cases within their data ecosystem.