Data Lake Platform

Data Lake Platform

Build and operate production data lakes and lakehouses: ingest, transform, store in open formats, and serve analytics reliably.

When to Use

• Design data lake/lakehouse architecture
• Set up ingestion pipelines (batch, incremental, CDC)
• Build SQL transformation layers (SQLMesh, dbt)
• Choose table formats and catalogs (Iceberg, Delta, Hudi)
• Deploy query/serving engines (Trino, ClickHouse, DuckDB)
• Implement streaming pipelines (Kafka, Flink)
• Set up orchestration (Dagster, Airflow, Prefect)
• Add governance, lineage, data quality, and cost controls

Triage Questions

1. Batch, streaming, or hybrid? What is the freshness SLO? 2. Append-only vs upserts/deletes (CDC)? Is time travel required? 3. Primary query pattern: BI dashboards (high concurrency), ad-hoc joins, embedded analytics? 4. PII/compliance: row/column-level access, retention, audit logging? 5. Platform constraints: self-hosted vs cloud, preferred engines, team strengths?

Default Baseline (Good Starting Point)

• Storage: object storage + open table format (usually Iceberg)
• Catalog: REST/Hive/Glue/Nessie/Unity (match your platform)
• Transforms: SQLMesh or dbt (pick one and standardize)
• Lake query: Trino (or Spark for heavy compute/ML workloads)
• Serving (optional): ClickHouse/StarRocks/Doris for low-latency BI
• Governance: DataHub/OpenMetadata + OpenLineage
• Orchestration: Dagster/Airflow/Prefect

Workflow

1. Pick table format + catalog: references/storage-formats.md (use assets/cross-platform/template-schema-evolution.md and assets/cross-platform/template-partitioning-strategy.md) 2. Design ingestion (batch/incremental/CDC): references/ingestion-patterns.md (use assets/cross-platform/template-ingestion-governance-checklist.md and assets/cross-platform/template-incremental-loading.md) 3. Design transformations (bronze/silver/gold or data products): references/transformation-patterns.md (use assets/cross-platform/template-data-pipeline.md) 4. Choose lake query vs serving engines: references/query-engine-patterns.md 5. Add governance, lineage, and quality gates: references/governance-catalog.md (use assets/cross-platform/template-data-quality-governance.md and assets/cross-platform/template-data-quality.md) 6. Plan operations + cost controls: references/operational-playbook.md and references/cost-optimization.md (use assets/cross-platform/template-data-quality-backfill-runbook.md and assets/cross-platform/template-cost-optimization.md)

Architecture Patterns

• Medallion (bronze/silver/gold): references/architecture-patterns.md
• Data mesh (domain-owned data products): references/architecture-patterns.md
• Streaming-first (Kappa): references/streaming-patterns.md

Quick Start

dlt + ClickHouse

pip install "dlt[clickhouse]"
dlt init rest_api clickhouse
python pipeline.py

SQLMesh + DuckDB

pip install sqlmesh
sqlmesh init duckdb
sqlmesh plan && sqlmesh run

Reliability and Safety

Do

• Define data contracts and owners up front
• Add quality gates (freshness, volume, schema, distribution) per tier
• Make every pipeline idempotent and re-runnable (backfills are normal)
• Treat access control and audit logging as first-class requirements

Avoid

• Skipping validation to "move fast"
• Storing PII without access controls
• Pipelines that can't be re-run safely
• Manual schema changes without version control

Resources

Templates

Related Skills

Fact-Checking

• Use web search/web fetch to verify current external facts, versions, pricing, deadlines, regulations, or platform behavior before final answers.
• Prefer primary sources; report source links and dates for volatile information.
• If web access is unavailable, state the limitation and mark guidance as unverified.

Cost Optimization Template

Overview

Strategies for reducing storage, compute, and operational costs in data lake environments.

Cost Categories

---

Storage Optimization

Compression Strategies

-- ClickHouse: Column-level compression
CREATE TABLE events (
    event_id UUID CODEC(ZSTD(3)),
    user_id UInt64 CODEC(Delta, ZSTD(1)),
    event_type LowCardinality(String),  -- Dictionary encoding
    properties String CODEC(ZSTD(5)),   -- High compression for JSON
    created_at DateTime CODEC(DoubleDelta, ZSTD(1))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (user_id, created_at);

Codec Selection Guide

Iceberg File Optimization

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

-- Remove orphan files
CALL catalog.system.remove_orphan_files(
    table => 'db.events',
    older_than => TIMESTAMP '2024-01-01 00:00:00'
);

-- Expire old snapshots
CALL catalog.system.expire_snapshots(
    table => 'db.events',
    older_than => TIMESTAMP '2024-06-01 00:00:00',
    retain_last => 10
);

Data Tiering

# Iceberg storage tiering
storage_tiers:
  hot:
    path: s3://data-lake-hot/
    retention: 30d
    storage_class: STANDARD

  warm:
    path: s3://data-lake-warm/
    retention: 365d
    storage_class: STANDARD_IA

  cold:
    path: s3://data-lake-archive/
    retention: 7y
    storage_class: GLACIER

# Automated tiering script
from datetime import datetime, timedelta

def tier_partitions(table, catalog):
    """Move old partitions to cheaper storage."""
    hot_cutoff = datetime.now() - timedelta(days=30)
    warm_cutoff = datetime.now() - timedelta(days=365)

    # Move to warm tier
    catalog.execute(f"""
        ALTER TABLE {table}
        SET PARTITION FIELD location = 's3://data-lake-warm/'
        WHERE created_at < '{hot_cutoff.isoformat()}'
          AND created_at >= '{warm_cutoff.isoformat()}'
    """)

ClickHouse Storage Policies

<!-- config.xml -->
<storage_configuration>
    <disks>
        <hot>
            <type>local</type>
            <path>/data/hot/</path>
        </hot>
        <cold>
            <type>s3</type>
            <endpoint>https://s3.amazonaws.com/bucket/cold/</endpoint>
        </cold>
    </disks>
    <policies>
        <tiered>
            <volumes>
                <hot>
                    <disk>hot</disk>
                </hot>
                <cold>
                    <disk>cold</disk>
                </cold>
            </volumes>
            <move_factor>0.1</move_factor>
        </tiered>
    </policies>
</storage_configuration>

-- Use tiered storage
CREATE TABLE events (...)
ENGINE = MergeTree()
SETTINGS storage_policy = 'tiered';

-- TTL-based tiering
CREATE TABLE events (
    created_at DateTime,
    data String
)
ENGINE = MergeTree()
ORDER BY created_at
TTL created_at + INTERVAL 30 DAY TO VOLUME 'cold',
    created_at + INTERVAL 365 DAY DELETE
SETTINGS storage_policy = 'tiered';

---

Compute Optimization

Query Optimization

-- ClickHouse: Use PREWHERE for early filtering
SELECT user_id, count()
FROM events
PREWHERE created_at >= '2024-01-01'  -- Filter before reading columns
WHERE event_type = 'purchase'
GROUP BY user_id;

-- Use sampling for exploration
SELECT event_type, count()
FROM events
SAMPLE 0.01  -- 1% sample
GROUP BY event_type;

-- Limit columns read
SELECT user_id, created_at  -- Don't SELECT *
FROM events
WHERE user_id = 123;

Materialized Views for Common Queries

-- Pre-aggregate expensive queries
CREATE MATERIALIZED VIEW hourly_events_mv
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, event_type)
AS SELECT
    toStartOfHour(created_at) AS hour,
    event_type,
    count() AS event_count,
    uniq(user_id) AS unique_users
FROM events
GROUP BY hour, event_type;

-- Query the view instead of raw table
SELECT * FROM hourly_events_mv
WHERE hour >= '2024-01-01'
  AND hour < '2024-02-01';

Projection Optimization (ClickHouse)

-- Create projection for different access patterns
ALTER TABLE events ADD PROJECTION events_by_type (
    SELECT * ORDER BY (event_type, created_at)
);

-- Materialize projection
ALTER TABLE events MATERIALIZE PROJECTION events_by_type;

-- ClickHouse auto-selects best projection
SELECT * FROM events WHERE event_type = 'purchase';

Resource Quotas

-- ClickHouse: Limit resource usage
CREATE SETTINGS PROFILE analyst_profile
SETTINGS
    max_memory_usage = 10000000000,          -- 10 GB
    max_execution_time = 300,                -- 5 minutes
    max_rows_to_read = 1000000000,           -- 1 billion rows
    max_bytes_to_read = 100000000000;        -- 100 GB

-- Apply to user
ALTER USER analyst SETTINGS PROFILE 'analyst_profile';

---

Ingestion Optimization

Batch Size Tuning

# dlt: Optimal batch sizes
pipeline = dlt.pipeline(
    pipeline_name="events",
    destination="clickhouse"
)

# For ClickHouse: larger batches = better compression
pipeline.run(
    source(),
    loader_file_format="parquet",  # Better than JSON
    write_disposition="append"
)

Buffer Tables (ClickHouse)

-- Buffer for high-frequency inserts
CREATE TABLE events_buffer AS events
ENGINE = Buffer(
    default, events,       -- Target table
    16,                    -- Num buffers
    10, 100,               -- Min/max seconds
    10000, 1000000,        -- Min/max rows
    10000000, 100000000    -- Min/max bytes
);

-- Insert to buffer (auto-flushes to events)
INSERT INTO events_buffer VALUES (...);

Async Inserts (ClickHouse 21.8+)

-- Enable async inserts for small batches
SET async_insert = 1;
SET wait_for_async_insert = 0;
SET async_insert_max_data_size = 10000000;  -- 10 MB
SET async_insert_busy_timeout_ms = 200;

-- ClickHouse batches small inserts automatically
INSERT INTO events VALUES (...);  -- Batched with other inserts

---

Cost Monitoring

Storage Metrics

-- ClickHouse: Storage usage by table
SELECT
    database,
    table,
    formatReadableSize(sum(bytes_on_disk)) AS size,
    formatReadableSize(sum(data_compressed_bytes)) AS compressed,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 2) AS ratio
FROM system.parts
WHERE active
GROUP BY database, table
ORDER BY sum(bytes_on_disk) DESC;

-- Storage by partition
SELECT
    partition,
    formatReadableSize(sum(bytes_on_disk)) AS size,
    sum(rows) AS rows
FROM system.parts
WHERE table = 'events' AND active
GROUP BY partition
ORDER BY partition DESC;

Query Cost Tracking

-- ClickHouse: Query resource usage
SELECT
    user,
    query_kind,
    count() AS queries,
    formatReadableSize(sum(read_bytes)) AS total_read,
    formatReadableSize(sum(memory_usage)) AS total_memory,
    round(sum(query_duration_ms) / 1000, 2) AS total_seconds
FROM system.query_log
WHERE event_date >= today() - 7
  AND type = 'QueryFinish'
GROUP BY user, query_kind
ORDER BY sum(read_bytes) DESC;

-- Expensive queries
SELECT
    query,
    formatReadableSize(read_bytes) AS read_size,
    formatReadableSize(memory_usage) AS memory,
    query_duration_ms / 1000 AS seconds
FROM system.query_log
WHERE event_date >= today()
  AND type = 'QueryFinish'
ORDER BY read_bytes DESC
LIMIT 20;

Cost Dashboard Queries

-- Daily cost estimate
SELECT
    event_date,
    formatReadableSize(sum(read_bytes)) AS bytes_read,
    round(sum(read_bytes) / 1e12 * 0.005, 2) AS estimated_cost_usd,  -- $5/TB
    count() AS query_count
FROM system.query_log
WHERE event_date >= today() - 30
  AND type = 'QueryFinish'
GROUP BY event_date
ORDER BY event_date DESC;

---

Automation Scripts

Automated Compaction

# scheduled_maintenance.py
from datetime import datetime, timedelta

def run_maintenance():
    """Daily maintenance tasks."""

    # 1. Compact small files
    execute("OPTIMIZE TABLE events FINAL")

    # 2. Expire old snapshots (Iceberg)
    cutoff = (datetime.now() - timedelta(days=7)).isoformat()
    execute(f"""
        CALL catalog.system.expire_snapshots(
            table => 'db.events',
            older_than => TIMESTAMP '{cutoff}'
        )
    """)

    # 3. Remove orphan files
    execute("""
        CALL catalog.system.remove_orphan_files(table => 'db.events')
    """)

    # 4. Analyze tables
    execute("ANALYZE TABLE events")

if __name__ == "__main__":
    run_maintenance()

Cost Alerting

# cost_monitor.py
def check_costs():
    """Alert if costs exceed threshold."""

    daily_cost = query("""
        SELECT sum(read_bytes) / 1e12 * 5 AS cost_usd
        FROM system.query_log
        WHERE event_date = today() AND type = 'QueryFinish'
    """)[0]["cost_usd"]

    if daily_cost > 100:  # $100/day threshold
        send_alert(f"Daily cost ${daily_cost:.2f} exceeds threshold")

    # Check storage growth
    storage_gb = query("""
        SELECT sum(bytes_on_disk) / 1e9 AS gb
        FROM system.parts WHERE active
    """)[0]["gb"]

    weekly_growth = query("""
        SELECT sum(bytes_on_disk) / 1e9 AS gb
        FROM system.parts
        WHERE modification_time >= today() - 7
    """)[0]["gb"]

    if weekly_growth > storage_gb * 0.1:  # >10% weekly growth
        send_alert(f"Storage growth {weekly_growth:.0f} GB this week")

---

Best Practices Checklist

Storage

• [ ] Enable compression (ZSTD level 3+ for text)
• [ ] Use LowCardinality for low-cardinality strings
• [ ] Implement data tiering (hot/warm/cold)
• [ ] Set TTL for data retention
• [ ] Compact small files regularly
• [ ] Expire old snapshots (Iceberg/Delta)

Compute

• [ ] Use materialized views for common aggregations
• [ ] Create projections for different access patterns
• [ ] Implement query resource limits
• [ ] Use PREWHERE for early filtering
• [ ] Avoid SELECT * in production queries
• [ ] Use sampling for exploration

Operations

• [ ] Automate maintenance tasks
• [ ] Monitor costs daily
• [ ] Set up cost alerts
• [ ] Review expensive queries weekly
• [ ] Track storage growth trends

Data Pipeline Template

Overview

End-to-end data pipeline from source to analytics-ready tables.

Pipeline Architecture

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   Sources   │───→│   Ingest    │───→│  Transform  │───→│    Serve    │
│ API/DB/File │    │  (dlt)      │    │  (SQLMesh)  │    │ (ClickHouse)│
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘
                          │                  │                  │
                          ▼                  ▼                  ▼
                   ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
                   │   Bronze    │    │   Silver    │    │    Gold     │
                   │ (Raw/Lake)  │    │  (Cleaned)  │    │  (Marts)    │
                   └─────────────┘    └─────────────┘    └─────────────┘

Step 1: Source Configuration

dlt Source Definition

# pipelines/sources/api_source.py
import dlt
from dlt.sources.rest_api import rest_api_source

@dlt.source(name="crm")
def crm_source(api_key: str = dlt.secrets.value):
    """CRM API data source."""

    config = {
        "client": {
            "base_url": "https://api.crm.example.com/v1/",
            "auth": {"type": "api_key", "api_key": api_key}
        },
        "resources": [
            {
                "name": "customers",
                "endpoint": {"path": "customers", "params": {"per_page": 100}},
                "primary_key": "id"
            },
            {
                "name": "orders",
                "endpoint": {"path": "orders"},
                "primary_key": "order_id"
            },
            {
                "name": "products",
                "endpoint": {"path": "products"},
                "primary_key": "sku"
            }
        ]
    }

    yield from rest_api_source(config)

Database Source

# pipelines/sources/db_source.py
import dlt
from dlt.sources.sql_database import sql_database

@dlt.source(name="postgres_source")
def postgres_source():
    """PostgreSQL CDC source."""
    return sql_database(
        credentials="postgresql://user:pass@host:5432/db",
        schema="public",
        table_names=["users", "transactions", "events"],
        incremental=dlt.sources.incremental("updated_at"),
        chunk_size=10000
    )

Step 2: Ingestion Pipeline

dlt Pipeline Configuration

# pipelines/bronze_pipeline.py
import dlt
from sources.api_source import crm_source
from sources.db_source import postgres_source

def run_bronze_pipeline():
    """Ingest raw data to bronze layer."""

    pipeline = dlt.pipeline(
        pipeline_name="bronze_ingestion",
        destination="filesystem",  # or "clickhouse", "duckdb"
        dataset_name="bronze",
        progress="log"
    )

    # Load API data
    api_info = pipeline.run(
        crm_source(),
        write_disposition="merge",
        primary_key="id"
    )
    print(f"API load: {api_info}")

    # Load database data
    db_info = pipeline.run(
        postgres_source(),
        write_disposition="merge"
    )
    print(f"DB load: {db_info}")

    return pipeline

if __name__ == "__main__":
    run_bronze_pipeline()

Filesystem Destination (Iceberg)

# pipelines/config.py
import dlt

# Configure Iceberg destination
destination_config = {
    "filesystem": {
        "bucket_url": "s3://data-lake/bronze/",
        "credentials": {
            "aws_access_key_id": dlt.secrets["aws_access_key_id"],
            "aws_secret_access_key": dlt.secrets["aws_secret_access_key"]
        }
    },
    "table_format": "iceberg",
    "iceberg": {
        "catalog_type": "rest",
        "catalog_uri": "http://iceberg-rest:8181"
    }
}

Step 3: Transformation Layer

SQLMesh Project Structure

transform/
├── sqlmesh.yaml
├── models/
│   ├── staging/
│   │   ├── stg_customers.sql
│   │   ├── stg_orders.sql
│   │   └── stg_products.sql
│   ├── intermediate/
│   │   └── int_order_items.sql
│   └── marts/
│       ├── fct_daily_sales.sql
│       └── dim_customers.sql
├── audits/
│   └── data_quality.sql
└── tests/
    └── test_sales.yaml

SQLMesh Configuration

# transform/sqlmesh.yaml
gateways:
  local:
    connection:
      type: clickhouse
      host: localhost
      port: 8123
      database: analytics

model_defaults:
  dialect: clickhouse
  start: 2024-01-01

default_gateway: local

Staging Model (Bronze → Silver)

-- models/staging/stg_customers.sql
MODEL (
  name silver.stg_customers,
  kind INCREMENTAL_BY_UNIQUE_KEY (
    unique_key [customer_id]
  ),
  cron '@daily',
  grain customer_id,
  audits (
    not_null(columns=[customer_id, email]),
    unique(columns=[customer_id])
  )
);

SELECT
    id AS customer_id,
    lower(trim(email)) AS email,
    coalesce(name, 'Unknown') AS name,
    created_at,
    updated_at,
    -- Data quality flags
    CASE
        WHEN email LIKE '%@%' THEN true
        ELSE false
    END AS is_valid_email
FROM bronze.customers
WHERE id IS NOT NULL
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY id
    ORDER BY updated_at DESC
) = 1;

Mart Model (Silver → Gold)

-- models/marts/fct_daily_sales.sql
MODEL (
  name gold.fct_daily_sales,
  kind FULL,
  cron '@daily',
  grain [date, product_id]
);

SELECT
    toDate(o.created_at) AS date,
    oi.product_id,
    p.category,
    count(DISTINCT o.order_id) AS order_count,
    sum(oi.quantity) AS units_sold,
    sum(oi.quantity * oi.unit_price) AS revenue,
    avg(oi.unit_price) AS avg_price
FROM silver.stg_orders o
JOIN silver.int_order_items oi ON o.order_id = oi.order_id
JOIN silver.stg_products p ON oi.product_id = p.product_id
WHERE o.status = 'completed'
GROUP BY date, oi.product_id, p.category;

Step 4: Serving Layer

ClickHouse Optimized Tables

-- Create optimized serving table
CREATE TABLE gold.sales_dashboard
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, category, product_id)
AS SELECT * FROM gold.fct_daily_sales;

-- Materialized view for real-time updates
CREATE MATERIALIZED VIEW gold.mv_hourly_sales
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, category)
AS SELECT
    toStartOfHour(created_at) AS hour,
    category,
    count() AS orders,
    sum(total) AS revenue
FROM silver.stg_orders
GROUP BY hour, category;

Step 5: Orchestration

Dagster Pipeline

# orchestration/pipeline.py
from dagster import asset, Definitions, ScheduleDefinition
import subprocess

@asset(group_name="bronze")
def bronze_crm():
    """Ingest CRM data to bronze."""
    subprocess.run(["python", "pipelines/bronze_pipeline.py"], check=True)
    return "CRM data ingested"

@asset(deps=[bronze_crm], group_name="silver")
def silver_transform():
    """Transform bronze to silver."""
    subprocess.run(["sqlmesh", "run", "--select-model", "silver.*"], check=True)
    return "Silver models updated"

@asset(deps=[silver_transform], group_name="gold")
def gold_marts():
    """Build gold marts."""
    subprocess.run(["sqlmesh", "run", "--select-model", "gold.*"], check=True)
    return "Gold marts updated"

daily_schedule = ScheduleDefinition(
    job=define_asset_job("daily_pipeline", selection="*"),
    cron_schedule="0 6 * * *"  # 6 AM daily
)

defs = Definitions(
    assets=[bronze_crm, silver_transform, gold_marts],
    schedules=[daily_schedule]
)

Monitoring

Pipeline Health Checks

# monitoring/health.py
def check_pipeline_health():
    checks = {
        "bronze_freshness": check_table_freshness("bronze.*", max_hours=24),
        "silver_freshness": check_table_freshness("silver.*", max_hours=25),
        "gold_freshness": check_table_freshness("gold.*", max_hours=26),
        "row_counts": check_row_counts_growing(),
        "quality_audits": check_sqlmesh_audits_passing()
    }
    return all(checks.values()), checks

def check_table_freshness(pattern, max_hours):
    """Check if tables were updated within threshold."""
    query = f"""
    SELECT table, max(_loaded_at) AS last_update
    FROM system.parts
    WHERE table LIKE '{pattern}'
    GROUP BY table
    HAVING dateDiff('hour', last_update, now()) > {max_hours}
    """
    # Return True if no stale tables
    return execute_query(query).empty

Best Practices

1. Idempotency: All pipelines should be re-runnable 2. Incremental: Process only new/changed data when possible 3. Schema evolution: Handle schema changes gracefully 4. Data quality: Validate at each layer boundary 5. Monitoring: Alert on freshness, row counts, quality issues 6. Documentation: Document lineage and business logic

Data Quality & Backfill Runbook Template

Use this runbook when data is late, incorrect, missing, or when you need to reprocess a historical window safely.

---

Core

Runbook Metadata

• Incident/ticket ID:
• Date/time (timezone):
• Owner/on-call:
• Affected datasets/tables:
• Downstream consumers impacted:
• Severity:

Trigger

• Freshness SLA breach
• Quality check failure (schema, nulls, duplicates, ranges)
• Upstream source correction
• Pipeline bug fix requiring reprocessing

Safety Checks (Before Action)

• [ ] Confirm source of truth for the backfill window
• [ ] Confirm idempotency/replay behavior (upsert keys, dedupe keys)
• [ ] Confirm downstream behavior (will consumers auto-refresh?)
• [ ] Confirm compute budget and expected runtime
• [ ] Decide whether to pause downstream jobs during backfill

Backfill Plan

Window and Strategy

• Backfill window: start/end
• Strategy: overwrite partition / merge-upsert / append + reconcile
• Expected output partitions/tables:

Execution Steps

1. Create a backfill branch/tag for pipeline config (auditability). 2. Run the backfill job for the defined window. 3. Capture job outputs (row counts, runtime, error logs). 4. Run validation suite (see below). 5. Re-enable downstream jobs and verify end-to-end freshness.

Validation (Must Pass Before Closing)

Contract Checks

• [ ] Schema matches contract (types, nullability)
• [ ] Primary/dedupe keys unique within window
• [ ] Freshness updated and within SLA

Data Quality Checks

• [ ] Row-count sanity vs baseline (bounds)
• [ ] Null-rate and distribution checks (key columns)
• [ ] Business invariants hold (e.g., totals, monotonicity)

Consumer Checks

• [ ] Dashboards refreshed and consistent
• [ ] Downstream tables rebuilt (if applicable)
• [ ] Sampling spot-check completed

Rollback Plan

• Rollback trigger:
• Rollback mechanism (restore snapshot / revert partitions / rerun last-known-good):
• Verification after rollback:

Communication

• Internal channel:
• Stakeholder update cadence:
• Customer-facing update needed: yes/no

Post-Incident Follow-Up

• Root cause summary:
• Preventive actions (tests, monitors, contracts, process):
• Runbook updates required:

---

Optional: AI/Automation

• Summarize validation results and highlight anomalies (human-verified)
• Suggest likely root causes from recent schema changes and upstream events (human-validated)
• Auto-generate stakeholder updates from structured runbook fields (human-approved)

Bounded Claims

• Automation cannot determine correctness without explicit rules and human review.
• Never auto-apply destructive backfills without an approval workflow.

Data Quality & Governance Checklist

Production-ready checklist for data lake quality, governance, and reliability.

---

Data Quality Contracts

Contract Definition Checklist

• [ ] Schema defined (columns, types, nullable, descriptions)
• [ ] Freshness SLA defined (max staleness in hours/minutes)
• [ ] Volume bounds defined (min/max row counts per load)
• [ ] Uniqueness constraints documented (primary keys, business keys)
• [ ] Referential integrity rules documented
• [ ] Allowed value ranges specified (for numeric/date columns)
• [ ] Contract version tracked in metadata

Quality Rules by Tier

Validation Implementation

# Great Expectations example
expectations = [
    # Completeness
    ExpectColumnValuesToNotBeNull("user_id"),
    ExpectColumnValuesToNotBeNull("event_timestamp"),

    # Accuracy
    ExpectColumnValuesToBeBetween("price", min_value=0, max_value=1000000),
    ExpectColumnValuesToMatchRegex("email", r'^[\w\.-]+@[\w\.-]+\.\w+$'),

    # Uniqueness
    ExpectCompoundColumnsToBeUnique(["user_id", "event_timestamp"]),

    # Freshness (check max timestamp)
    ExpectColumnMaxToBeBetween(
        "event_timestamp",
        min_value=datetime.now() - timedelta(hours=1)
    ),

    # Volume
    ExpectTableRowCountToBeBetween(min_value=1000, max_value=10000000),
]

---

Governance & Access Control

IAM & Permissions Checklist

• [ ] Role-based access control (RBAC) implemented
• [ ] Data classification applied (PII, sensitive, public)
• [ ] Row-level security configured for multi-tenant data
• [ ] Column-level masking for sensitive fields
• [ ] Service accounts have least-privilege access
• [ ] Access audit logging enabled
• [ ] Access reviews scheduled quarterly

Permission Matrix Template

Data Classification

---

Security Checklist

Encryption

• [ ] Encryption at rest enabled (S3 SSE, GCS CMEK, Azure Blob)
• [ ] Encryption in transit (TLS 1.3 for all connections)
• [ ] Key rotation policy defined (90 days recommended)
• [ ] Key management service configured (AWS KMS, GCP KMS, HashiCorp Vault)

Network Security

• [ ] VPC/private network for data infrastructure
• [ ] No public endpoints for data stores
• [ ] Firewall rules restrict ingress/egress
• [ ] Private Link / VPC endpoints for cloud services
• [ ] Bastion host for administrative access

Audit & Compliance

• [ ] All data access logged
• [ ] Query audit logs retained (1+ year)
• [ ] Schema change audit trail
• [ ] Data lineage tracked (DataHub, OpenMetadata)
• [ ] Compliance reports automated (SOC2, GDPR, HIPAA as applicable)

---

Reliability Patterns

Backfill & Reprocessing Checklist

• [ ] Idempotent pipelines (re-run safe)
• [ ] Partition-based backfill supported
• [ ] Historical data retention policy defined
• [ ] Backfill runbook documented
• [ ] Backfill testing in staging environment
• [ ] Alerting for long-running backfills

Backfill Procedure Template

## Backfill Procedure: [Table Name]

### Pre-backfill
1. [ ] Identify affected partitions/date range
2. [ ] Estimate resource requirements (time, compute)
3. [ ] Notify downstream consumers
4. [ ] Take snapshot of current state (if destructive)

### Execution
1. [ ] Disable downstream dependencies (or pause)
2. [ ] Run backfill: `sqlmesh run --start-date 2024-01-01 --end-date 2024-03-01`
3. [ ] Monitor progress and resource usage
4. [ ] Validate row counts match expectations

### Post-backfill
1. [ ] Run data quality checks
2. [ ] Compare metrics before/after (spot check)
3. [ ] Re-enable downstream dependencies
4. [ ] Update documentation with backfill date
5. [ ] Notify stakeholders of completion

Idempotency Patterns

---

Cost Control Checklist

Storage Optimization

• [ ] Partitioning strategy defined (by date, region, etc.)
• [ ] Compaction scheduled (Iceberg: daily, target 512MB files)
• [ ] Snapshot expiration policy (keep 7 days)
• [ ] Z-ordering/clustering on common filter columns
• [ ] Compression enabled (Zstd recommended)
• [ ] Cold storage tiering for old data

Compute Optimization

• [ ] Query result caching enabled
• [ ] Materialized views for expensive aggregations
• [ ] Resource quotas per user/team
• [ ] Auto-scaling configured with limits
• [ ] Idle cluster shutdown policy
• [ ] Cost attribution tags on all resources

Cost Monitoring

• [ ] Daily cost reports configured
• [ ] Budget alerts at 50%, 80%, 100%
• [ ] Cost anomaly detection enabled
• [ ] Monthly cost review meeting scheduled
• [ ] Chargeback/showback model defined

---

Do / Avoid

GOOD: Do

• Define data contracts before building pipelines
• Implement quality gates at each tier (Bronze → Silver → Gold)
• Use idempotent operations for all transformations
• Enable audit logging from day one
• Automate data quality checks in CI/CD
• Document SLAs for every critical table
• Plan for backfills in pipeline design

BAD: Avoid

• Skipping data quality validation to "move fast"
• Storing PII without classification and access controls
• Creating pipelines that can't be re-run safely
• Using shared service accounts without audit trails
• Ignoring cost controls until the bill arrives
• Manual schema changes without version control
• Single point of failure in critical pipelines

---

Anti-Patterns

---

Optional: AI/Automation

Note: These are enhancements, not requirements. Implement only after core governance is solid.

Automated Quality Monitoring

• Anomaly detection on data volumes and distributions
• Auto-alerting on schema drift
• ML-based freshness prediction

AI-Assisted Governance

• Auto-classification of PII columns using NLP
• Metadata enrichment from column statistics
• Natural language data catalog search

Bounded Claims

• AI quality detection should supplement, not replace, explicit rules
• Human review required for PII classification decisions
• Auto-generated metadata must be validated before production use

---

Related Templates

• template-medallion-architecture.md — Bronze/Silver/Gold patterns
• template-data-quality.md — Great Expectations integration
• template-cost-optimization.md — Storage and compute cost control

---

Last Updated: December 2025

Data Quality Template

Overview

Data quality framework for validation at each layer of the data lake.

Quality Dimensions

---

SQLMesh Audits

Built-in Audits

-- models/staging/stg_orders.sql
MODEL (
  name silver.stg_orders,
  kind INCREMENTAL_BY_TIME_RANGE (time_column created_at),
  grain order_id,
  audits (
    -- Completeness
    not_null(columns=[order_id, customer_id, total_amount, created_at]),

    -- Uniqueness
    unique(columns=[order_id]),

    -- Validity
    accepted_values(column=status, values=['pending', 'confirmed', 'shipped', 'delivered', 'cancelled']),

    -- Custom audit reference
    assert_positive_amounts
  )
);

SELECT
    order_id,
    customer_id,
    total_amount,
    status,
    created_at
FROM bronze.raw_orders
WHERE created_at BETWEEN @start_dt AND @end_dt;

Custom Audits

-- audits/assert_positive_amounts.sql
AUDIT (
  name assert_positive_amounts,
  dialect clickhouse
);

-- Fails if any rows returned
SELECT *
FROM @this_model
WHERE total_amount < 0
   OR quantity < 0;

-- audits/assert_referential_integrity.sql
AUDIT (
  name assert_orders_have_customers,
  dialect clickhouse
);

SELECT o.*
FROM @this_model o
LEFT JOIN silver.stg_customers c ON o.customer_id = c.customer_id
WHERE c.customer_id IS NULL;

-- audits/assert_no_future_dates.sql
AUDIT (
  name assert_no_future_dates,
  dialect clickhouse
);

SELECT *
FROM @this_model
WHERE created_at > now() + INTERVAL 1 HOUR;

---

Great Expectations

Installation

pip install great-expectations
great_expectations init

Expectation Suite

# expectations/bronze_events_suite.py
import great_expectations as gx

context = gx.get_context()

# Create expectation suite
suite = context.add_expectation_suite("bronze_events")

# Add expectations
validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="bronze_events"
)

# Completeness
validator.expect_column_values_to_not_be_null("event_id")
validator.expect_column_values_to_not_be_null("user_id")
validator.expect_column_values_to_not_be_null("created_at")

# Uniqueness
validator.expect_column_values_to_be_unique("event_id")

# Validity
validator.expect_column_values_to_match_regex(
    "email",
    r"^[\w.-]+@[\w.-]+\.\w+$"
)
validator.expect_column_values_to_be_between(
    "amount",
    min_value=0,
    max_value=1000000
)
validator.expect_column_values_to_be_in_set(
    "status",
    ["pending", "active", "completed", "cancelled"]
)

# Consistency
validator.expect_column_pair_values_to_be_equal(
    "calculated_total",
    "sum_of_line_items"
)

# Save suite
validator.save_expectation_suite()

Run Validation

# validation/run_checks.py
import great_expectations as gx

context = gx.get_context()

# Run checkpoint
result = context.run_checkpoint(
    checkpoint_name="bronze_validation",
    batch_request={
        "datasource_name": "clickhouse",
        "data_asset_name": "bronze.raw_events"
    }
)

if not result.success:
    # Alert on failures
    failed_expectations = [
        r for r in result.run_results.values()
        if not r.success
    ]
    send_alert(failed_expectations)

---

Soda Core

Installation

pip install soda-core-duckdb  # or soda-core-clickhouse

Configuration

# soda/configuration.yml
data_source clickhouse:
  type: clickhouse
  host: localhost
  port: 8123
  database: analytics
  username: ${CLICKHOUSE_USER}
  password: ${CLICKHOUSE_PASSWORD}

Check Definition

# soda/checks/silver_orders.yml
checks for silver.stg_orders:
  # Freshness
  - freshness(created_at) < 24h

  # Row count
  - row_count > 0
  - row_count_change < 50%

  # Completeness
  - missing_count(order_id) = 0
  - missing_count(customer_id) = 0
  - missing_count(total_amount) = 0

  # Uniqueness
  - duplicate_count(order_id) = 0

  # Validity
  - invalid_count(email) = 0:
      valid regex: '^[\w.-]+@[\w.-]+\.\w+$'

  - invalid_count(status) = 0:
      valid values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled']

  - min(total_amount) >= 0
  - max(total_amount) < 1000000

  # Consistency
  - values in (customer_id) must exist in silver.stg_customers (customer_id)

  # Schema
  - schema:
      fail:
        when required column missing: [order_id, customer_id, total_amount, status, created_at]

Run Checks

soda scan -d clickhouse -c soda/configuration.yml soda/checks/silver_orders.yml

---

ClickHouse Quality Checks

Inline Quality Metrics

-- Create quality metrics table
CREATE TABLE meta.data_quality_results (
    check_time DateTime DEFAULT now(),
    table_name String,
    check_name String,
    check_type String,
    passed UInt8,
    failed_count UInt64,
    details String
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(check_time)
ORDER BY (check_time, table_name, check_name);

-- Run quality checks
INSERT INTO meta.data_quality_results
SELECT
    now() AS check_time,
    'silver.stg_orders' AS table_name,
    'null_order_id' AS check_name,
    'completeness' AS check_type,
    countIf(order_id IS NULL) = 0 AS passed,
    countIf(order_id IS NULL) AS failed_count,
    '' AS details
FROM silver.stg_orders
WHERE created_at >= today() - 1;

Quality Dashboard Query

-- Quality summary by table
SELECT
    table_name,
    check_type,
    countIf(passed = 1) AS checks_passed,
    countIf(passed = 0) AS checks_failed,
    round(countIf(passed = 1) * 100.0 / count(), 2) AS pass_rate
FROM meta.data_quality_results
WHERE check_time >= today() - 7
GROUP BY table_name, check_type
ORDER BY table_name, check_type;

-- Recent failures
SELECT
    check_time,
    table_name,
    check_name,
    failed_count,
    details
FROM meta.data_quality_results
WHERE passed = 0
  AND check_time >= today() - 1
ORDER BY check_time DESC;

---

Data Contracts

Contract Definition

# contracts/orders_contract.yaml
contract:
  name: orders_v1
  version: "1.0.0"
  owner: data-team@company.com

schema:
  - name: order_id
    type: String
    required: true
    unique: true

  - name: customer_id
    type: UInt64
    required: true
    foreign_key: customers.customer_id

  - name: total_amount
    type: Decimal(18,2)
    required: true
    constraints:
      - min: 0
      - max: 1000000

  - name: status
    type: String
    required: true
    allowed_values:
      - pending
      - confirmed
      - shipped
      - delivered
      - cancelled

  - name: created_at
    type: DateTime
    required: true
    constraints:
      - max: now() + 1h

sla:
  freshness: 24h
  completeness: 99.9%
  availability: 99.5%

alerts:
  - type: slack
    channel: "#data-quality"
    on: [schema_change, sla_breach, quality_failure]

Contract Validation

# contracts/validate.py
import yaml
from dataclasses import dataclass

@dataclass
class ContractViolation:
    contract: str
    field: str
    violation_type: str
    details: str

def validate_contract(table_name: str, contract_path: str) -> list[ContractViolation]:
    with open(contract_path) as f:
        contract = yaml.safe_load(f)

    violations = []

    for field in contract["schema"]:
        # Check required fields
        if field["required"]:
            null_count = query(f"SELECT count() FROM {table_name} WHERE {field['name']} IS NULL")
            if null_count > 0:
                violations.append(ContractViolation(
                    contract=contract["contract"]["name"],
                    field=field["name"],
                    violation_type="completeness",
                    details=f"{null_count} null values"
                ))

        # Check constraints
        if "constraints" in field:
            for constraint in field["constraints"]:
                # Validate each constraint type
                pass

    return violations

---

Alerting Integration

Slack Alerts

# monitoring/alerts.py
import requests

def send_quality_alert(check_results: list, webhook_url: str):
    failures = [r for r in check_results if not r["passed"]]

    if not failures:
        return

    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": "🚨 Data Quality Alert"}
        },
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*{len(failures)} checks failed*"
            }
        }
    ]

    for failure in failures[:5]:  # Limit to 5
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"• *{failure['table']}*: {failure['check']} - {failure['count']} failures"
            }
        })

    requests.post(webhook_url, json={"blocks": blocks})

---

Best Practices

1. Validate at boundaries: Check data when entering each layer 2. Fail fast: Stop pipelines on critical quality issues 3. Track trends: Monitor quality metrics over time 4. Document expectations: Use data contracts for critical tables 5. Alert appropriately: Critical issues → PagerDuty; warnings → Slack 6. Root cause analysis: Track why quality issues occur

Incremental Loading Template

Overview

Patterns for efficient incremental data loading across the data lake stack.

Incremental Strategies

---

dlt Incremental Loading

Time-Based Incremental

import dlt
from dlt.sources.rest_api import rest_api_source

@dlt.source
def api_source():
    @dlt.resource(
        name="orders",
        write_disposition="merge",
        primary_key="order_id"
    )
    def orders(
        updated_at=dlt.sources.incremental(
            "updated_at",
            initial_value="2024-01-01T00:00:00Z"
        )
    ):
        """Incrementally load orders by updated_at."""
        response = requests.get(
            "https://api.example.com/orders",
            params={
                "updated_after": updated_at.last_value,
                "per_page": 100
            }
        )
        for order in response.json()["orders"]:
            yield order

    return orders

# Pipeline automatically tracks state
pipeline = dlt.pipeline(
    pipeline_name="orders_incremental",
    destination="clickhouse"
)
pipeline.run(api_source())

Cursor-Based Incremental

@dlt.resource(
    write_disposition="append",
    primary_key="event_id"
)
def events(
    event_id=dlt.sources.incremental(
        "event_id",
        initial_value=0,
        primary_key=True  # Use as cursor
    )
):
    """Load events incrementally by ID."""
    while True:
        response = requests.get(
            "https://api.example.com/events",
            params={
                "after_id": event_id.last_value,
                "limit": 1000
            }
        )
        batch = response.json()["events"]
        if not batch:
            break
        yield batch

Database CDC with dlt

from dlt.sources.sql_database import sql_database

source = sql_database(
    credentials="postgresql://user:pass@host:5432/db",
    table_names=["users", "orders"],
    incremental=dlt.sources.incremental(
        cursor_path="updated_at",
        initial_value="2024-01-01"
    ),
    # Backend queries: WHERE updated_at > @last_value
    backend_kwargs={"chunk_size": 50000}
)

---

SQLMesh Incremental Models

INCREMENTAL_BY_TIME_RANGE

-- Best for time-series data
MODEL (
  name silver.stg_events,
  kind INCREMENTAL_BY_TIME_RANGE (
    time_column created_at,
    batch_size 1,          -- Days per batch
    batch_concurrency 4,   -- Parallel batches
    lookback 2             -- Re-process last 2 periods
  ),
  cron '@hourly',
  grain event_id
);

SELECT
    event_id,
    user_id,
    event_type,
    properties,
    created_at
FROM bronze.raw_events
WHERE created_at BETWEEN @start_dt AND @end_dt
  AND event_id IS NOT NULL;

INCREMENTAL_BY_UNIQUE_KEY

-- Best for SCD Type 1 (upsert)
MODEL (
  name silver.stg_users,
  kind INCREMENTAL_BY_UNIQUE_KEY (
    unique_key [user_id],
    when_matched WHEN MATCHED THEN UPDATE SET
      email = source.email,
      name = source.name,
      updated_at = source.updated_at
  ),
  cron '@daily',
  grain user_id
);

SELECT
    user_id,
    email,
    name,
    created_at,
    updated_at
FROM bronze.raw_users
WHERE updated_at >= @execution_date - INTERVAL 1 DAY;

SCD Type 2 (History Tracking)

MODEL (
  name silver.dim_users_history,
  kind SCD_TYPE_2 (
    unique_key [user_id],
    valid_from_name valid_from,
    valid_to_name valid_to,
    invalidate_hard_deletes true
  ),
  cron '@daily',
  grain [user_id, valid_from]
);

SELECT
    user_id,
    email,
    subscription_tier,
    updated_at AS effective_date
FROM bronze.raw_users;

---

ClickHouse Incremental Patterns

ReplacingMergeTree (Deduplication)

-- Automatically keeps latest version
CREATE TABLE silver.users (
    user_id UInt64,
    email String,
    name String,
    updated_at DateTime
)
ENGINE = ReplacingMergeTree(updated_at)
ORDER BY user_id;

-- Insert new/updated records
INSERT INTO silver.users
SELECT * FROM bronze.raw_users
WHERE updated_at > (
    SELECT max(updated_at) FROM silver.users
);

-- Query with deduplication
SELECT * FROM silver.users FINAL;

-- Or use argMax for latest value
SELECT
    user_id,
    argMax(email, updated_at) AS email,
    argMax(name, updated_at) AS name,
    max(updated_at) AS updated_at
FROM silver.users
GROUP BY user_id;

CollapsingMergeTree (State Changes)

-- Track state changes with sign column
CREATE TABLE silver.balances (
    user_id UInt64,
    balance Decimal(18, 2),
    updated_at DateTime,
    sign Int8  -- 1 = insert, -1 = delete
)
ENGINE = CollapsingMergeTree(sign)
ORDER BY (user_id, updated_at);

-- Insert new state (cancel old + add new)
INSERT INTO silver.balances
SELECT user_id, balance, now(), -1 FROM silver.balances WHERE user_id = 123
UNION ALL
SELECT 123, 1500.00, now(), 1;

Materialized View (Real-time Aggregation)

-- Source table
CREATE TABLE bronze.events (
    event_id UUID,
    user_id UInt64,
    event_type String,
    created_at DateTime
)
ENGINE = MergeTree()
ORDER BY (created_at, event_id);

-- Auto-updating aggregate
CREATE MATERIALIZED VIEW silver.hourly_events
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (hour, event_type)
AS SELECT
    toStartOfHour(created_at) AS hour,
    event_type,
    count() AS event_count,
    uniq(user_id) AS unique_users
FROM bronze.events
GROUP BY hour, event_type;

---

Iceberg Incremental Patterns

Merge Into (Upsert)

-- Apache Spark / Trino
MERGE INTO silver.customers t
USING bronze.raw_customers s
ON t.customer_id = s.customer_id
WHEN MATCHED AND s.updated_at > t.updated_at THEN
    UPDATE SET *
WHEN NOT MATCHED THEN
    INSERT *;

Incremental Read

# PyIceberg incremental read
from pyiceberg.catalog import load_catalog

catalog = load_catalog("rest")
table = catalog.load_table("silver.events")

# Read only new snapshots
scan = table.scan(
    snapshot_id=last_processed_snapshot
).to_arrow()

Time Travel for Comparison

-- Compare current vs previous state
SELECT
    'new' AS status,
    current.*
FROM silver.customers current
LEFT JOIN silver.customers VERSION AS OF 'yesterday' AS prev
    ON current.customer_id = prev.customer_id
WHERE prev.customer_id IS NULL

UNION ALL

SELECT
    'changed' AS status,
    current.*
FROM silver.customers current
JOIN silver.customers VERSION AS OF 'yesterday' AS prev
    ON current.customer_id = prev.customer_id
WHERE current.updated_at > prev.updated_at;

---

State Management

dlt State Tracking

# dlt automatically manages state in .dlt/
# State location: .dlt/pipelines/{pipeline_name}/state/

# Manual state access
pipeline = dlt.pipeline(pipeline_name="my_pipeline")
state = pipeline.state

# Check last incremental value
last_value = state["sources"]["api"]["resources"]["orders"]["incremental"]["updated_at"]["last_value"]

# Reset state (force full refresh)
pipeline.drop()

SQLMesh State

# View model state
sqlmesh info

# Force full refresh of model
sqlmesh run --model silver.stg_orders --restate-model

# Backfill specific date range
sqlmesh plan --start 2024-01-01 --end 2024-01-31

Custom State Table

-- Track pipeline state in ClickHouse
CREATE TABLE meta.pipeline_state (
    pipeline_name String,
    resource_name String,
    last_value String,
    last_run DateTime,
    rows_processed UInt64
)
ENGINE = ReplacingMergeTree(last_run)
ORDER BY (pipeline_name, resource_name);

-- Update after each run
INSERT INTO meta.pipeline_state VALUES
    ('orders_pipeline', 'orders', '2024-06-15T10:30:00Z', now(), 15000);

---

Best Practices

DO

1. Use time-based incremental when source has reliable timestamps 2. Add lookback period to catch late-arriving data 3. Track state externally for complex pipelines 4. Validate row counts after incremental loads 5. Monitor data freshness with alerts

DON'T

1. Don't assume timestamps are reliable - validate first 2. Don't skip deduplication - duplicates happen 3. Don't use full refresh for large tables without reason 4. Don't ignore late data - design for it 5. Don't forget to test incremental logic with edge cases

---

Validation Queries

-- Check for gaps in incremental data
SELECT
    toDate(created_at) AS date,
    count() AS records,
    min(created_at) AS min_time,
    max(created_at) AS max_time
FROM silver.stg_events
GROUP BY date
ORDER BY date;

-- Detect duplicate records
SELECT
    event_id,
    count() AS duplicates
FROM silver.stg_events
GROUP BY event_id
HAVING count() > 1;

-- Compare source vs target counts
SELECT
    'bronze' AS layer, count() AS records FROM bronze.raw_events
UNION ALL
SELECT
    'silver' AS layer, count() AS records FROM silver.stg_events;

Data Lake Ingestion & Governance Checklist

Use this checklist when onboarding a new source, dataset, or data product into a lake/lakehouse.

---

Core

1) Dataset Intake

• Dataset name:
• Business owner:
• Technical owner/on-call:
• Source system(s):
• Consumers (dashboards, ML features, downstream services):
• Data classification: public / internal / confidential / restricted (PII/PHI/PCI)
• Freshness target (SLA/SLO):
• Retention requirements (legal/regulatory + business):

2) Ingestion Design (Batch / Streaming / CDC)

• [ ] Ingestion mode chosen: batch / streaming / CDC
• [ ] Contract defined:
• [ ] Schema (types, nullability, semantics)
• [ ] Primary key / natural key / dedupe key
• [ ] Event time vs processing time (if applicable)
• [ ] Allowed late data window (if applicable)
• [ ] Idempotency strategy:
• [ ] Upsert/merge keys defined
• [ ] Re-runs are safe (no double counts)
• [ ] Exactly-once is not assumed; use at-least-once + dedupe
• [ ] Schema evolution policy:
• [ ] Additive changes allowed by default
• [ ] Breaking changes require versioning and consumer notice
• [ ] Backward/forward compatibility rules documented
• [ ] Failure handling:
• [ ] Dead-letter/quarantine path defined
• [ ] Retries with backoff/jitter
• [ ] Partial loads are detectable and alertable

3) Storage and Table Format

• [ ] Table format chosen (open, multi-engine where possible)
• [ ] Partitioning/clustering strategy documented (aligned to common filters)
• [ ] Compaction/maintenance plan (small files, manifests, vacuum) scheduled
• [ ] Naming conventions and dataset layout standardized

4) Governance and Access Control

• [ ] Catalog entry created (owner, description, tags, lineage links)
• [ ] Data classification tags applied (PII columns identified)
• [ ] RBAC policy defined (who can read/write/admin)
• [ ] Row/column-level security policy defined (where required)
• [ ] Audit logging enabled for access and schema changes
• [ ] Encryption in transit and at rest verified

5) Quality Gates and Reliability

• [ ] Data quality checks defined (schema, not-null, uniqueness, ranges, freshness)
• [ ] SLAs/SLOs defined and monitored (freshness, completeness, latency)
• [ ] Backfill strategy documented (time window, compute budget, verification)
• [ ] Reprocessing strategy documented (how to rebuild from source of truth)
• [ ] Runbook exists for top failure modes (late data, schema change, upstream outage)

6) Cost Controls

• [ ] Retention and lifecycle policy enforced (tiering, deletion, archival)
• [ ] Compute guardrails (quotas, scheduling, environment budgets)
• [ ] Query cost controls (partition pruning, pre-aggregations, caching)
• [ ] Regular maintenance jobs scheduled (compaction, stats, clustering)

---

Do / Avoid

Do

• Do define contracts and ownership before building pipelines
• Do plan for backfills and reprocessing from day one
• Do enforce least privilege and audit trails for sensitive datasets
• Do design partitions for the queries you will run (not for aesthetics)

Avoid

• Avoid relying on “schema on read” without validation gates
• Avoid shared service accounts with no traceability
• Avoid unbounded retention and unbounded scans (cost runaway)
• Avoid pipelines that cannot be replayed safely

---

Optional: AI/Automation

• Auto-detect anomalies (volume, distribution, freshness) as a supplement to rules
• Assist with metadata enrichment (tags, owners, column descriptions) with review
• Suggest partition keys from observed query patterns (human-approved)

Bounded Claims

• Automation does not replace explicit contracts, tests, or access controls.
• PII classification suggestions require human validation.

Medallion Architecture Template

Overview

Bronze → Silver → Gold data quality progression for data lakes.

Directory Structure

data-lake/
├── bronze/                    # Raw data (append-only)
│   ├── events/
│   ├── users/
│   └── orders/
├── silver/                    # Cleaned, validated
│   ├── stg_events/
│   ├── stg_users/
│   └── stg_orders/
└── gold/                      # Business-ready
    ├── fct_daily_events/
    ├── dim_users/
    └── mart_sales/

Bronze Layer (Raw)

dlt Ingestion

import dlt

@dlt.source
def api_source():
    @dlt.resource(
        name="events",
        write_disposition="append",
        table_format="iceberg"
    )
    def events():
        for batch in fetch_events():
            yield batch

    return events

pipeline = dlt.pipeline(
    pipeline_name="bronze_ingestion",
    destination="filesystem",  # or clickhouse, duckdb
    dataset_name="bronze"
)

pipeline.run(api_source())

Bronze Table (ClickHouse)

CREATE TABLE bronze.raw_events (
    _dlt_load_id String,
    _dlt_id String,
    _ingested_at DateTime DEFAULT now(),
    raw_data String  -- Original JSON
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(_ingested_at)
ORDER BY (_ingested_at, _dlt_id)
TTL _ingested_at + INTERVAL 2 YEAR;

Silver Layer (Cleaned)

SQLMesh Model

-- models/staging/stg_events.sql
MODEL (
  name silver.stg_events,
  kind INCREMENTAL_BY_TIME_RANGE (
    time_column created_at
  ),
  cron '@hourly',
  grain event_id,
  audits (
    not_null(columns=[event_id, user_id, created_at])
  )
);

SELECT
    JSONExtractString(raw_data, 'event_id') AS event_id,
    JSONExtractUInt(raw_data, 'user_id') AS user_id,
    JSONExtractString(raw_data, 'event_type') AS event_type,
    parseDateTimeBestEffort(
        JSONExtractString(raw_data, 'created_at')
    ) AS created_at,
    raw_data AS properties,
    _ingested_at AS _loaded_at
FROM bronze.raw_events
WHERE _ingested_at BETWEEN @start_dt AND @end_dt
  AND JSONExtractString(raw_data, 'event_id') IS NOT NULL
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY JSONExtractString(raw_data, 'event_id')
    ORDER BY _ingested_at DESC
) = 1;

Gold Layer (Business-Ready)

Fact Table

-- models/marts/fct_daily_events.sql
MODEL (
  name gold.fct_daily_events,
  kind FULL,
  cron '@daily',
  grain [date, event_type]
);

SELECT
    toDate(created_at) AS date,
    event_type,
    count() AS event_count,
    uniq(user_id) AS unique_users,
    min(created_at) AS first_event,
    max(created_at) AS last_event
FROM silver.stg_events
GROUP BY date, event_type;

Dimension Table

-- models/marts/dim_users.sql
MODEL (
  name gold.dim_users,
  kind SCD_TYPE_2 (
    unique_key [user_id],
    valid_from_name valid_from,
    valid_to_name valid_to
  ),
  cron '@daily',
  grain user_id
);

SELECT
    user_id,
    email,
    name,
    created_at,
    CASE
        WHEN total_orders >= 10 THEN 'vip'
        WHEN total_orders >= 3 THEN 'regular'
        ELSE 'new'
    END AS segment
FROM silver.stg_users u
LEFT JOIN (
    SELECT user_id, count() AS total_orders
    FROM silver.stg_orders
    GROUP BY user_id
) o USING (user_id);

Data Quality

Great Expectations Suite

# bronze_quality.py
validator.expect_column_values_to_not_be_null("_dlt_id")
validator.expect_column_to_exist("raw_data")

# silver_quality.py
validator.expect_column_values_to_not_be_null("event_id")
validator.expect_column_values_to_be_unique("event_id")
validator.expect_column_values_to_match_regex(
    "email", r"^[\w.-]+@[\w.-]+\.\w+$"
)

# gold_quality.py
validator.expect_column_values_to_be_between(
    "event_count", min_value=0
)
validator.expect_table_row_count_to_be_between(
    min_value=1
)

Orchestration (Dagster)

from dagster import asset, Definitions

@asset(group_name="bronze")
def bronze_events():
    """Ingest raw events"""
    pipeline = dlt.pipeline(...)
    return pipeline.run(...)

@asset(deps=[bronze_events], group_name="silver")
def silver_events():
    """Clean and validate events"""
    ctx = sqlmesh.Context()
    ctx.run(select_models=["silver.stg_events"])

@asset(deps=[silver_events], group_name="gold")
def gold_daily_events():
    """Aggregate daily metrics"""
    ctx = sqlmesh.Context()
    ctx.run(select_models=["gold.fct_daily_events"])

Best Practices

1. Bronze: Never modify raw data, append-only 2. Silver: Deduplicate, validate, type columns 3. Gold: Aggregate for specific use cases 4. Lineage: Track transformations in catalog 5. Quality: Validate at each layer transition

Migration Checklist Template

Overview

Step-by-step checklist for migrating data platforms to modern lakehouse architecture.

---

Pre-Migration Assessment

Source System Inventory

• [ ] Document all source systems
• Databases (PostgreSQL, MySQL, SQL Server, Oracle)
• APIs (REST, GraphQL, webhooks)
• File systems (SFTP, S3, local)
• SaaS applications (Salesforce, HubSpot, Stripe)

• [ ] Catalog existing data
• Table names, schemas, row counts
• Data volumes (GB/TB per table)
• Update frequencies (real-time, hourly, daily, batch)
• Data quality issues (nulls, duplicates, inconsistencies)

• [ ] Map dependencies
• Downstream consumers (dashboards, reports, APIs)
• ETL pipelines (Airflow DAGs, cron jobs, stored procedures)
• Business-critical queries and SLAs

Current State Assessment

┌─────────────────────────────────────────────────────────────┐
│ CURRENT STATE ASSESSMENT                                    │
├─────────────────────────────────────────────────────────────┤
│ Source Systems: ______ (count)                             │
│ Total Data Volume: ______ TB                               │
│ Daily Ingestion: ______ GB                                 │
│ Active Users: ______                                       │
│ Critical Dashboards: ______                                │
│ SLA Requirements: ______                                   │
│ Current Monthly Cost: $______                              │
└─────────────────────────────────────────────────────────────┘

---

Architecture Design

Target Architecture Selection

• [ ] Choose lakehouse pattern
• [ ] Medallion (Bronze/Silver/Gold) - recommended for most cases
• [ ] Data Mesh (domain-oriented) - for large organizations
• [ ] Lambda/Kappa (streaming) - for real-time requirements

• [ ] Select core technologies

• [ ] Define data zones

┌─────────────┐   ┌─────────────┐   ┌─────────────┐
│   BRONZE    │──→│   SILVER    │──→│    GOLD     │
│ (Raw/Lake)  │   │ (Validated) │   │  (Marts)    │
└─────────────┘   └─────────────┘   └─────────────┘
     │                  │                  │
  Append-only      Deduplicated      Business-ready
  Schema-on-read   Schema-enforced   Aggregated
  Full history     Current state     Domain models

Infrastructure Planning

• [ ] Storage requirements
• Current data size: ______ TB
• 3-year projection: ______ TB
• Hot storage: ______ TB (SSD/NVMe)
• Warm storage: ______ TB (HDD/S3 Standard)
• Cold storage: ______ TB (S3 Glacier)

• [ ] Compute requirements
• Peak concurrent queries: ______
• Average query complexity: ______
• Batch processing windows: ______

• [ ] Network requirements
• Data transfer between regions: ______ GB/day
• Egress costs consideration

---

Migration Phases

Phase 1: Foundation (Weeks 1-2)

• [ ] Set up infrastructure
• [ ] Provision object storage (S3/MinIO/GCS)
• [ ] Deploy query engine (ClickHouse cluster)
• [ ] Configure Iceberg REST catalog
• [ ] Set up orchestration (Dagster/Airflow)

• [ ] Configure security
• [ ] IAM roles and policies
• [ ] Network security groups
• [ ] Encryption at rest and in transit
• [ ] Secrets management (Vault, AWS Secrets Manager)

• [ ] Establish CI/CD
• [ ] Git repository structure
• [ ] SQLMesh/dbt project scaffolding
• [ ] Testing framework (Great Expectations/Soda)
• [ ] Deployment pipelines

Phase 2: Bronze Layer (Weeks 3-4)

• [ ] Implement ingestion pipelines

# Example: dlt pipeline for each source
@dlt.source
def source_system():
    @dlt.resource(
        write_disposition="append",
        table_format="iceberg"
    )
    def table_name():
        yield from fetch_data()
    return table_name

pipeline = dlt.pipeline(
    pipeline_name="bronze_ingestion",
    destination="filesystem",
    dataset_name="bronze"
)

• [ ] Source system checklist
• [ ] Source 1: ______ (status: pending/in-progress/complete)
• [ ] Source 2: ______
• [ ] Source 3: ______
• [ ] Source N: ______

• [ ] Validate bronze data
• [ ] Row counts match source
• [ ] Schema captured correctly
• [ ] Incremental logic working
• [ ] Historical data backfilled

Phase 3: Silver Layer (Weeks 5-6)

• [ ] Build staging models

-- SQLMesh model example
MODEL (
  name silver.stg_customers,
  kind INCREMENTAL_BY_UNIQUE_KEY (unique_key [customer_id]),
  audits (not_null(columns=[customer_id, email]), unique(columns=[customer_id]))
);

SELECT
    id AS customer_id,
    lower(trim(email)) AS email,
    name,
    created_at,
    updated_at
FROM bronze.raw_customers
WHERE id IS NOT NULL
QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1;

• [ ] Silver layer checklist
• [ ] Deduplication logic verified
• [ ] Data type conversions correct
• [ ] Null handling defined
• [ ] Data quality checks passing
• [ ] Schema documentation complete

Phase 4: Gold Layer (Weeks 7-8)

• [ ] Build business models
• [ ] Fact tables (events, transactions, activities)
• [ ] Dimension tables (users, products, locations)
• [ ] Aggregation tables (daily/weekly/monthly metrics)

• [ ] Migrate existing reports
• [ ] Identify all dashboard queries
• [ ] Map to new gold tables
• [ ] Validate results match legacy system
• [ ] Document query migration

Phase 5: Cutover (Weeks 9-10)

• [ ] Parallel run period
• [ ] Run old and new systems simultaneously
• [ ] Compare query results daily
• [ ] Measure performance differences
• [ ] Document discrepancies

• [ ] User acceptance testing
• [ ] Train users on new system
• [ ] Collect feedback
• [ ] Address issues
• [ ] Sign-off from stakeholders

• [ ] Production cutover
• [ ] Schedule cutover window
• [ ] Notify all stakeholders
• [ ] Execute cutover runbook
• [ ] Monitor for issues
• [ ] Rollback plan ready

---

Validation Checklist

Data Accuracy

-- Compare row counts
SELECT 'old_system' AS source, count(*) FROM old_db.customers
UNION ALL
SELECT 'new_system' AS source, count(*) FROM gold.dim_customers;

-- Compare aggregates
SELECT 'old_system', sum(amount), avg(amount) FROM old_db.orders
UNION ALL
SELECT 'new_system', sum(amount), avg(amount) FROM gold.fct_orders;

-- Sample comparison (random 1000 records)
SELECT * FROM old_db.customers
WHERE id IN (SELECT id FROM old_db.customers ORDER BY rand() LIMIT 1000)
EXCEPT
SELECT * FROM gold.dim_customers
WHERE customer_id IN (...);

Performance Benchmarks

• [ ] Query performance comparison

Data Quality Metrics

• [ ] Quality gate thresholds
• Completeness: >99.9% for required fields
• Uniqueness: 0 duplicates on primary keys
• Validity: >99% pass format checks
• Freshness: Within SLA (e.g., <24 hours)

---

Post-Migration

Documentation

• [ ] Technical documentation
• [ ] Architecture diagrams
• [ ] Data flow diagrams
• [ ] Schema documentation
• [ ] Runbooks for common operations

• [ ] User documentation
• [ ] Data dictionary
• [ ] Query examples
• [ ] Dashboard guide
• [ ] FAQ

Monitoring Setup

• [ ] Alerts configured
• [ ] Pipeline failures
• [ ] Data freshness SLA breaches
• [ ] Quality check failures
• [ ] Resource usage thresholds

• [ ] Dashboards created
• [ ] Pipeline health dashboard
• [ ] Data quality dashboard
• [ ] Cost tracking dashboard
• [ ] Performance metrics dashboard

Decommissioning

• [ ] Legacy system retirement
• [ ] Confirm all consumers migrated
• [ ] Export final data snapshot
• [ ] Archive for compliance (if required)
• [ ] Terminate resources
• [ ] Update documentation

---

Risk Mitigation

Common Risks

Rollback Plan

ROLLBACK PROCEDURE
─────────────────
1. STOP new system ingestion pipelines
2. VERIFY old system still receiving data
3. REDIRECT dashboards to old system
4. NOTIFY users of rollback
5. INVESTIGATE root cause
6. PLAN remediation
7. SCHEDULE retry

---

Success Criteria

Business Metrics

• [ ] Query performance: ≥______% improvement
• [ ] Data freshness: ≤______ hours
• [ ] Cost reduction: ≥______% savings
• [ ] User satisfaction: ≥______% positive feedback

Technical Metrics

• [ ] Pipeline success rate: ≥99.5%
• [ ] Data quality score: ≥99%
• [ ] System availability: ≥99.9%
• [ ] P99 query latency: ≤______ seconds

Partitioning Strategy Template

Overview

Data partitioning strategies for optimal query performance and storage efficiency.

Partitioning Types

---

ClickHouse Partitioning

Time-Based Partitioning

-- Monthly partitions (most common)
CREATE TABLE events (
    event_id UUID,
    user_id UInt64,
    event_type String,
    created_at DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (user_id, created_at)
SETTINGS index_granularity = 8192;

-- Daily partitions (high volume)
CREATE TABLE events_daily (
    event_id UUID,
    user_id UInt64,
    created_at DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(created_at)
ORDER BY (user_id, created_at);

-- Weekly partitions
CREATE TABLE events_weekly (
    event_id UUID,
    created_at DateTime
)
ENGINE = MergeTree()
PARTITION BY toMonday(created_at)
ORDER BY created_at;

Composite Partitioning

-- Region + Month
CREATE TABLE sales (
    sale_id UUID,
    region String,
    amount Decimal(18, 2),
    sale_date Date
)
ENGINE = MergeTree()
PARTITION BY (region, toYYYYMM(sale_date))
ORDER BY (sale_date, sale_id);

Ordering Key Strategy

-- Order by query patterns
-- Most selective column first
CREATE TABLE user_events (
    user_id UInt64,
    session_id UUID,
    event_type LowCardinality(String),
    created_at DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (user_id, session_id, created_at)
-- Queries filtering by user_id will be fastest

Skip Indices

CREATE TABLE events (
    event_id UUID,
    user_id UInt64,
    event_type String,
    properties String,
    created_at DateTime,
    -- Skip indices for additional columns
    INDEX idx_event_type event_type TYPE set(100) GRANULARITY 4,
    INDEX idx_properties properties TYPE tokenbf_v1(10240, 3, 0) GRANULARITY 4
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (user_id, created_at);

---

Apache Iceberg Partitioning

Hidden Partitioning

-- Iceberg transforms (no partition columns in data)
CREATE TABLE events (
    event_id STRING,
    user_id BIGINT,
    event_type STRING,
    created_at TIMESTAMP
)
USING iceberg
PARTITIONED BY (
    months(created_at),      -- Monthly partitions
    bucket(16, user_id)      -- Hash buckets for user_id
);

Partition Transforms

-- Time transforms
PARTITIONED BY (
    years(created_at),       -- Year
    months(created_at),      -- Month
    days(created_at),        -- Day
    hours(created_at)        -- Hour
)

-- Identity (exact value)
PARTITIONED BY (
    region,                  -- Each region = partition
    identity(status)         -- Each status = partition
)

-- Bucket (hash)
PARTITIONED BY (
    bucket(16, user_id),     -- 16 buckets by user_id hash
    bucket(8, order_id)
)

-- Truncate (for strings/numbers)
PARTITIONED BY (
    truncate(2, zip_code)    -- First 2 chars of zip
)

Partition Evolution

-- Add partition field (no rewrite needed)
ALTER TABLE events ADD PARTITION FIELD bucket(16, user_id);

-- Remove partition field
ALTER TABLE events DROP PARTITION FIELD months(created_at);

-- Change partition granularity
ALTER TABLE events REPLACE PARTITION FIELD
    months(created_at) WITH days(created_at);

Query Optimization

# PyIceberg - partition pruning
from pyiceberg.catalog import load_catalog

catalog = load_catalog("rest")
table = catalog.load_table("db.events")

# Efficient query (uses partition pruning)
scan = table.scan(
    filter="created_at >= '2024-01-01' AND created_at < '2024-02-01'"
)
# Only reads January partition files

# Check partition metrics
for file in table.scan().to_arrow().column("file_path"):
    print(file)

---

Delta Lake Partitioning

Basic Partitioning

# Create partitioned table
df.write.format("delta") \
    .partitionBy("year", "month") \
    .save("/delta/events")

# SQL
spark.sql("""
    CREATE TABLE events (
        event_id STRING,
        user_id BIGINT,
        created_at TIMESTAMP,
        year INT GENERATED ALWAYS AS (YEAR(created_at)),
        month INT GENERATED ALWAYS AS (MONTH(created_at))
    )
    USING DELTA
    PARTITIONED BY (year, month)
""")

Z-Order Clustering

-- Optimize for multi-dimensional queries
OPTIMIZE events ZORDER BY (user_id, event_type);

-- Combine with partitioning
-- Partition by time, Z-order by query columns
OPTIMIZE events
WHERE created_at >= '2024-01-01'
ZORDER BY (user_id, product_id);

Liquid Clustering (Delta 3.0+)

-- Auto-clustering without explicit partitioning
CREATE TABLE events (
    event_id STRING,
    user_id BIGINT,
    event_type STRING,
    created_at TIMESTAMP
)
USING DELTA
CLUSTER BY (user_id, event_type);

-- Trigger clustering
OPTIMIZE events;

---

DuckDB Partitioning

Hive-Style Partitioning

-- Write partitioned Parquet
COPY events TO 's3://bucket/events'
(FORMAT PARQUET, PARTITION_BY (year, month));

-- Creates structure:
-- s3://bucket/events/year=2024/month=01/data.parquet
-- s3://bucket/events/year=2024/month=02/data.parquet

Query Partitioned Data

-- DuckDB auto-detects partition columns
SELECT * FROM read_parquet('s3://bucket/events/*/*.parquet',
    hive_partitioning=true)
WHERE year = 2024 AND month = 1;
-- Only reads matching partitions

---

Partition Sizing Guidelines

Target Partition Size

Anti-Patterns

-- [FAIL] Too many partitions (>10k)
PARTITION BY toYYYYMMDD(created_at)  -- Daily for 10 years = 3650 partitions

-- [FAIL] Too few records per partition
PARTITION BY (region, category, toYYYYMMDD(date))  -- Explosive combination

-- [FAIL] High-cardinality partition key
PARTITION BY user_id  -- Millions of partitions

-- [OK] Balanced approach
PARTITION BY toYYYYMM(created_at)  -- ~120 partitions over 10 years

---

ClickHouse Partition Management

View Partitions

-- List partitions
SELECT
    partition,
    name,
    rows,
    formatReadableSize(bytes_on_disk) AS size,
    modification_time
FROM system.parts
WHERE table = 'events' AND active
ORDER BY partition DESC;

-- Partition statistics
SELECT
    partition,
    count() AS parts,
    sum(rows) AS total_rows,
    formatReadableSize(sum(bytes_on_disk)) AS total_size
FROM system.parts
WHERE table = 'events' AND active
GROUP BY partition
ORDER BY partition DESC;

Partition Operations

-- Detach partition (keeps data, removes from table)
ALTER TABLE events DETACH PARTITION 202301;

-- Attach partition back
ALTER TABLE events ATTACH PARTITION 202301;

-- Drop partition (deletes data)
ALTER TABLE events DROP PARTITION 202301;

-- Move partition to another table
ALTER TABLE events MOVE PARTITION 202301 TO TABLE events_archive;

-- Freeze partition (backup)
ALTER TABLE events FREEZE PARTITION 202301;

TTL with Partitions

CREATE TABLE events (
    event_id UUID,
    created_at DateTime,
    data String
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY created_at
TTL created_at + INTERVAL 2 YEAR DELETE;
-- Drops old partitions automatically

---

Best Practices

DO

1. Partition by query patterns - most filtered column 2. Keep partitions 1-50 GB - too small = overhead, too large = slow queries 3. Use time-based for time-series - natural data locality 4. Add secondary clustering - Z-order or ordering key 5. Monitor partition sizes - rebalance if skewed

DON'T

1. Don't partition by high-cardinality columns - user_id, session_id 2. Don't create >10k partitions - metadata overhead 3. Don't skip partitioning for large tables - full scans are slow 4. Don't ignore partition pruning - always filter by partition key 5. Don't mix partition and sort key - they serve different purposes

---

Partition Pruning Validation

-- ClickHouse: Check if partition pruning works
EXPLAIN indexes = 1
SELECT * FROM events
WHERE created_at >= '2024-01-01' AND created_at < '2024-02-01';

-- Look for "Parts: X/Y" where X << Y indicates pruning worked

-- Query system.query_log for actual parts read
SELECT
    query,
    read_rows,
    read_bytes,
    result_rows
FROM system.query_log
WHERE query LIKE '%events%'
ORDER BY event_time DESC
LIMIT 10;

Schema Evolution Template

Overview

Patterns for handling schema changes in data lake tables without breaking downstream consumers.

Schema Change Types

---

Apache Iceberg Schema Evolution

Add Column

-- Spark SQL
ALTER TABLE catalog.db.events ADD COLUMN user_agent STRING AFTER user_id;

-- With default value
ALTER TABLE catalog.db.events ADD COLUMN is_processed BOOLEAN DEFAULT false;

-- Nested column
ALTER TABLE catalog.db.events ADD COLUMN metadata.source STRING;

Rename Column

-- Safe rename (preserves data)
ALTER TABLE catalog.db.events RENAME COLUMN user_agent TO browser_info;

-- Iceberg tracks column by ID, not name
-- Downstream queries using old name will fail

Change Column Type

-- Widen type (safe)
ALTER TABLE catalog.db.events ALTER COLUMN user_id TYPE BIGINT;

-- Iceberg supports: int → long, float → double, decimal precision increase

Drop Column

-- Iceberg "soft deletes" columns (data remains in files)
ALTER TABLE catalog.db.events DROP COLUMN deprecated_field;

-- Files not rewritten - just metadata update
-- Query with time travel still sees old column

Required to Optional

ALTER TABLE catalog.db.events ALTER COLUMN email DROP NOT NULL;

Schema Evolution Best Practices (Iceberg)

# PyIceberg schema operations
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import NestedField, StringType, LongType

catalog = load_catalog("rest")
table = catalog.load_table("db.events")

# Get current schema
current_schema = table.schema()
print(f"Current schema version: {table.metadata.current_schema_id}")

# Add column with update
with table.update_schema() as update:
    update.add_column("new_field", StringType(), doc="New field description")
    update.add_column("nested.subfield", LongType())

# View schema history
for schema in table.metadata.schemas:
    print(f"Schema {schema.schema_id}: {len(schema.fields)} fields")

---

Delta Lake Schema Evolution

Schema Enforcement vs Evolution

# Default: Schema enforcement (rejects mismatched schemas)
df.write.format("delta").mode("append").save("/delta/events")

# Enable schema evolution
df.write.format("delta") \
    .option("mergeSchema", "true") \
    .mode("append") \
    .save("/delta/events")

# Or set at table level
spark.sql("""
    ALTER TABLE delta.events
    SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name')
""")

Column Mapping (Delta 2.0+)

-- Enable column mapping for renames/drops
ALTER TABLE events SET TBLPROPERTIES (
    'delta.columnMapping.mode' = 'name',
    'delta.minReaderVersion' = '2',
    'delta.minWriterVersion' = '5'
);

-- Now can rename columns
ALTER TABLE events RENAME COLUMN old_name TO new_name;

-- And drop columns
ALTER TABLE events DROP COLUMN deprecated_field;

Schema Evolution Operations

-- Add column
ALTER TABLE events ADD COLUMN new_field STRING AFTER existing_field;

-- Change type (only widening allowed)
ALTER TABLE events ALTER COLUMN amount TYPE DECIMAL(20, 4);

-- Add constraint
ALTER TABLE events ADD CONSTRAINT positive_amount CHECK (amount >= 0);

---

ClickHouse Schema Evolution

Add Column

-- Add column with default
ALTER TABLE events ADD COLUMN browser_info String DEFAULT 'unknown';

-- Add column at position
ALTER TABLE events ADD COLUMN session_id UUID AFTER user_id;

-- For distributed tables
ALTER TABLE events ON CLUSTER my_cluster ADD COLUMN browser_info String;

Modify Column

-- Change type
ALTER TABLE events MODIFY COLUMN user_id UInt64;  -- Was UInt32

-- Change default
ALTER TABLE events MODIFY COLUMN status String DEFAULT 'pending';

-- Add codec for compression
ALTER TABLE events MODIFY COLUMN event_data String CODEC(ZSTD(3));

Rename Column (ClickHouse 23.4+)

ALTER TABLE events RENAME COLUMN old_name TO new_name;

Drop Column

-- Immediate drop (data remains until merge)
ALTER TABLE events DROP COLUMN deprecated_field;

-- Clear data explicitly
ALTER TABLE events CLEAR COLUMN deprecated_field;

-- Distributed table
ALTER TABLE events ON CLUSTER my_cluster DROP COLUMN deprecated_field;

Materialized Column

-- Add computed column
ALTER TABLE events ADD COLUMN day Date MATERIALIZED toDate(created_at);

-- Update existing data
ALTER TABLE events MATERIALIZE COLUMN day;

---

SQLMesh Schema Management

Model Schema Definition

-- models/staging/stg_events.sql
MODEL (
  name silver.stg_events,
  kind INCREMENTAL_BY_TIME_RANGE (time_column created_at),
  columns (
    event_id STRING NOT NULL,
    user_id INT64 NOT NULL,
    event_type STRING NOT NULL,
    -- New column: browser_info added 2024-06-01
    browser_info STRING,
    created_at TIMESTAMP NOT NULL
  )
);

Plan and Apply Changes

# Preview schema changes
sqlmesh plan

# Output shows:
# Models:
#   silver.stg_events (schema change)
#     Added columns:
#       - browser_info STRING

# Apply changes
sqlmesh plan --auto-apply

Backfill After Schema Change

# Backfill data for new column
sqlmesh plan --restate-model silver.stg_events --start 2024-01-01

---

dlt Schema Evolution

Auto Schema Evolution

import dlt

# dlt auto-detects and evolves schema
pipeline = dlt.pipeline(
    pipeline_name="events",
    destination="clickhouse"
)

# First load: creates table with initial schema
pipeline.run(events_batch_1)

# Second load: auto-adds new columns if data has them
pipeline.run(events_batch_2_with_new_fields)

# Check schema
print(pipeline.default_schema.tables["events"])

Schema Contract Mode

# Strict mode: reject schema changes
@dlt.resource(
    name="events",
    schema_contract={"tables": "freeze", "columns": "freeze"}
)
def events():
    yield data

# Evolve mode: allow additive changes only
@dlt.resource(
    schema_contract={"tables": "evolve", "columns": "evolve"}
)
def events():
    yield data

# Discard mode: drop unknown columns silently
@dlt.resource(
    schema_contract={"columns": "discard_value"}
)
def events():
    yield data

Manual Schema Definition

@dlt.resource(
    columns={
        "event_id": {"data_type": "text", "nullable": False},
        "user_id": {"data_type": "bigint", "nullable": False},
        "amount": {"data_type": "decimal", "precision": 18, "scale": 2},
        "created_at": {"data_type": "timestamp", "nullable": False}
    }
)
def events():
    yield data

---

Migration Patterns

Pattern 1: Add Column with Backfill

-- Step 1: Add nullable column
ALTER TABLE events ADD COLUMN browser_info String;

-- Step 2: Backfill from another source
ALTER TABLE events UPDATE browser_info = extractBrowserFromUserAgent(user_agent)
WHERE browser_info IS NULL OR browser_info = '';

-- Step 3: Optionally add NOT NULL constraint after backfill
-- (ClickHouse doesn't support ALTER COLUMN NOT NULL)

Pattern 2: Rename Column Safely

-- Step 1: Add new column
ALTER TABLE events ADD COLUMN browser_info String;

-- Step 2: Copy data
ALTER TABLE events UPDATE browser_info = user_agent WHERE 1=1;

-- Step 3: Update downstream consumers (allow grace period)

-- Step 4: Drop old column
ALTER TABLE events DROP COLUMN user_agent;

Pattern 3: Type Change with New Column

-- Can't change String to Int directly
-- Step 1: Add new column with correct type
ALTER TABLE events ADD COLUMN user_id_new UInt64;

-- Step 2: Copy and convert
ALTER TABLE events UPDATE user_id_new = toUInt64(user_id) WHERE 1=1;

-- Step 3: Rename columns
ALTER TABLE events RENAME COLUMN user_id TO user_id_old;
ALTER TABLE events RENAME COLUMN user_id_new TO user_id;

-- Step 4: Drop old column after validation
ALTER TABLE events DROP COLUMN user_id_old;

---

Version Tracking

Schema Registry Table

CREATE TABLE meta.schema_versions (
    table_name String,
    version UInt32,
    schema_json String,
    created_at DateTime DEFAULT now(),
    created_by String,
    change_description String
)
ENGINE = MergeTree()
ORDER BY (table_name, version);

-- Record schema changes
INSERT INTO meta.schema_versions VALUES
('silver.stg_events', 2, '{"columns":[...]}', now(), 'data-team', 'Added browser_info column');

Migration Scripts Directory

migrations/
├── events/
│   ├── v001_initial_schema.sql
│   ├── v002_add_browser_info.sql
│   ├── v003_rename_user_agent.sql
│   └── v004_add_session_id.sql
└── orders/
    ├── v001_initial_schema.sql
    └── v002_add_discount_field.sql

---

Best Practices

1. Always add columns as nullable first - then backfill and add constraints 2. Never rename columns directly - add new, migrate, drop old 3. Document all schema changes - in migration scripts and registry 4. Test with production data copies - before applying to production 5. Communicate changes to consumers - allow grace period for updates 6. Use schema contracts - enforce compatibility in CI/CD 7. Track schema versions - for debugging and rollback

Airbyte Connection Template

Overview

Setting up Airbyte connectors for data ingestion.

Quick Setup

Docker Compose

# Clone Airbyte
git clone https://github.com/airbytehq/airbyte.git
cd airbyte

# Start Airbyte
./run-ab-platform.sh

# Access UI: http://localhost:8000
# Default: airbyte / password

Kubernetes (Helm)

helm repo add airbyte https://airbytehq.github.io/helm-charts
helm install airbyte airbyte/airbyte -n airbyte --create-namespace

---

Connection Configuration

Source: PostgreSQL

# Terraform / API config
source:
  name: "postgres-source"
  sourceDefinitionId: "decd338e-5647-4c0b-adf4-da0e75f5a750"
  connectionConfiguration:
    host: "postgres.example.com"
    port: 5432
    database: "production"
    username: "${POSTGRES_USER}"
    password: "${POSTGRES_PASSWORD}"
    schemas: ["public"]
    ssl_mode:
      mode: "require"
    replication_method:
      method: "CDC"
      plugin: "pgoutput"
      publication: "airbyte_publication"
      replication_slot: "airbyte_slot"

Source: REST API

source:
  name: "api-source"
  sourceDefinitionId: "dfd88b22-b603-4c3d-aad7-3701784586b1"
  connectionConfiguration:
    api_url: "https://api.example.com"
    authentication:
      type: "Bearer"
      api_token: "${API_TOKEN}"
    pagination:
      type: "CursorPagination"
      cursor_value: "{{ response.next_cursor }}"

Destination: ClickHouse

destination:
  name: "clickhouse-dest"
  destinationDefinitionId: "ce0d828e-1dc4-496c-b122-2da42e637e48"
  connectionConfiguration:
    host: "clickhouse.example.com"
    port: 8123
    database: "bronze"
    username: "${CLICKHOUSE_USER}"
    password: "${CLICKHOUSE_PASSWORD}"
    ssl: true
    raw_data_schema: "_airbyte_raw"

Destination: S3 (Data Lake)

destination:
  name: "s3-dest"
  destinationDefinitionId: "4816b78f-1489-44c1-9060-4b19d5fa9362"
  connectionConfiguration:
    s3_bucket_name: "data-lake-bronze"
    s3_bucket_path: "airbyte/${SOURCE_NAME}"
    s3_bucket_region: "us-east-1"
    format:
      format_type: "Parquet"
      compression_codec: "ZSTD"
    access_key_id: "${AWS_ACCESS_KEY}"
    secret_access_key: "${AWS_SECRET_KEY}"

---

Connection Settings

Sync Configuration

connection:
  name: "postgres-to-clickhouse"
  sourceId: "source-uuid"
  destinationId: "destination-uuid"

  # Sync mode
  syncCatalog:
    streams:
      - stream:
          name: "users"
          namespace: "public"
        config:
          syncMode: "incremental"
          destinationSyncMode: "append_dedup"
          cursorField: ["updated_at"]
          primaryKey: [["id"]]

      - stream:
          name: "events"
          namespace: "public"
        config:
          syncMode: "incremental"
          destinationSyncMode: "append"
          cursorField: ["created_at"]

  # Schedule
  scheduleType: "cron"
  scheduleData:
    cron:
      cronExpression: "0 */6 * * *"  # Every 6 hours
      cronTimeZone: "UTC"

  # Normalization
  normalizationOperation: "basic"

  # Resource requirements
  resourceRequirements:
    cpu_request: "1"
    cpu_limit: "2"
    memory_request: "1Gi"
    memory_limit: "2Gi"

Sync Modes

---

API Usage

Create Connection (Python)

import requests

AIRBYTE_URL = "http://localhost:8000/api/v1"

# Create source
source_response = requests.post(
    f"{AIRBYTE_URL}/sources/create",
    json={
        "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750",
        "workspaceId": "workspace-uuid",
        "name": "postgres-source",
        "connectionConfiguration": {
            "host": "postgres.example.com",
            "port": 5432,
            "database": "production",
            "username": "airbyte",
            "password": "secret"
        }
    }
)

# Create destination
dest_response = requests.post(
    f"{AIRBYTE_URL}/destinations/create",
    json={
        "destinationDefinitionId": "ce0d828e-1dc4-496c-b122-2da42e637e48",
        "workspaceId": "workspace-uuid",
        "name": "clickhouse-dest",
        "connectionConfiguration": {
            "host": "clickhouse.example.com",
            "port": 8123,
            "database": "bronze"
        }
    }
)

# Create connection
conn_response = requests.post(
    f"{AIRBYTE_URL}/connections/create",
    json={
        "sourceId": source_response.json()["sourceId"],
        "destinationId": dest_response.json()["destinationId"],
        "syncCatalog": {...},
        "scheduleType": "cron",
        "scheduleData": {
            "cron": {
                "cronExpression": "0 */6 * * *",
                "cronTimeZone": "UTC"
            }
        }
    }
)

Trigger Sync

# Manual sync trigger
requests.post(
    f"{AIRBYTE_URL}/connections/sync",
    json={"connectionId": "connection-uuid"}
)

# Check sync status
status = requests.post(
    f"{AIRBYTE_URL}/jobs/get",
    json={"id": job_id}
)

---

Terraform

# provider.tf
terraform {
  required_providers {
    airbyte = {
      source  = "airbytehq/airbyte"
      version = "~> 0.3"
    }
  }
}

provider "airbyte" {
  server_url = "http://localhost:8000/api/public/v1"
  username   = var.airbyte_username
  password   = var.airbyte_password
}

# source.tf
resource "airbyte_source_postgres" "postgres" {
  name          = "postgres-source"
  workspace_id  = airbyte_workspace.main.workspace_id

  configuration = {
    host     = "postgres.example.com"
    port     = 5432
    database = "production"
    username = var.postgres_username
    password = var.postgres_password
    schemas  = ["public"]
  }
}

# destination.tf
resource "airbyte_destination_clickhouse" "clickhouse" {
  name          = "clickhouse-dest"
  workspace_id  = airbyte_workspace.main.workspace_id

  configuration = {
    host     = "clickhouse.example.com"
    port     = 8123
    database = "bronze"
    username = var.clickhouse_username
    password = var.clickhouse_password
  }
}

# connection.tf
resource "airbyte_connection" "postgres_to_clickhouse" {
  name           = "postgres-to-clickhouse"
  source_id      = airbyte_source_postgres.postgres.source_id
  destination_id = airbyte_destination_clickhouse.clickhouse.destination_id

  schedule = {
    schedule_type = "cron"
    cron_expression = "0 */6 * * *"
  }
}

---

Monitoring

Check Sync Status

def get_sync_status(connection_id):
    response = requests.post(
        f"{AIRBYTE_URL}/jobs/list",
        json={
            "configTypes": ["sync"],
            "configId": connection_id
        }
    )

    jobs = response.json()["jobs"]
    if jobs:
        latest = jobs[0]
        return {
            "status": latest["job"]["status"],
            "started_at": latest["job"]["createdAt"],
            "bytes_synced": latest["attempts"][-1].get("bytesSynced"),
            "records_synced": latest["attempts"][-1].get("recordsSynced")
        }

Alerts

# Prometheus alerts
groups:
  - name: airbyte
    rules:
      - alert: AirbyteSyncFailed
        expr: airbyte_job_status{status="failed"} > 0
        for: 5m
        labels:
          severity: critical

      - alert: AirbyteSyncStale
        expr: time() - airbyte_last_successful_sync_timestamp > 86400
        for: 1h
        labels:
          severity: warning

---

Best Practices

1. Use CDC for databases - Lower latency, less load 2. Set appropriate schedules - Based on data freshness needs 3. Configure resource limits - Prevent memory issues 4. Enable normalization - For structured destination tables 5. Monitor sync durations - Alert on anomalies 6. Use secrets management - Never hardcode credentials

dlt Database Source Template

Purpose: Extract data from relational databases (Postgres, MySQL, MongoDB) using dlt for ELT pipelines.

Installation

# PostgreSQL
pip install dlt[postgres]
pip install psycopg2-binary

# MySQL
pip install dlt[mysql]
pip install pymysql

# MongoDB
pip install dlt[mongodb]
pip install pymongo

# SQL Server
pip install dlt[mssql]
pip install pyodbc

PostgreSQL Source

Basic Table Extraction

import dlt
from dlt.sources.sql_database import sql_database

# Extract all tables from schema
source = sql_database(
    credentials="postgresql://user:password@localhost:5432/database",
    schema="public"
)

pipeline = dlt.pipeline(
    pipeline_name="postgres_to_duckdb",
    destination="duckdb",
    dataset_name="postgres_raw"
)

load_info = pipeline.run(source)
print(load_info)

Select Specific Tables

source = sql_database(
    credentials="postgresql://user:password@localhost:5432/database",
    schema="public",
    table_names=["customers", "orders", "products"]
)

pipeline = dlt.pipeline(
    pipeline_name="postgres_selected",
    destination="bigquery",
    dataset_name="postgres_raw"
)

load_info = pipeline.run(source)

Custom SQL Query

import dlt
from sqlalchemy import create_engine

# Create engine
engine = create_engine("postgresql://user:password@localhost:5432/database")

# Define custom query
@dlt.resource(name="active_customers")
def get_active_customers():
    query = """
    SELECT
        customer_id,
        email,
        created_at,
        last_order_date
    FROM customers
    WHERE status = 'active'
      AND last_order_date >= CURRENT_DATE - INTERVAL '30 days'
    """

    with engine.connect() as conn:
        result = conn.execute(query)
        yield from result

pipeline = dlt.pipeline(
    pipeline_name="custom_query",
    destination="snowflake",
    dataset_name="analytics"
)

load_info = pipeline.run([get_active_customers()])

MySQL Source

Extract with Reflection

import dlt
from dlt.sources.sql_database import sql_database

source = sql_database(
    credentials="mysql+pymysql://user:password@localhost:3306/database",
    schema="production",
    reflection_level="full"  # Extract schema metadata
)

pipeline = dlt.pipeline(
    pipeline_name="mysql_to_postgres",
    destination="postgres",
    dataset_name="mysql_raw"
)

load_info = pipeline.run(source)

Incremental Loading by Timestamp

import dlt
from dlt.sources.sql_database import sql_table

@dlt.resource(
    name="orders",
    write_disposition="append",
    primary_key="order_id"
)
def load_orders_incremental():
    # Use dlt.sources.incremental for cursor-based loading
    last_value = dlt.sources.incremental(
        "updated_at",
        initial_value="2024-01-01T00:00:00"
    )

    from sqlalchemy import create_engine
    engine = create_engine("mysql+pymysql://user:password@localhost:3306/database")

    query = f"""
    SELECT * FROM orders
    WHERE updated_at >= '{last_value.start_value}'
    ORDER BY updated_at
    """

    with engine.connect() as conn:
        result = conn.execute(query)
        yield from result

pipeline = dlt.pipeline(
    pipeline_name="mysql_incremental",
    destination="duckdb",
    dataset_name="mysql_raw"
)

load_info = pipeline.run([load_orders_incremental()])

MongoDB Source

Collection Extraction

import dlt
from pymongo import MongoClient

@dlt.resource(name="users")
def load_mongodb_collection():
    client = MongoClient("mongodb://localhost:27017/")
    db = client["production"]
    collection = db["users"]

    # Stream documents
    for doc in collection.find():
        # Remove MongoDB _id (not JSON serializable)
        doc["_id"] = str(doc["_id"])
        yield doc

pipeline = dlt.pipeline(
    pipeline_name="mongodb_to_postgres",
    destination="postgres",
    dataset_name="mongodb_raw"
)

load_info = pipeline.run([load_mongodb_collection()])

Incremental MongoDB Loading

import dlt
from pymongo import MongoClient
from datetime import datetime

@dlt.resource(
    name="events",
    write_disposition="append",
    primary_key="event_id"
)
def load_events_incremental():
    last_timestamp = dlt.sources.incremental(
        "timestamp",
        initial_value=datetime(2024, 1, 1)
    )

    client = MongoClient("mongodb://localhost:27017/")
    db = client["analytics"]
    collection = db["events"]

    # Query with incremental filter
    query = {"timestamp": {"$gte": last_timestamp.start_value}}

    for doc in collection.find(query).sort("timestamp", 1):
        doc["_id"] = str(doc["_id"])
        yield doc

pipeline = dlt.pipeline(
    pipeline_name="mongodb_incremental",
    destination="snowflake",
    dataset_name="events_raw"
)

load_info = pipeline.run([load_events_incremental()])

SQL Server Source

import dlt
from dlt.sources.sql_database import sql_database

source = sql_database(
    credentials="mssql+pyodbc://user:password@localhost:1433/database?driver=ODBC+Driver+17+for+SQL+Server",
    schema="dbo",
    table_names=["sales", "customers"]
)

pipeline = dlt.pipeline(
    pipeline_name="sqlserver_to_bigquery",
    destination="bigquery",
    dataset_name="sqlserver_raw"
)

load_info = pipeline.run(source)

Advanced Patterns

Parallel Table Loading

import dlt
from dlt.sources.sql_database import sql_database

source = sql_database(
    credentials="postgresql://user:password@localhost:5432/database",
    schema="public",
    parallel=True,  # Enable parallel extraction
    chunk_size=10000  # Rows per chunk
)

pipeline = dlt.pipeline(
    pipeline_name="parallel_extract",
    destination="duckdb",
    dataset_name="postgres_raw"
)

load_info = pipeline.run(source)

Schema Evolution Handling

import dlt

source = sql_database(
    credentials="postgresql://user:password@localhost:5432/database",
    schema="public",
    detect_precision_hints=True  # Preserve column types
)

pipeline = dlt.pipeline(
    pipeline_name="schema_evolution",
    destination="postgres",
    dataset_name="raw"
)

# Schema will auto-evolve if source changes
load_info = pipeline.run(source)

Cross-Database Joins

import dlt
from sqlalchemy import create_engine

# Extract from two databases
@dlt.resource(name="enriched_orders")
def join_across_databases():
    pg_engine = create_engine("postgresql://user:password@localhost:5432/sales")
    mysql_engine = create_engine("mysql+pymysql://user:password@localhost:3306/customers")

    # Extract from PostgreSQL
    with pg_engine.connect() as conn:
        orders = conn.execute("SELECT * FROM orders").fetchall()

    # Extract from MySQL
    with mysql_engine.connect() as conn:
        customers = {row[0]: row for row in conn.execute("SELECT * FROM customers")}

    # Join in Python
    for order in orders:
        customer = customers.get(order.customer_id)
        if customer:
            yield {
                **dict(order),
                "customer_name": customer.name,
                "customer_email": customer.email
            }

pipeline = dlt.pipeline(
    pipeline_name="cross_db_join",
    destination="snowflake",
    dataset_name="enriched"
)

load_info = pipeline.run([join_across_databases()])

Configuration (.dlt/config.toml)

[sources.postgres]
schema = "public"
chunk_size = 10000
parallel = true

[sources.mysql]
schema = "production"
reflection_level = "full"

[sources.mongodb]
database = "analytics"
batch_size = 1000

[destination.postgres]
credentials = "postgres://user:password@localhost:5432/warehouse"

[destination.snowflake]
database = "ANALYTICS"
schema = "RAW_DATA"

Secrets (.dlt/secrets.toml)

[sources.postgres.credentials]
database = "sales"
username = "etl_user"
password = "your_password"
host = "localhost"
port = 5432

[sources.mysql.credentials]
database = "customers"
username = "etl_user"
password = "your_password"
host = "localhost"
port = 3306

[sources.mongodb.credentials]
connection_string = "mongodb://user:password@localhost:27017/"

[sources.mssql.credentials]
connection_string = "mssql+pyodbc://user:password@localhost:1433/database?driver=ODBC+Driver+17+for+SQL+Server"

Performance Optimization

Chunked Extraction

import dlt
from dlt.sources.sql_database import sql_table

@dlt.resource(name="large_table")
def extract_in_chunks():
    from sqlalchemy import create_engine
    engine = create_engine("postgresql://user:password@localhost:5432/database")

    chunk_size = 50000
    offset = 0

    while True:
        query = f"SELECT * FROM large_table LIMIT {chunk_size} OFFSET {offset}"

        with engine.connect() as conn:
            result = conn.execute(query).fetchall()

            if not result:
                break

            yield from result
            offset += chunk_size

pipeline = dlt.pipeline(
    pipeline_name="chunked_extract",
    destination="duckdb",
    dataset_name="raw"
)

load_info = pipeline.run([extract_in_chunks()])

Connection Pooling

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    "postgresql://user:password@localhost:5432/database",
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

Monitoring

import dlt

pipeline = dlt.pipeline(
    pipeline_name="monitored_pipeline",
    destination="postgres",
    dataset_name="raw"
)

load_info = pipeline.run(source)

# Check statistics
print(f"Packages: {len(load_info.load_packages)}")
print(f"Loaded tables: {load_info.loads_ids}")

# Query loaded data
with pipeline.sql_client() as client:
    result = client.execute_sql("SELECT COUNT(*) FROM customers")
    print(f"Loaded {result[0][0]} customers")

Best Practices

• BEST: Use connection pooling for large extractions
• BEST: Implement incremental loading for large tables
• BEST: Extract in parallel when possible
• BEST: Use chunking for tables with millions of rows
• BEST: Store credentials in secrets.toml
• BEST: Enable schema evolution for dynamic sources
• BEST: Monitor extraction performance and row counts
• BEST: Use SQL queries to filter data at source (reduce transfer)

dlt Incremental Loading Template

Purpose: Implement efficient incremental data loading strategies using dlt to avoid full table scans.

Why Incremental Loading?

• Reduce data transfer: Load only new/changed records
• Faster pipelines: Avoid scanning entire tables
• Lower costs: Minimize compute and network usage
• Enable real-time: Process changes as they happen

Incremental by Timestamp

Basic Timestamp Tracking

import dlt
from dlt.sources.incremental import Incremental

@dlt.resource(
    name="orders",
    write_disposition="append",
    primary_key="order_id"
)
def load_orders_incremental():
    # Track last processed timestamp
    last_timestamp = dlt.sources.incremental(
        cursor_path="updated_at",
        initial_value="2024-01-01T00:00:00"
    )

    # Your data source (API, database, etc.)
    orders = fetch_orders_since(last_timestamp.start_value)

    yield orders

pipeline = dlt.pipeline(
    pipeline_name="orders_incremental",
    destination="duckdb",
    dataset_name="sales"
)

load_info = pipeline.run([load_orders_incremental()])

Database Query with Incremental Filter

import dlt
from sqlalchemy import create_engine

@dlt.resource(
    name="users",
    write_disposition="merge",  # Upsert existing records
    primary_key="user_id"
)
def load_users_incremental():
    last_updated = dlt.sources.incremental(
        cursor_path="updated_at",
        initial_value="2024-01-01"
    )

    engine = create_engine("postgresql://user:password@localhost:5432/db")

    query = f"""
    SELECT * FROM users
    WHERE updated_at >= '{last_updated.start_value}'
    ORDER BY updated_at
    """

    with engine.connect() as conn:
        result = conn.execute(query)
        yield from result

pipeline = dlt.pipeline(
    pipeline_name="users_incremental",
    destination="postgres",
    dataset_name="raw"
)

load_info = pipeline.run([load_users_incremental()])

REST API with Date Range

import dlt
from datetime import datetime, timedelta

@dlt.resource(
    name="api_events",
    write_disposition="append"
)
def load_events_incremental():
    last_date = dlt.sources.incremental(
        cursor_path="event_date",
        initial_value="2024-01-01"
    )

    # API endpoint with date filter
    response = requests.get(
        "https://api.example.com/events",
        params={
            "start_date": last_date.start_value,
            "end_date": datetime.now().isoformat()
        }
    )

    yield response.json()["events"]

pipeline = dlt.pipeline(
    pipeline_name="events_incremental",
    destination="snowflake",
    dataset_name="events"
)

load_info = pipeline.run([load_events_incremental()])

Incremental by ID (Auto-Increment)

Track Last Processed ID

import dlt

@dlt.resource(
    name="transactions",
    write_disposition="append",
    primary_key="transaction_id"
)
def load_transactions_incremental():
    last_id = dlt.sources.incremental(
        cursor_path="transaction_id",
        initial_value=0
    )

    # Fetch records with ID > last_id
    transactions = fetch_transactions_after(last_id.start_value)

    yield transactions

pipeline = dlt.pipeline(
    pipeline_name="transactions_incremental",
    destination="bigquery",
    dataset_name="finance"
)

load_info = pipeline.run([load_transactions_incremental()])

Incremental with Merge (Upsert)

Handle Updates and Deletes

import dlt

@dlt.resource(
    name="products",
    write_disposition="merge",  # Upsert mode
    primary_key="product_id",
    merge_key="updated_at"  # Track changes by timestamp
)
def load_products_incremental():
    last_updated = dlt.sources.incremental(
        cursor_path="updated_at",
        initial_value="2024-01-01"
    )

    from sqlalchemy import create_engine
    engine = create_engine("mysql+pymysql://user:password@localhost:3306/db")

    query = f"""
    SELECT * FROM products
    WHERE updated_at >= '{last_updated.start_value}'
    ORDER BY updated_at
    """

    with engine.connect() as conn:
        result = conn.execute(query)
        yield from result

pipeline = dlt.pipeline(
    pipeline_name="products_incremental",
    destination="postgres",
    dataset_name="catalog"
)

load_info = pipeline.run([load_products_incremental()])

Soft Deletes with Merge

import dlt

@dlt.resource(
    name="customers",
    write_disposition="merge",
    primary_key="customer_id"
)
def load_customers_with_deletes():
    last_updated = dlt.sources.incremental(
        cursor_path="updated_at",
        initial_value="2024-01-01"
    )

    # Fetch active and deleted records
    query = f"""
    SELECT
        customer_id,
        name,
        email,
        updated_at,
        is_deleted
    FROM customers
    WHERE updated_at >= '{last_updated.start_value}'
    ORDER BY updated_at
    """

    from sqlalchemy import create_engine
    engine = create_engine("postgresql://user:password@localhost:5432/db")

    with engine.connect() as conn:
        result = conn.execute(query)
        yield from result

pipeline = dlt.pipeline(
    pipeline_name="customers_with_deletes",
    destination="snowflake",
    dataset_name="crm"
)

load_info = pipeline.run([load_customers_with_deletes()])

Incremental with Lookback Window

Handle Late-Arriving Data

import dlt
from datetime import datetime, timedelta

@dlt.resource(
    name="orders_with_lookback",
    write_disposition="merge",
    primary_key="order_id"
)
def load_orders_with_lookback():
    last_updated = dlt.sources.incremental(
        cursor_path="updated_at",
        initial_value="2024-01-01"
    )

    # Lookback 7 days to catch late updates
    lookback_start = (
        datetime.fromisoformat(last_updated.start_value) - timedelta(days=7)
    ).isoformat()

    query = f"""
    SELECT * FROM orders
    WHERE updated_at >= '{lookback_start}'
    ORDER BY updated_at
    """

    from sqlalchemy import create_engine
    engine = create_engine("postgresql://user:password@localhost:5432/db")

    with engine.connect() as conn:
        result = conn.execute(query)
        yield from result

pipeline = dlt.pipeline(
    pipeline_name="orders_lookback",
    destination="bigquery",
    dataset_name="sales"
)

load_info = pipeline.run([load_orders_with_lookback()])

Nested Incremental (Parent-Child)

Incremental Loading for Nested Resources

import dlt

@dlt.resource(
    name="accounts",
    write_disposition="merge",
    primary_key="account_id"
)
def load_accounts_incremental():
    last_updated = dlt.sources.incremental(
        cursor_path="updated_at",
        initial_value="2024-01-01"
    )

    accounts = fetch_accounts_since(last_updated.start_value)
    yield accounts

@dlt.resource(
    name="account_transactions",
    write_disposition="append",
    primary_key="transaction_id"
)
def load_transactions_for_accounts(accounts):
    last_transaction_date = dlt.sources.incremental(
        cursor_path="transaction_date",
        initial_value="2024-01-01"
    )

    for account in accounts:
        transactions = fetch_transactions(
            account_id=account["account_id"],
            since=last_transaction_date.start_value
        )
        yield transactions

pipeline = dlt.pipeline(
    pipeline_name="accounts_incremental",
    destination="postgres",
    dataset_name="finance"
)

# Load both resources
accounts_data = load_accounts_incremental()
load_info = pipeline.run([
    accounts_data,
    load_transactions_for_accounts(accounts_data)
])

State Management

Check Current State

import dlt

pipeline = dlt.pipeline(
    pipeline_name="my_pipeline",
    destination="duckdb",
    dataset_name="raw"
)

# View current incremental state
state = pipeline.state
print(state)

# Access specific resource state
orders_state = state.get("resources", {}).get("orders", {})
print(f"Last processed timestamp: {orders_state.get('incremental', {}).get('updated_at')}")

Reset State

# Drop state to force full refresh
pipeline.drop_state()

Partial Reset

# Reset state for specific resource
state = pipeline.state
if "resources" in state and "orders" in state["resources"]:
    del state["resources"]["orders"]
pipeline.sync_state()

Advanced Patterns

Multi-Field Incremental

import dlt

@dlt.resource(
    name="events",
    write_disposition="append"
)
def load_events_multi_field():
    # Track by multiple fields
    last_date = dlt.sources.incremental(
        cursor_path="event_date",
        initial_value="2024-01-01"
    )

    last_id = dlt.sources.incremental(
        cursor_path="event_id",
        initial_value=0
    )

    # Fetch using both filters
    events = fetch_events(
        date_gte=last_date.start_value,
        id_gt=last_id.start_value
    )

    yield events

Incremental with Deduplication

import dlt

@dlt.resource(
    name="user_events",
    write_disposition="merge",
    primary_key="event_id",
    merge_key=["user_id", "event_timestamp"]
)
def load_deduplicated_events():
    last_timestamp = dlt.sources.incremental(
        cursor_path="event_timestamp",
        initial_value="2024-01-01"
    )

    # Fetch events
    events = fetch_events_since(last_timestamp.start_value)

    # dlt will deduplicate based on merge_key
    yield events

pipeline = dlt.pipeline(
    pipeline_name="deduplicated_events",
    destination="snowflake",
    dataset_name="events"
)

load_info = pipeline.run([load_deduplicated_events()])

Monitoring Incremental Loads

import dlt

pipeline = dlt.pipeline(
    pipeline_name="monitored_incremental",
    destination="postgres",
    dataset_name="raw"
)

load_info = pipeline.run([load_orders_incremental()])

# Check incremental metadata
print(f"Load info: {load_info}")
print(f"Loaded packages: {len(load_info.load_packages)}")

# Query state
state = pipeline.state
orders_state = state.get("resources", {}).get("orders", {})
print(f"Last cursor value: {orders_state.get('incremental', {})}")

# Verify loaded data
with pipeline.sql_client() as client:
    result = client.execute_sql("""
        SELECT
            COUNT(*) as total_rows,
            MAX(updated_at) as max_timestamp
        FROM orders
    """)
    print(f"Total rows: {result[0][0]}, Latest timestamp: {result[0][1]}")

Best Practices

• BEST: Use write_disposition="merge" for updates/deletes
• BEST: Use write_disposition="append" for immutable events
• BEST: Add lookback windows for late-arriving data
• BEST: Always specify primary_key for merge operations
• BEST: Use indexed timestamp columns at source for fast queries
• BEST: Monitor state to detect stalled pipelines
• BEST: Test full refresh vs incremental results
• BEST: Handle edge cases (timezone conversions, null timestamps)
• BEST: Use ORDER BY on cursor field for consistent results
• BEST: Implement alerts for cursor value staleness

dlt Pipeline Setup Template

Purpose: Set up data loading pipelines with dlt (data load tool) for ELT workflows.

Installation

pip install dlt[postgres]  # For Postgres destination
pip install dlt[snowflake]  # For Snowflake
pip install dlt[bigquery]   # For BigQuery
pip install dlt[duckdb]     # For DuckDB

Project Structure

my_pipeline/
├── .dlt/
│   ├── config.toml       # Configuration
│   └── secrets.toml      # Credentials (gitignored)
├── pipelines/
│   ├── github_pipeline.py
│   └── stripe_pipeline.py
└── requirements.txt

Basic Pipeline Example

import dlt
from dlt.sources.rest_api import rest_api_source

# Define pipeline
pipeline = dlt.pipeline(
    pipeline_name="github_data",
    destination="duckdb",
    dataset_name="github_raw"
)

# Load data
source = rest_api_source({
    "client": {
        "base_url": "https://api.github.com/repos/dlt-hub/dlt/"
    },
    "resources": ["issues", "pulls"]
})

load_info = pipeline.run(source)
print(load_info)

Configuration (.dlt/config.toml)

[sources.github]
owner = "dlt-hub"
repo = "dlt"

[destination.postgres]
credentials = "postgres://user:password@localhost:5432/db"

[destination.snowflake]
database = "ANALYTICS"
schema = "RAW_DATA"

Secrets (.dlt/secrets.toml)

[sources.github.credentials]
access_token = "ghp_your_token_here"

[destination.postgres.credentials]
database = "analytics"
username = "etl_user"
password = "your_password"
host = "localhost"
port = 5432

Run Pipeline

if __name__ == "__main__":
    load_info = pipeline.run(source)
    
    # Check for errors
    print(f"Load info: {load_info}")
    
    # Query loaded data
    with pipeline.sql_client() as client:
        result = client.execute_sql("SELECT COUNT(*) FROM issues")
        print(f"Loaded {result[0][0]} issues")

Best Practices

• BEST: Use .dlt/secrets.toml for credentials (never commit)
• BEST: Enable schema evolution for dynamic APIs
• BEST: Use incremental loading for large datasets
• BEST: Add error handling and retries
• BEST: Monitor pipeline runs with logging

StarRocks Setup Template

Overview

Setting up StarRocks for high-performance analytics with external catalog support.

Docker Compose

version: '3.8'

services:
  fe:
    image: starrocks/fe-ubuntu:3.2-latest
    hostname: fe
    ports:
      - "8030:8030"   # HTTP
      - "9030:9030"   # MySQL
    volumes:
      - fe_meta:/opt/starrocks/fe/meta

  be:
    image: starrocks/be-ubuntu:3.2-latest
    hostname: be
    depends_on:
      - fe
    volumes:
      - be_storage:/opt/starrocks/be/storage
    environment:
      FE_HOST: fe

volumes:
  fe_meta:
  be_storage:

---

Table Creation

Primary Key Table (Recommended)

CREATE TABLE events (
    event_id BIGINT,
    user_id BIGINT,
    event_type VARCHAR(50),
    properties JSON,
    created_at DATETIME
)
PRIMARY KEY (event_id)
DISTRIBUTED BY HASH(event_id) BUCKETS 16
PROPERTIES (
    "replication_num" = "3",
    "enable_persistent_index" = "true"
);

Duplicate Key Table

CREATE TABLE event_logs (
    event_id BIGINT,
    user_id BIGINT,
    event_type VARCHAR(50),
    created_at DATETIME
)
DUPLICATE KEY(event_id, user_id)
PARTITION BY RANGE(created_at) (
    PARTITION p202406 VALUES LESS THAN ("2024-07-01")
)
DISTRIBUTED BY HASH(user_id) BUCKETS 16;

---

External Catalogs

Iceberg Catalog

CREATE EXTERNAL CATALOG iceberg_catalog
PROPERTIES (
    "type" = "iceberg",
    "iceberg.catalog.type" = "rest",
    "iceberg.catalog.uri" = "http://iceberg-rest:8181",
    "aws.s3.access_key" = "${AWS_ACCESS_KEY}",
    "aws.s3.secret_key" = "${AWS_SECRET_KEY}",
    "aws.s3.region" = "us-east-1"
);

-- Query Iceberg table
SELECT * FROM iceberg_catalog.db.events
WHERE created_at >= '2024-01-01';

Hive Catalog

CREATE EXTERNAL CATALOG hive_catalog
PROPERTIES (
    "type" = "hive",
    "hive.metastore.uris" = "thrift://hive-metastore:9083"
);

Delta Lake Catalog

CREATE EXTERNAL CATALOG delta_catalog
PROPERTIES (
    "type" = "deltalake",
    "hive.metastore.uris" = "thrift://hive-metastore:9083"
);

---

Materialized Views

Async Refresh MV

-- Create MV on external data
CREATE MATERIALIZED VIEW mv_daily_events
DISTRIBUTED BY HASH(date) BUCKETS 8
REFRESH ASYNC START('2024-01-01 00:00:00') EVERY (INTERVAL 1 HOUR)
AS SELECT
    DATE(created_at) AS date,
    event_type,
    COUNT(*) AS events,
    COUNT(DISTINCT user_id) AS users
FROM iceberg_catalog.db.events
GROUP BY DATE(created_at), event_type;

-- Manual refresh
REFRESH MATERIALIZED VIEW mv_daily_events;

---

Data Loading

Stream Load

curl --location-trusted -u admin: \
    -H "label:events_load" \
    -H "column_separator:," \
    -T events.csv \
    http://fe:8030/api/db/events/_stream_load

Routine Load (Kafka)

CREATE ROUTINE LOAD db.events_load ON events
COLUMNS(event_id, user_id, event_type, properties, created_at)
PROPERTIES (
    "desired_concurrent_number" = "3",
    "format" = "json",
    "jsonpaths" = "[\"$.event_id\",\"$.user_id\",\"$.event_type\",\"$.properties\",\"$.created_at\"]"
)
FROM KAFKA (
    "kafka_broker_list" = "kafka:9092",
    "kafka_topic" = "events"
);

---

Query Optimization

Query Cache

-- Enable query cache
SET enable_query_cache = true;
SET query_cache_entry_max_bytes = 1048576;
SET query_cache_entry_max_rows = 10000;

Query Profile

-- Enable profiling
SET enable_profile = true;

-- View profile
SHOW PROFILELIST;
ANALYZE PROFILE FROM 'profile_id';

---

Best Practices

1. Use Primary Key - For real-time upserts 2. External catalogs - Query Iceberg/Delta directly 3. Async MVs - Pre-aggregate external data 4. Enable query cache - For repeated queries 5. Use persistent index - For Primary Key tables

Dashboard/Question Request Template

Requestor

• Team/owner:
• Stakeholders:
• Due date / cadence:

Goal

• Business question:
• Primary KPI / threshold:
• Decisions enabled:

Data

• Source tables/models:
• Grain (day/week/month):
• Filters/segments:
• Timezone:
• PII present? (yes/no; masking plan)

Outputs

• Charts/tables needed:
• Slices (dimensions):
• Alerts/subscriptions (channel + cadence):
• Export format (CSV/XLS/Embed):

Acceptance Criteria

• [ ] Loads in < X seconds
• [ ] Filters working and documented
• [ ] Definitions documented on dashboard
• [ ] Permissions set (collections, row-level filters)