Snowflake Complete Guide

-- Installeer SnowSQL: https://docs.snowflake.com/en/user-guide/snowsql-install-config

-- Verbind met Snowflake
!connect account=yz12345.eu-west-1
user=jansmit
password=************
warehouse=COMPUTE_WH
database=DEMO_DB
schema=PUBLIC
role=SYSADMIN;

-- Maak je eerste database
CREATE OR REPLACE DATABASE SALES_DB 
COMMENT = 'Productie database voor sales data';

-- Maak een nieuw schema
CREATE OR REPLACE SCHEMA SALES_DB.RAW 
COMMENT = 'Raw landing zone voor bron systemen';

-- Maak je eerste tabel
CREATE OR REPLACE TABLE SALES_DB.RAW.CUSTOMERS (
    CUSTOMER_ID NUMBER(38,0) AUTOINCREMENT START 1 INCREMENT 1,
    FIRST_NAME VARCHAR(100),
    LAST_NAME VARCHAR(100),
    EMAIL VARCHAR(255) UNIQUE,
    PHONE VARCHAR(20),
    ADDRESS VARCHAR(500),
    CITY VARCHAR(100),
    COUNTRY VARCHAR(100),
    CREATED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP(),
    UPDATED_AT TIMESTAMP_NTZ(9) DEFAULT CURRENT_TIMESTAMP()
) 
COMMENT = 'Klant master data tabel';

-- Maak een stage voor data loading
CREATE OR REPLACE STAGE SALES_DB.RAW.CUSTOMER_STAGE
    URL = 's3://your-bucket/customer-data/'
    STORAGE_INTEGRATION = YOUR_S3_INTEGRATION
    FILE_FORMAT = (TYPE = 'CSV' SKIP_HEADER = 1);

-- Load data van S3
COPY INTO SALES_DB.RAW.CUSTOMERS
FROM @SALES_DB.RAW.CUSTOMER_STAGE
PATTERN = '.*customers.*\.csv'
ON_ERROR = 'CONTINUE';

-- Query data met Time Travel
SELECT * FROM SALES_DB.RAW.CUSTOMERS
AT(OFFSET => -60*5); -- Data van 5 minuten geleden

-- Toon alle databases
SHOW DATABASES;

-- Toon kosten van vandaag
SELECT 
    WAREHOUSE_NAME,
    SUM(CREDITS_USED) AS TOTAL_CREDITS,
    SUM(CREDITS_USED_COMPUTE) AS COMPUTE_CREDITS,
    SUM(CREDITS_USED_CLOUD_SERVICES) AS SERVICES_CREDITS
FROM INFORMATION_SCHEMA.WAREHOUSE_METERING_HISTORY
WHERE START_TIME >= CURRENT_DATE()
GROUP BY WAREHOUSE_NAME
ORDER BY TOTAL_CREDITS DESC;
        

# Installeer: pip install snowflake-connector-python snowflake-snowpark-python

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, sum as sum_, avg, count
import pandas as pd
import json

def create_snowflake_session():
    """Maak een Snowflake Snowpark sessie"""
    connection_params = {
        "account": "yz12345.eu-west-1",
        "user": "DATAPARTNER_USER",
        "password": "your_secure_password",
        "role": "DATA_ENGINEER",
        "warehouse": "COMPUTE_WH",
        "database": "ANALYTICS_DB",
        "schema": "PRODUCTION"
    }
    
    session = Session.builder.configs(connection_params).create()
    print("✅ Snowflake Snowpark sessie succesvol aangemaakt")
    return session

def demonstrate_snowpark_features(session):
    """Demonstreer Snowpark functionaliteiten"""
    
    # 1. Laad data in Snowpark DataFrame (lazy evaluation)
    sales_df = session.table("SALES_DB.RAW.SALES_TRANSACTIONS")
    
    print("Aantal rijen in sales:", sales_df.count())
    print("Schema van sales:", sales_df.schema)
    
    # 2. Data transformaties
    daily_sales = sales_df \
        .group_by("TRANSACTION_DATE") \
        .agg(
            sum_("AMOUNT").alias("TOTAL_SALES"),
            avg("AMOUNT").alias("AVG_TRANSACTION"),
            count("*").alias("TRANSACTION_COUNT")
        ) \
        .filter(col("TOTAL_SALES") > 10000) \
        .sort(col("TRANSACTION_DATE").desc())
    
    # 3. Toon SQL die gegenereerd wordt
    print("Generated SQL:")
    print(daily_sales.queries['queries'][0])
    
    # 4. Materialiseer de resultaten
    results = daily_sales.limit(10).collect()
    print("\nTop 10 verkoopdagen:")
    for row in results:
        print(f"{row.TRANSACTION_DATE}: €{row.TOTAL_SALES:,.2f}")
    
    # 5. Pandas integratie
    pandas_df = daily_sales.to_pandas()
    print(f"\nPandas DataFrame shape: {pandas_df.shape}")
    
    return pandas_df

def create_udf_for_calculations(session):
    """Maak een User Defined Function in Snowflake"""
    
    # Register een Python UDF
    @udf(name="calculate_discount", 
         input_types=[FloatType(), FloatType()], 
         return_type=FloatType(),
         is_permanent=True,
         stage_location="@PYTHON_UDFS",
         packages=["pandas"])
    def calculate_discount(price, discount_percentage):
        """Bereken de prijs na korting"""
        if discount_percentage > 50:
            discount_percentage = 50  # Max 50% korting
        return price * (1 - discount_percentage/100)
    
    print("✅ UDF succesvol geregistreerd in Snowflake")
    
    # Gebruik de UDF in een query
    result = session.sql("""
        SELECT 
            PRODUCT_ID,
            PRICE,
            DISCOUNT_PERCENTAGE,
            calculate_discount(PRICE, DISCOUNT_PERCENTAGE) as FINAL_PRICE
        FROM SALES_DB.RAW.PRODUCTS
        WHERE DISCOUNT_PERCENTAGE > 0
        LIMIT 10
    """).collect()
    
    return result

if __name__ == "__main__":
    # Maak Snowpark sessie
    session = create_snowflake_session()
    
    try:
        # Demo Snowpark features
        sales_data = demonstrate_snowpark_features(session)
        
        # Maak en gebruik UDF
        udf_results = create_udf_for_calculations(session)
        
        # Zero-copy cloning demo
        print("\n--- Zero-Copy Cloning Demo ---")
        session.sql("CREATE OR REPLACE DATABASE SALES_DB_CLONE CLONE SALES_DB").collect()
        print("Database succesvol gekloond zonder extra storage kosten!")
        
        # Time Travel demo
        print("\n--- Time Travel Demo ---")
        history = session.sql("""
            SELECT 
                MIN(CREATED) AS OLDEST_DATA,
                MAX(CREATED) AS NEWEST_DATA
            FROM SALES_DB.RAW.SALES_TRANSACTIONS
            AT(OFFSET => -3600) -- 1 uur geleden
        """).collect()
        print(f"Data van 1 uur geleden: {history[0]}")
        
    finally:
        session.close()
        print("\n✅ Sessie gesloten")