📜 Cheatsheet Phase 3

❄ Snowflake - Commandes Essentielles

-- Creer un warehouse
CREATE WAREHOUSE my_wh
  WITH WAREHOUSE_SIZE = 'MEDIUM'
  AUTO_SUSPEND = 300
  AUTO_RESUME = TRUE
  MIN_CLUSTER_COUNT = 1
  MAX_CLUSTER_COUNT = 3;

-- Time Travel (jusqu'a 90 jours)
SELECT * FROM my_table
  AT(TIMESTAMP => '2026-01-15 10:00:00'::TIMESTAMP);

SELECT * FROM my_table
  BEFORE(STATEMENT => '01b2c3d4-...');

-- Zero-Copy Cloning
CREATE TABLE dev_table CLONE prod_table;
CREATE DATABASE dev_db CLONE prod_db;
CREATE SCHEMA test_schema CLONE prod_schema;

-- Streams (CDC)
CREATE STREAM my_stream ON TABLE source_table;
SELECT * FROM my_stream; -- changes only

-- Tasks (scheduling)
CREATE TASK my_task
  WAREHOUSE = my_wh
  SCHEDULE = 'USING CRON 0 9 * * * UTC'
AS
  INSERT INTO target SELECT * FROM my_stream;

-- Data Sharing
CREATE SHARE my_share;
GRANT USAGE ON DATABASE db TO SHARE my_share;
GRANT SELECT ON TABLE db.schema.t TO SHARE my_share;

-- Snowpipe (auto-ingest)
CREATE PIPE my_pipe AUTO_INGEST = TRUE AS
  COPY INTO my_table FROM @my_stage;

🏗 Snowflake Architecture

┌─────────────────────────────────┐
│       Cloud Services Layer      │
│  (Auth, Metadata, Optimizer)    │
├─────────────────────────────────┤
│       Compute Layer             │
│  ┌─────┐ ┌─────┐ ┌─────┐     │
│  │ WH1 │ │ WH2 │ │ WH3 │     │
│  │ ETL │ │ BI  │ │ ML  │     │
│  └─────┘ └─────┘ └─────┘     │
├─────────────────────────────────┤
│       Storage Layer             │
│    (Micro-partitions, S3/GCS)   │
│    Columnar, Compressed, Immutable│
└─────────────────────────────────┘

☁ Google BigQuery

-- Partitionnement
CREATE TABLE dataset.my_table (
  id INT64,
  event_date DATE,
  user_id STRING,
  amount NUMERIC
)
PARTITION BY event_date
CLUSTER BY user_id;

-- BigQuery ML - Modele en SQL
CREATE OR REPLACE MODEL dataset.my_model
OPTIONS(model_type='logistic_reg',
        input_label_cols=['is_churned'])
AS SELECT * FROM dataset.training_data;

-- Prediction
SELECT * FROM ML.PREDICT(
  MODEL dataset.my_model,
  TABLE dataset.new_data);

-- Federated Query
SELECT * FROM
  EXTERNAL_QUERY('connection_id',
  'SELECT * FROM postgres_table');

-- Materialized View
CREATE MATERIALIZED VIEW dataset.mv AS
SELECT date, SUM(revenue) as total
FROM dataset.sales
GROUP BY date;

-- Scheduled Query
-- Via Console ou API: bq mk --transfer_config

⚡ Databricks & Apache Spark

# PySpark - DataFrame Operations
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, avg

spark = SparkSession.builder \
    .appName("DataPipeline") \
    .getOrCreate()

# Lire Delta Lake
df = spark.read.format("delta").load("/data/bronze/events")

# Transformations
result = (df
    .filter(col("event_date") >= "2026-01-01")
    .groupBy("category")
    .agg(
        sum("amount").alias("total_amount"),
        avg("amount").alias("avg_amount")
    )
    .orderBy(col("total_amount").desc())
)

# Ecrire en Delta
result.write.format("delta") \
    .mode("overwrite") \
    .save("/data/gold/category_summary")

# Delta Lake - MERGE (Upsert)
from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, "/data/silver/users")
target.alias("t").merge(
    source_df.alias("s"),
    "t.user_id = s.user_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# Unity Catalog
spark.sql("USE CATALOG main")
spark.sql("USE SCHEMA analytics")
spark.sql("GRANT SELECT ON TABLE t TO GROUP analysts")

📈 Comparaison Cloud Data Warehouses

🔨 dbt (data build tool)

# Structure projet dbt
my_project/
├── dbt_project.yml
├── profiles.yml        # connexion DWH
├── models/
│   ├── staging/        # 1:1 avec sources
│   │   ├── stg_orders.sql
│   │   └── _stg_models.yml
│   ├── intermediate/   # logique metier
│   │   └── int_orders_enriched.sql
│   └── marts/          # tables finales
│       ├── fct_orders.sql
│       └── dim_customers.sql
├── tests/
│   └── assert_positive_amounts.sql
├── macros/
│   └── cents_to_dollars.sql
├── seeds/
│   └── country_codes.csv
└── snapshots/
    └── snap_customers.sql

-- Materialization types
-- models/staging/stg_orders.sql
{{ config(materialized='view') }}

SELECT
    id as order_id,
    customer_id,
    order_date,
    status,
    amount_cents / 100.0 as amount
FROM {{ source('raw', 'orders') }}

-- models/marts/fct_orders.sql
{{ config(
    materialized='incremental',
    unique_key='order_id',
    on_schema_change='append_new_columns'
) }}

SELECT
    o.order_id,
    o.customer_id,
    c.customer_name,
    o.order_date,
    o.amount,
    o.status
FROM {{ ref('stg_orders') }} o
JOIN {{ ref('dim_customers') }} c
  ON o.customer_id = c.customer_id
{% if is_incremental() %}
WHERE o.order_date > (
    SELECT MAX(order_date) FROM {{ this }}
)
{% endif %}

🔎 dbt Tests & Macros

# _stg_models.yml - Tests generiques
version: 2
models:
  - name: stg_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: ['pending','shipped','delivered','cancelled']
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id

# dbt_expectations (package)
      - name: amount
        tests:
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 100000

-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name, precision=2) %}
    ROUND({{ column_name }} / 100.0, {{ precision }})
{% endmacro %}

-- Usage dans un modele:
SELECT
    order_id,
    {{ cents_to_dollars('amount_cents') }} as amount
FROM {{ source('raw', 'orders') }}

-- Snapshot (SCD Type 2)
{% snapshot snap_customers %}
{{ config(
    target_schema='snapshots',
    unique_key='customer_id',
    strategy='timestamp',
    updated_at='updated_at'
) }}
SELECT * FROM {{ source('raw', 'customers') }}
{% endsnapshot %}

🌈 Apache Airflow

# DAG avec TaskFlow API
from airflow.decorators import dag, task
from datetime import datetime

@dag(
    schedule='0 6 * * *',  # 6h chaque jour
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['production', 'dbt']
)
def daily_pipeline():

    @task()
    def extract():
        """Extract data from API"""
        import requests
        data = requests.get('https://api.example.com/data').json()
        return data

    @task()
    def transform(raw_data):
        """Clean and transform data"""
        cleaned = [
            {**row, 'amount': float(row['amount'])}
            for row in raw_data
            if row['status'] != 'cancelled'
        ]
        return cleaned

    @task()
    def load(data):
        """Load to data warehouse"""
        from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
        hook = SnowflakeHook('snowflake_conn')
        hook.insert_rows('target_table', data)

    @task()
    def run_dbt():
        """Run dbt transformations"""
        import subprocess
        subprocess.run(['dbt', 'build', '--target', 'prod'], check=True)

    # Define dependencies
    raw = extract()
    clean = transform(raw)
    load(clean) >> run_dbt()

daily_pipeline()

🔄 Dagster & Prefect

# Dagster - Software-Defined Assets
from dagster import asset, AssetExecutionContext
from dagster_dbt import DbtCliResource, dbt_assets

@asset(group_name="ingestion")
def raw_orders(context: AssetExecutionContext):
    """Extract orders from source"""
    import pandas as pd
    df = pd.read_csv("s3://bucket/orders.csv")
    df.to_parquet("/data/raw/orders.parquet")
    context.log.info(f"Loaded {len(df)} rows")
    return df

@asset(deps=[raw_orders], group_name="transform")
def clean_orders():
    """Clean and validate orders"""
    import pandas as pd
    df = pd.read_parquet("/data/raw/orders.parquet")
    return df.dropna(subset=['order_id', 'amount'])

# Prefect - Flows & Tasks
from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(cache_key_fn=task_input_hash,
      cache_expiration=timedelta(hours=1),
      retries=3)
def extract_data(url: str) -> dict:
    import httpx
    return httpx.get(url).json()

@task
def transform_data(data: dict) -> list:
    return [r for r in data['results'] if r['active']]

@task
def load_data(records: list):
    # Load to database
    print(f"Loaded {len(records)} records")

@flow(name="ETL Pipeline", log_prints=True)
def etl_pipeline():
    data = extract_data("https://api.example.com/data")
    clean = transform_data(data)
    load_data(clean)

etl_pipeline()

📂 Formats de Fichiers Data

# Python - Lire/Ecrire Parquet
import pandas as pd

# Parquet
df.to_parquet('data.parquet', compression='snappy')
df = pd.read_parquet('data.parquet')

# Avec PyArrow (plus performant)
import pyarrow.parquet as pq
table = pq.read_table('data.parquet',
    columns=['col1', 'col2'])  # lecture selective

🗃 Open Table Formats

-- Delta Lake (Databricks SQL)
CREATE TABLE events USING DELTA
LOCATION 's3://bucket/events';

-- Iceberg (Spark)
CREATE TABLE catalog.db.events
USING iceberg
PARTITIONED BY (days(event_date));

-- Iceberg - Schema evolution
ALTER TABLE catalog.db.events
ADD COLUMNS (new_col STRING);
-- Pas besoin de rewrite!

📤 Outils d'Ingestion & CDC

# Airbyte - Docker Compose
git clone https://github.com/airbytehq/airbyte.git
cd airbyte
./run-ab-platform.sh
# UI: http://localhost:8000

# Debezium - Kafka Connect config
{
  "name": "postgres-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.dbname": "mydb",
    "topic.prefix": "cdc",
    "plugin.name": "pgoutput"
  }
}

🚀 Architecture Pipeline Moderne

     Sources              Ingestion          Storage
  ┌──────────┐        ┌──────────┐      ┌──────────┐
  │ Postgres │───────>│ Airbyte/ │─────>│  Bronze  │
  │ APIs     │───────>│ Fivetran │─────>│ (Raw S3) │
  │ Fichiers │───────>│ Debezium │─────>│  Parquet │
  └──────────┘        └──────────┘      └────┬─────┘
                                              │
     Transformation        Processing         │
  ┌──────────────┐     ┌──────────┐          │
  │     dbt      │<────│  Silver  │<─────────┘
  │  (staging)   │────>│ (Clean)  │
  │  (marts)     │     └────┬─────┘
  └──────────────┘          │
                             v
     Serving            ┌──────────┐
  ┌──────────┐          │   Gold   │
  │ Tableau  │<─────────│(Business)│
  │ Looker   │<─────────│          │
  │ Metabase │          └──────────┘
  └──────────┘
                    Orchestration: Airflow/Dagster
                    Quality: Great Expectations/Soda
                    Catalog: DataHub/Atlan

⌨ dbt CLI Reference Rapide

🏅 Medallion Architecture avec dbt

-- BRONZE: Raw data (view ou table)
-- models/staging/stg_raw_events.sql
{{ config(materialized='view') }}
SELECT
    _airbyte_extracted_at as loaded_at,
    *
FROM {{ source('raw_airbyte', 'events') }}

-- SILVER: Cleaned & typed
-- models/intermediate/int_events_cleaned.sql
{{ config(materialized='incremental',
          unique_key='event_id') }}
SELECT
    event_id,
    CAST(event_date AS DATE) as event_date,
    LOWER(TRIM(user_email)) as user_email,
    COALESCE(amount, 0) as amount,
    CASE
        WHEN status IN ('active','completed') THEN status
        ELSE 'unknown'
    END as status
FROM {{ ref('stg_raw_events') }}
{% if is_incremental() %}
WHERE loaded_at > (SELECT MAX(loaded_at) FROM {{ this }})
{% endif %}

-- GOLD: Business metrics
-- models/marts/fct_daily_revenue.sql
{{ config(materialized='table') }}
SELECT
    event_date,
    COUNT(DISTINCT user_email) as unique_users,
    COUNT(*) as total_events,
    SUM(amount) as total_revenue,
    AVG(amount) as avg_order_value
FROM {{ ref('int_events_cleaned') }}
WHERE status = 'completed'
GROUP BY event_date

🗜 Algorithmes de Compression

# Python - Parquet avec compression Zstd
import pandas as pd

df.to_parquet(
    'data/orders/',
    engine='pyarrow',
    compression='zstd',           # Meilleur ratio/vitesse
    partition_cols=['date'],       # Partitionnement Hive
    row_group_size=100_000,
)

# Lecture avec predicate pushdown
import pyarrow.parquet as pq
table = pq.read_table('data/orders/',
    columns=['id', 'amount'],     # Column pruning
    filters=[('date', '>=', '2024-06-01')]  # Partition pruning
)

🔄 Patterns CDC (Change Data Capture)

  ETL (Legacy)
  Source ──> Transform ──> Warehouse
             (serveur ETL = bottleneck)

  ELT (Modern Data Stack)
  Source ──> Warehouse (raw) ──> Transform (dbt)
             Utilise la puissance du cloud DWH

  CDC (Real-time)
  Source DB ──> WAL/Binlog ──> Kafka ──> Consumer ──> DWH
               (Debezium)    (events)  (Flink/Spark)

⚙ dbt CI/CD Pipeline

# .github/workflows/dbt-ci.yml
name: dbt CI/CD
on:
  pull_request:
    paths: ['dbt_project/**']
  push:
    branches: [main]

jobs:
  dbt-ci:
    if: github.event_name == 'pull_request'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install dbt-core dbt-snowflake
      - run: dbt deps
      # Slim CI: ne build que les modeles modifies
      - run: |
          dbt build --target ci \
            --select state:modified+ \
            --defer --state target-prod/ \
            --fail-fast

  dbt-deploy:
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - run: dbt build --target prod
      - run: dbt docs generate

🌐 Airflow Executors & Services Manages

🧊 Apache Iceberg Avance

-- Partition Evolution (unique a Iceberg!)
ALTER TABLE db.orders
ADD PARTITION FIELD months(ordered_at);
-- Ancien partitionnement coexiste avec le nouveau
-- Pas de reecriture des donnees existantes!

-- Hidden Partitioning (transforms)
CREATE TABLE db.events
USING iceberg
PARTITIONED BY (
    days(event_ts),        -- Partition par jour
    bucket(16, user_id)    -- Hash partitioning
);
-- L'utilisateur ecrit WHERE event_ts = '...'
-- Iceberg applique automatiquement le pruning

-- Compaction (small files)
CALL system.rewrite_data_files(
    table => 'db.orders',
    options => map(
        'target-file-size-bytes', '134217728'  -- 128MB
    )
);

-- Expirer les anciens snapshots
CALL system.expire_snapshots(
    table => 'db.orders',
    older_than => TIMESTAMP '2024-06-01'
);

-- Delta UniForm (Databricks)
-- Genere automatiquement les metadonnees Iceberg
ALTER TABLE delta_table
SET TBLPROPERTIES ('delta.universalFormat.enabledFormats' = 'iceberg');