Performance Optimization: Tuning Data Pipelines and Queries

Performance Optimization: Making Data Systems Fast

Performance optimization systematically identifies and eliminates bottlenecks in data pipelines and queries to reduce latency, increase throughput, and minimize resource consumption.

Why Performance Optimization Matters

Cost Impact:

• A query that takes 10 minutes costs 10x more than one that takes 1 minute
• At scale, performance optimization directly impacts cost

User Impact:

• Slow queries waste user time
• Poor performance leads to frustration

Key Insight: At scale, performance optimization directly impacts cost and user satisfaction.

Architecture Overview

The Performance Optimization Framework has five layers: Layer 1 - Query Optimization (partition pruning with 10-100x speedup, indexing with B-tree/Bitmap/Bloom, materialized views), Layer 2 - Pipeline Optimization (parallelism with Airflow TaskGroups, incremental processing, batching with bulk operations), Layer 3 - Infrastructure (right-sizing for workload match, auto-scaling with elastic compute, spot instances for 70% savings), Layer 4 - Storage (columnar formats, compression with Snappy/Zstd, file size 128-256 MB target), and Layer 5 - Monitoring (query profile, metrics, alerting for SLA violations).

Query Optimization

-- BAD: Full table scan
SELECT *
FROM fact_orders
WHERE customer_id = 12345;

-- GOOD: Partition pruning + selective columns
SELECT order_id, order_date, total_amount
FROM fact_orders
WHERE order_date >= '2025-01-01'
  AND order_date < '2025-02-01'
  AND customer_id = 12345;

-- Analyze query performance
EXPLAIN ANALYZE
SELECT
    c.customer_name,
    SUM(f.net_amount) AS total_spent
FROM fact_orders f
JOIN dim_customer c ON f.customer_key = c.customer_key
WHERE f.order_date BETWEEN '2025-01-01' AND '2025-12-31'
GROUP BY c.customer_name
ORDER BY total_spent DESC
LIMIT 10;

-- Snowflake: Query profile
SELECT *
FROM TABLE(GET_QUERY_OPERATOR_STATS('query_id_here'));

-- Redshift: Analyze query plan
EXPLAIN
SELECT * FROM fact_orders
WHERE order_date = '2025-01-15' AND customer_id = 12345;

-- Materialized View for repeated aggregations
CREATE MATERIALIZED VIEW mv_daily_revenue AS
SELECT
    order_date,
    SUM(net_amount) AS daily_revenue,
    COUNT(*) AS order_count
FROM fact_orders
GROUP BY order_date;

-- Refresh materialized view
ALTER MATERIALIZED VIEW mv_daily_revenue REFRESH;

-- Partition-specific query
SELECT * FROM fact_orders
WHERE order_date = '2025-01-15'  -- Prunes to single partition
  AND customer_id = 12345;       -- Uses index within partition

Pipeline Optimization

# Parallel Pipeline Execution with Airflow
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime, timedelta
import concurrent.futures

default_args = {
    'owner': 'data-engineering',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

def extract_data(source_name: str, **context):
    """Extract data from a source."""
    print(f"Extracting data from {source_name}")
    # Simulate extraction
    return {"source": source_name, "rows": 10000}

def transform_data(data: dict, transform_name: str, **context):
    """Apply transformation."""
    print(f"Applying {transform_name} to {data['source']}")
    return {**data, "transformed": True}

def load_data(data: dict, target: str, **context):
    """Load data into target."""
    print(f"Loading {data['source']} into {target}")

with DAG(
    dag_id='parallel_pipeline',
    default_args=default_args,
    schedule_interval='0 6 * * *',
    start_date=datetime(2025, 1, 1),
) as dag:

    # Independent extractions (run in parallel)
    with TaskGroup("extract") as extract_group:
        extract_shopify = PythonOperator(
            task_id='extract_shopify',
            python_callable=extract_data,
            op_kwargs={'source_name': 'shopify'},
        )
        extract_stripe = PythonOperator(
            task_id='extract_stripe',
            python_callable=extract_data,
            op_kwargs={'source_name': 'stripe'},
        )
        extract_segment = PythonOperator(
            task_id='extract_segment',
            python_callable=extract_data,
            op_kwargs={'source_name': 'segment'},
        )

    # Transformations (some parallel, some sequential)
    with TaskGroup("transform") as transform_group:
        transform_orders = PythonOperator(
            task_id='transform_orders',
            python_callable=transform_data,
            op_kwargs={'transform_name': 'orders'},
        )
        transform_payments = PythonOperator(
            task_id='transform_payments',
            python_callable=transform_data,
            op_kwargs={'transform_name': 'payments'},
        )
        transform_events = PythonOperator(
            task_id='transform_events',
            python_callable=transform_data,
            op_kwargs={'transform_name': 'events'},
        )

    # Load (parallel)
    with TaskGroup("load") as load_group:
        load_dim = PythonOperator(
            task_id='load_dimensions',
            python_callable=load_data,
            op_kwargs={'target': 'dim_tables'},
        )
        load_fact = PythonOperator(
            task_id='load_facts',
            python_callable=load_data,
            op_kwargs={'target': 'fact_tables'},
        )

    # Dependencies
    [extract_shopify, extract_stripe, extract_segment] >> \
    [transform_orders, transform_payments, transform_events] >> \
    [load_dim, load_fact]

Caching Strategies

-- Snowflake: Result caching (automatic)
-- Identical queries within 24 hours return cached results
SELECT * FROM fact_orders WHERE order_date = '2025-01-15';

-- Materialized View for caching aggregations
CREATE MATERIALIZED VIEW mv_customer_metrics AS
SELECT
    customer_key,
    COUNT(*) AS order_count,
    SUM(net_amount) AS lifetime_value,
    AVG(net_amount) AS avg_order_value
FROM fact_orders
GROUP BY customer_key;

-- BigQuery: BI Engine for in-memory caching
-- Reserve BI Engine capacity via Console

-- Redshift: Result caching
-- Automatic for identical queries

-- Application-level caching with Redis
import redis
import json

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_cached_query(query_hash: str):
    """Get cached query result."""
    cached = redis_client.get(f"query:{query_hash}")
    if cached:
        return json.loads(cached)
    return None

def cache_query_result(query_hash: str, result: list, ttl: int = 3600):
    """Cache query result with TTL."""
    redis_client.setex(
        f"query:{query_hash}",
        ttl,
        json.dumps(result)
    )

Performance Profiling Tools

-- Snowflake: Query profile analysis
SELECT
    query_id,
    query_text,
    execution_status,
    ROUND(bytes_scanned / 1024 / 1024 / 1024, 2) AS gb_scanned,
    partitions_scanned,
    partitions_total,
    ROUND(partitions_scanned * 100.0 / partitions_total, 2) AS scan_pct,
    compilation_time_ms,
    execution_time_ms,
    total_execution_time_ms,
    ROUND(credits_used, 4) AS credits,
    result_cache_hit
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE start_time >= DATEADD(hour, -1, CURRENT_DATE())
ORDER BY execution_time_ms DESC
LIMIT 10;

-- BigQuery: Slot utilization analysis
SELECT
    job_id,
    query,
    TIMESTAMP_DIFF(end_time, start_time, MILLISECOND) AS duration_ms,
    total_slot_ms / TIMESTAMP_DIFF(end_time, start_time, MILLISECOND) AS avg_slots,
    total_bytes_processed / 1024 / 1024 / 1024 AS data_gb,
    statement_type
FROM `region-us`.INFORMATION_SCHEMA.JOBS
WHERE creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
ORDER BY total_slot_ms DESC
LIMIT 10;

-- Spark: Analyze shuffle and spill
EXPLAIN EXTENDED
SELECT
    customer_id,
    SUM(amount) AS total_spent
FROM fact_orders
GROUP BY customer_id
ORDER BY total_spent DESC;

Key Concepts Summary

Performance Metrics

10 Best Practices

1. Profile queries before optimizing — identify the actual bottleneck before fixing
2. Use EXPLAIN plans — understand how the query engine executes your queries
3. Partition by date — the single most impactful optimization for time-series data
4. Materialize repeated aggregations — dashboards should query materialized views
5. Implement incremental processing — process only new/changed data, not full rebuilds
6. Cache expensive computations — use result caching, materialized views, or Redis
7. Parallelize independent stages — use Airflow TaskGroups for parallel execution
8. Compress data aggressively — columnar formats (Parquet) with Snappy compression
9. Right-size compute resources — monitor utilization and adjust weekly
10. Set performance budgets — define latency targets and alert on violations

See Also

• Partitioning & Indexing — Range, hash, and bloom filter strategies
• Cost Optimization — Right-sizing and auto-suspend for cost savings
• Snowflake Fundamentals — Micro-partition pruning and clustering keys
• BigQuery Fundamentals — BI Engine and materialized views
• dbt Advanced — Model build optimization and ephemeral materialization
• Interview Prep — Performance optimization interview questions