Transforming Data

Data Transformation

Transform raw data into analytical assets using modern transformation patterns, frameworks, and orchestration tools.

Purpose

Select and implement data transformation patterns across the modern data stack. Transform raw data into clean, tested, and documented analytical datasets using SQL (dbt), Python DataFrames (pandas, polars, PySpark), and pipeline orchestration (Airflow, Dagster, Prefect).

When to Use

Invoke this skill when:

• Choosing between ETL and ELT transformation patterns
• Building dbt models (staging, intermediate, marts)
• Implementing incremental data loads and merge strategies
• Migrating pandas code to polars for performance improvements
• Orchestrating data pipelines with dependencies and retries
• Adding data quality tests and validation
• Processing large datasets with PySpark
• Creating production-ready transformation workflows

Quick Start: Common Patterns

dbt Incremental Model

{{
  config(
    materialized='incremental',
    unique_key='order_id'
  )
}}

select order_id, customer_id, order_created_at, sum(revenue) as total_revenue
from {{ ref('int_order_items_joined') }}
group by 1, 2, 3

{% if is_incremental() %}
    where order_created_at > (select max(order_created_at) from {{ this }})
{% endif %}

polars High-Performance Transformation

import polars as pl

result = (
    pl.scan_csv('large_dataset.csv')
    .filter(pl.col('year') == 2024)
    .with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
    .group_by('region')
    .agg(pl.col('revenue').sum())
    .collect()  # Execute lazy query
)

Airflow Data Pipeline

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

with DAG(
    dag_id='daily_sales_pipeline',
    schedule_interval='0 2 * * *',
    default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)},
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    extract = PythonOperator(task_id='extract', python_callable=extract_data)
    transform = PythonOperator(task_id='transform', python_callable=transform_data)
    extract >> transform

Decision Frameworks

ETL vs ELT Selection

Use ELT (Extract, Load, Transform) when:

• Using modern cloud data warehouse (Snowflake, BigQuery, Databricks)
• Transformation logic changes frequently
• Team includes SQL analysts
• Data volume 10GB-1TB+ (leverage warehouse parallelism)

Tools: dbt, Dataform, Snowflake tasks, BigQuery scheduled queries

Use ETL (Extract, Transform, Load) when:

• Regulatory compliance requires pre-load data redaction (PII/PHI)
• Target system lacks compute power
• Real-time streaming with immediate transformation
• Legacy systems without cloud warehouse

Tools: AWS Glue, Azure Data Factory, custom Python scripts

Use Hybrid when combining sensitive data cleansing (ETL) with analytics transformations (ELT).

Default recommendation: ELT with dbt unless specific compliance or performance constraints require ETL.

For detailed patterns, see references/etl-vs-elt-patterns.md.

DataFrame Library Selection

Choose pandas when:

• Data size < 500MB
• Prototyping or exploratory analysis
• Need compatibility with pandas-only libraries

Choose polars when:

• Data size 500MB-100GB
• Performance critical (10-100x faster than pandas)
• Production pipelines with memory constraints
• Want lazy evaluation with query optimization

Choose PySpark when:

• Data size > 100GB
• Need distributed processing across cluster
• Existing Spark infrastructure (EMR, Databricks)

Migration path: pandas → polars (easier, similar API) or pandas → PySpark (requires cluster)

For comparisons and migration guides, see references/dataframe-comparison.md.

Orchestration Tool Selection

Choose Airflow when:

• Enterprise production (proven at scale)
• Need 5,000+ integrations
• Managed services available (AWS MWAA, GCP Cloud Composer)

Choose Dagster when:

• Heavy dbt usage (native dbt_assets integration)
• Data lineage and asset-based workflows prioritized
• ML pipelines requiring testability

Choose Prefect when:

• Dynamic workflows (runtime task generation)
• Cloud-native architecture preferred
• Pythonic API with decorators

Safe default: Airflow (battle-tested) unless specific needs for Dagster/Prefect.

For detailed patterns, see references/orchestration-patterns.md.

SQL Transformations with dbt

Model Layer Structure

1. Staging Layer (models/staging/)

• 1:1 with source tables
• Minimal transformations (renaming, type casting, basic filtering)
• Materialized as views or ephemeral

2. Intermediate Layer (models/intermediate/)

• Business logic and complex joins
• Not exposed to end users
• Often ephemeral (CTEs only)

3. Marts Layer (models/marts/)

• Final models for reporting
• Fact tables (events, transactions)
• Dimension tables (customers, products)
• Materialized as tables or incremental

dbt Materialization Types

View: Query re-run each time model referenced. Use for fast queries, staging layer.

Table: Full refresh on each run. Use for frequently queried models, expensive computations.

Incremental: Only processes new/changed records. Use for large fact tables, event logs.

Ephemeral: CTE only, not persisted. Use for intermediate calculations.

dbt Testing

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id
      - name: total_revenue
        tests:
          - dbt_utils.accepted_range:
              min_value: 0

For comprehensive dbt patterns, see:

• references/dbt-best-practices.md
• references/incremental-strategies.md

Python DataFrame Transformations

pandas Transformation

import pandas as pd

df = pd.read_csv('sales.csv')
result = (
    df
    .query('year == 2024')
    .assign(revenue=lambda x: x['quantity'] * x['price'])
    .groupby('region')
    .agg({'revenue': ['sum', 'mean']})
)

polars Transformation (10-100x Faster)

import polars as pl

result = (
    pl.scan_csv('sales.csv')  # Lazy evaluation
    .filter(pl.col('year') == 2024)
    .with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
    .group_by('region')
    .agg([
        pl.col('revenue').sum().alias('revenue_sum'),
        pl.col('revenue').mean().alias('revenue_mean')
    ])
    .collect()  # Execute lazy query
)

Key differences:

• polars uses scan_csv() (lazy) vs pandas read_csv() (eager)
• polars uses with_columns() vs pandas assign()
• polars uses pl.col() expressions vs pandas string references
• polars requires collect() to execute lazy queries

PySpark for Distributed Processing

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("Transform").getOrCreate()
df = spark.read.csv('sales.csv', header=True, inferSchema=True)

result = (
    df
    .filter(F.col('year') == 2024)
    .withColumn('revenue', F.col('quantity') * F.col('price'))
    .groupBy('region')
    .agg(F.sum('revenue').alias('total_revenue'))
)

For migration guides, see references/dataframe-comparison.md.

Pipeline Orchestration

Airflow DAG Structure

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

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

with DAG(
    dag_id='data_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # Daily at 2 AM
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    task1 = PythonOperator(task_id='extract', python_callable=extract_fn)
    task2 = PythonOperator(task_id='transform', python_callable=transform_fn)
    task1 >> task2  # Define dependency

Task Dependency Patterns

Linear: A >> B >> C (sequential) Fan-out: A >> [B, C, D] (parallel after A) Fan-in: [A, B, C] >> D (D waits for all)

For Airflow, Dagster, and Prefect patterns, see references/orchestration-patterns.md.

Data Quality and Testing

dbt Tests

Generic tests (reusable): unique, not_null, accepted_values, relationships

Singular tests (custom SQL):

-- tests/assert_positive_revenue.sql
select * from {{ ref('fct_orders') }}
where total_revenue < 0

Great Expectations

import great_expectations as gx

context = gx.get_context()
suite = context.add_expectation_suite("orders_suite")

suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="total_revenue", min_value=0
    )
)

For comprehensive testing patterns, see references/data-quality-testing.md.

Advanced SQL Patterns

Window functions for analytics:

select
    order_date,
    daily_revenue,
    avg(daily_revenue) over (
        partition by region
        order by order_date
        rows between 6 preceding and current row
    ) as revenue_7d_ma,
    sum(daily_revenue) over (
        partition by region
        order by order_date
    ) as cumulative_revenue
from daily_sales

For advanced window functions, see references/window-functions-guide.md.

Production Best Practices

Idempotency

Ensure transformations produce same result when run multiple times:

• Use merge statements in incremental models
• Implement deduplication logic
• Use unique_key in dbt incremental models

Incremental Loading

{% if is_incremental() %}
    where created_at > (select max(created_at) from {{ this }})
{% endif %}

Error Handling

try:
    result = perform_transformation()
    validate_result(result)
except ValidationError as e:
    log_error(e)
    raise

Monitoring

• Set up Airflow email/Slack alerts on task failure
• Monitor dbt test failures
• Track data freshness (SLAs)
• Log row counts and data quality metrics

Tool Recommendations

SQL Transformations: dbt Core (industry standard, multi-warehouse, rich ecosystem)

pip install dbt-core dbt-snowflake

Python DataFrames: polars (10-100x faster than pandas, multi-threaded, lazy evaluation)

pip install polars

Orchestration: Apache Airflow (battle-tested at scale, 5,000+ integrations)

pip install apache-airflow

Examples

Working examples in:

• examples/python/pandas-basics.py - pandas transformations
• examples/python/polars-migration.py - pandas to polars migration
• examples/python/pyspark-transformations.py - PySpark operations
• examples/python/airflow-data-pipeline.py - Complete Airflow DAG
• examples/sql/dbt-staging-model.sql - dbt staging layer
• examples/sql/dbt-intermediate-model.sql - dbt intermediate layer
• examples/sql/dbt-incremental-model.sql - Incremental patterns
• examples/sql/window-functions.sql - Advanced SQL

Scripts

• scripts/generate_dbt_models.py - Generate dbt model boilerplate
• scripts/benchmark_dataframes.py - Compare pandas vs polars performance

Related Skills

For data ingestion patterns, see ingesting-data. For data visualization, see visualizing-data. For database design, see databases-* skills. For real-time streaming, see streaming-data. For data platform architecture, see ai-data-engineering. For monitoring pipelines, see observability.

Data Quality Testing

Table of Contents

• Overview
• dbt Testing
• Generic Tests
• Singular Tests
• Custom Generic Tests (Macros)
• Great Expectations
• Setup
• Common Expectations
• Run Validation
• Integrating Tests into Pipelines
• Airflow Integration
• Best Practices
• Test Severity Levels
• Common Quality Checks

Overview

Data quality testing ensures transformations produce accurate, complete, and consistent data before consumption by downstream systems.

dbt Testing

Generic Tests

Built-in tests for common validations.

# models/marts/schema.yml

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null

      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id

      - name: order_status
        tests:
          - accepted_values:
              values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled']

Singular Tests

Custom SQL tests for complex validations.

-- tests/assert_positive_revenue.sql

select *
from {{ ref('fct_orders') }}
where total_revenue < 0

Test fails if any rows are returned.

Custom Generic Tests (Macros)

-- macros/test_row_count_min.sql

{% test row_count_min(model, min_rows=1) %}

select count(*) as row_count
from {{ model }}
having count(*) < {{ min_rows }}

{% endtest %}

Usage:

models:
  - name: fct_orders
    tests:
      - row_count_min:
          min_rows: 100

Great Expectations

Setup

import great_expectations as gx

context = gx.get_context()

# Create expectation suite
suite = context.add_expectation_suite(
    expectation_suite_name="orders_suite"
)

Common Expectations

# Column exists
suite.add_expectation(
    gx.expectations.ExpectColumnToExist(column="order_id")
)

# No nulls
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
)

# Values in range
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="total_amount",
        min_value=0,
        max_value=100000
    )
)

# Unique values
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="order_id")
)

# Values in set
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeInSet(
        column="order_status",
        value_set=['pending', 'confirmed', 'shipped', 'delivered']
    )
)

Run Validation

# Validate DataFrame
validator = context.sources.pandas_default.read_dataframe(df)
results = validator.validate(expectation_suite_name="orders_suite")

if not results.success:
    raise ValueError("Data quality checks failed")

Integrating Tests into Pipelines

Airflow Integration

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator

def run_quality_checks(**context):
    # Great Expectations validation
    context_ge = gx.get_context()
    validator = context_ge.sources.pandas_default.read_dataframe(df)
    results = validator.validate(expectation_suite_name="orders_suite")

    if not results.success:
        raise ValueError("Quality checks failed")

with DAG('data_pipeline', ...) as dag:
    extract = PythonOperator(task_id='extract', ...)
    transform = DbtCloudRunJobOperator(task_id='dbt_run', ...)
    test = DbtCloudRunJobOperator(task_id='dbt_test', ...)  # Run dbt tests
    quality_check = PythonOperator(task_id='quality_check', python_callable=run_quality_checks)

    extract >> transform >> test >> quality_check

Best Practices

1. Test early: Validate in staging layer before expensive transformations 2. Test incrementally: Run tests on each dbt run 3. Fail fast: Stop pipeline on critical test failures 4. Monitor trends: Track test pass rates over time 5. Document expectations: Explain why tests exist

Test Severity Levels

# dbt: Warn vs Error
models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique:
              severity: error  # Fail pipeline
          - not_null:
              severity: warn   # Log but don't fail

Common Quality Checks

1. Completeness: No missing required fields 2. Uniqueness: Primary keys are unique 3. Consistency: Values match reference data 4. Accuracy: Aggregations match source totals 5. Timeliness: Data freshness within SLA 6. Validity: Values in expected ranges/formats

-- Freshness check
select max(updated_at) as last_update
from {{ ref('fct_orders') }}
having max(updated_at) < current_timestamp - interval '24' hour

DataFrame Library Comparison: pandas vs polars vs PySpark

Table of Contents

1. Quick Comparison 2. pandas Deep Dive 3. polars Deep Dive 4. PySpark Deep Dive 5. Migration Guides 6. Performance Benchmarks 7. When to Use Which

---

Quick Comparison

---

pandas Deep Dive

Strengths

1. Mature Ecosystem: 15+ years of development, massive community 2. Rich Functionality: 2,000+ methods and functions 3. Extensive Documentation: Tutorials, Stack Overflow answers everywhere 4. Library Compatibility: Works with scikit-learn, matplotlib, seaborn, etc.

Weaknesses

1. Single-threaded: Cannot use multiple CPU cores 2. Memory Inefficient: Entire dataset must fit in RAM 3. Slow at Scale: Performance degrades significantly beyond 1GB 4. Inconsistent API: Multiple ways to do same thing

Common Operations

Reading Data

import pandas as pd

# CSV
df = pd.read_csv('data.csv')

# Parquet
df = pd.read_parquet('data.parquet')

# SQL
df = pd.read_sql('SELECT * FROM orders', connection)

# Excel
df = pd.read_excel('data.xlsx', sheet_name='Sheet1')

Filtering

# Simple filter
df_filtered = df[df['age'] > 25]

# Multiple conditions
df_filtered = df[(df['age'] > 25) & (df['city'] == 'NYC')]

# Query method (SQL-like)
df_filtered = df.query('age > 25 and city == "NYC"')

Grouping and Aggregation

# Group by single column
result = df.groupby('region')['sales'].sum()

# Group by multiple columns
result = df.groupby(['region', 'product'])['sales'].agg(['sum', 'mean', 'count'])

# Named aggregations
result = df.groupby('region').agg(
    total_sales=('sales', 'sum'),
    avg_sales=('sales', 'mean'),
    order_count=('order_id', 'nunique')
)

Transformations

# Add calculated columns
df['revenue'] = df['quantity'] * df['price']

# Apply function
df['revenue_category'] = df['revenue'].apply(
    lambda x: 'high' if x > 1000 else 'low'
)

# Assign (method chaining)
df = (
    df
    .assign(revenue=lambda x: x['quantity'] * x['price'])
    .assign(revenue_category=lambda x: x['revenue'].apply(categorize))
)

pandas Performance Tips

1. Use vectorized operations (avoid .apply() when possible) 2. Use categorical dtype for low-cardinality columns 3. Read only needed columns (usecols parameter) 4. Use chunking for large files (chunksize parameter) 5. Consider `dtype` optimization (int32 vs int64, etc.)

---

polars Deep Dive

Strengths

1. Blazingly Fast: 10-100x faster than pandas 2. Multi-threaded: Automatic parallelization across CPU cores 3. Lazy Evaluation: Query optimization before execution 4. Memory Efficient: Streaming and minimal copies 5. Modern API: Consistent, expression-based interface

Weaknesses

1. Smaller Ecosystem: Fewer integrations than pandas (but growing fast) 2. Breaking Changes: Rapid development means occasional API changes 3. Less StackOverflow: Smaller community (though docs are excellent)

Common Operations

Reading Data

import polars as pl

# Eager (load immediately)
df = pl.read_csv('data.csv')
df = pl.read_parquet('data.parquet')

# Lazy (for query optimization)
df = pl.scan_csv('data.csv')  # Returns LazyFrame
df = pl.scan_parquet('data.parquet')

Filtering

# Eager
df_filtered = df.filter(pl.col('age') > 25)

# Multiple conditions
df_filtered = df.filter(
    (pl.col('age') > 25) & (pl.col('city') == 'NYC')
)

# Lazy (then collect)
result = (
    pl.scan_csv('data.csv')
    .filter(pl.col('age') > 25)
    .collect()  # Execute lazy query
)

Grouping and Aggregation

# Group by single column
result = df.group_by('region').agg(pl.col('sales').sum())

# Group by multiple columns with multiple aggregations
result = df.group_by(['region', 'product']).agg([
    pl.col('sales').sum().alias('total_sales'),
    pl.col('sales').mean().alias('avg_sales'),
    pl.col('order_id').n_unique().alias('order_count')
])

Transformations

# Add calculated columns
df = df.with_columns([
    (pl.col('quantity') * pl.col('price')).alias('revenue')
])

# Multiple transformations
df = df.with_columns([
    (pl.col('quantity') * pl.col('price')).alias('revenue'),
    pl.when(pl.col('revenue') > 1000)
      .then(pl.lit('high'))
      .otherwise(pl.lit('low'))
      .alias('revenue_category')
])

Lazy Evaluation Pattern

# Build query plan (no execution yet)
lazy_df = (
    pl.scan_csv('large_file.csv')
    .filter(pl.col('year') == 2024)
    .with_columns([
        (pl.col('quantity') * pl.col('price')).alias('revenue')
    ])
    .group_by('region')
    .agg([
        pl.col('revenue').sum().alias('total_revenue'),
        pl.col('order_id').n_unique().alias('order_count')
    ])
    .sort('total_revenue', descending=True)
)

# Execute optimized query
result = lazy_df.collect()

# Or execute with streaming (memory efficient)
result = lazy_df.collect(streaming=True)

polars Performance Features

1. Automatic parallelization across CPU cores 2. SIMD (Single Instruction, Multiple Data) vectorization 3. Query optimization (predicate pushdown, projection pushdown) 4. Streaming execution for out-of-memory datasets 5. Zero-copy operations where possible

---

PySpark Deep Dive

Strengths

1. Distributed Computing: Scale to petabytes across cluster 2. Fault Tolerance: Automatic recovery from node failures 3. Ecosystem Integration: Works with Hadoop, Hive, Delta Lake 4. SQL Support: Can use SQL alongside DataFrame API

Weaknesses

1. Infrastructure Required: Needs cluster (EMR, Databricks, local) 2. Startup Overhead: Slower for small datasets 3. Debugging Difficulty: Distributed errors harder to trace 4. Learning Curve: More complex than pandas/polars

Common Operations

Reading Data

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("DataTransform").getOrCreate()

# CSV
df = spark.read.csv('s3://bucket/data.csv', header=True, inferSchema=True)

# Parquet
df = spark.read.parquet('s3://bucket/data.parquet')

# Multiple files
df = spark.read.parquet('s3://bucket/data/*.parquet')

Filtering

# Simple filter
df_filtered = df.filter(F.col('age') > 25)

# Multiple conditions
df_filtered = df.filter(
    (F.col('age') > 25) & (F.col('city') == 'NYC')
)

# SQL-style
df_filtered = df.filter("age > 25 AND city = 'NYC'")

Grouping and Aggregation

# Group by single column
result = df.groupBy('region').agg(F.sum('sales').alias('total_sales'))

# Group by multiple columns
result = df.groupBy('region', 'product').agg(
    F.sum('sales').alias('total_sales'),
    F.mean('sales').alias('avg_sales'),
    F.countDistinct('order_id').alias('order_count')
)

Transformations

# Add calculated columns
df = df.withColumn('revenue', F.col('quantity') * F.col('price'))

# Multiple transformations
df = (
    df
    .withColumn('revenue', F.col('quantity') * F.col('price'))
    .withColumn('revenue_category',
        F.when(F.col('revenue') > 1000, 'high').otherwise('low')
    )
)

PySpark Performance Tips

1. Partition data appropriately (avoid skew) 2. Use broadcast joins for small tables 3. Cache intermediate results when reused 4. Avoid shuffles when possible 5. Use Spark SQL for complex queries (query optimizer)

---

Migration Guides

pandas → polars

High compatibility: Most operations have direct equivalents

Example migration:

# Before (pandas)
import pandas as pd

df = pd.read_csv('sales.csv')
result = (
    df[df['year'] == 2024]
    .assign(revenue=lambda x: x['quantity'] * x['price'])
    .groupby('region')
    .agg({'revenue': ['sum', 'mean']})
)

# After (polars)
import polars as pl

df = pl.scan_csv('sales.csv')  # Lazy for optimization
result = (
    df
    .filter(pl.col('year') == 2024)
    .with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
    .group_by('region')
    .agg([
        pl.col('revenue').sum().alias('revenue_sum'),
        pl.col('revenue').mean().alias('revenue_mean')
    ])
    .collect()  # Execute lazy query
)

pandas → PySpark

Lower compatibility: Different paradigms (single-node vs distributed)

# Before (pandas)
import pandas as pd

df = pd.read_csv('sales.csv')
result = df[df['year'] == 2024].groupby('region')['sales'].sum()

# After (PySpark)
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.getOrCreate()
df = spark.read.csv('sales.csv', header=True, inferSchema=True)
result = (
    df
    .filter(F.col('year') == 2024)
    .groupBy('region')
    .agg(F.sum('sales').alias('total_sales'))
)

---

Performance Benchmarks

Benchmark Setup

Dataset: 10 million rows, 200MB CSV Task: Filter, calculate derived column, group by, aggregate Hardware: 16-core CPU, 32GB RAM

Results

Speed improvement:

• polars: 5.2x faster than pandas
• polars (lazy): 18.4x faster than pandas

Large Dataset (100GB)

---

When to Use Which

Choose pandas When:

• Data size < 500MB
• Prototyping and exploratory analysis
• Need scikit-learn integration
• Team is already pandas-expert
• Extensive library ecosystem required (e.g., specific ML libraries)

Choose polars When:

• Data size 500MB - 100GB
• Production pipelines with performance requirements
• Single-machine processing sufficient
• Want modern, clean API
• Memory efficiency is important

Choose PySpark When:

• Data size > 100GB
• Need distributed processing
• Existing Spark infrastructure (Databricks, EMR)
• Integration with Hadoop ecosystem
• Team has Spark expertise

Hybrid Approach

Many teams use multiple libraries:

# Use pandas for small exploratory work
import pandas as pd
df_sample = pd.read_csv('data.csv', nrows=10000)
df_sample.describe()

# Use polars for production pipeline
import polars as pl
df = pl.scan_csv('data.csv').filter(...).collect()

# Use PySpark for massive historical backfill
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.read.parquet('s3://bucket/historical/*.parquet')

---

Code Comparison: Same Task

Task: Read CSV, filter 2024 data, calculate revenue, group by region, get top 10

pandas

import pandas as pd

df = pd.read_csv('sales.csv')
result = (
    df
    .query('year == 2024')
    .assign(revenue=lambda x: x['quantity'] * x['price'])
    .groupby('region')
    .agg({'revenue': 'sum'})
    .sort_values('revenue', ascending=False)
    .head(10)
)

polars

import polars as pl

result = (
    pl.scan_csv('sales.csv')
    .filter(pl.col('year') == 2024)
    .with_columns([(pl.col('quantity') * pl.col('price')).alias('revenue')])
    .group_by('region')
    .agg(pl.col('revenue').sum())
    .sort('revenue', descending=True)
    .head(10)
    .collect()
)

PySpark

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.getOrCreate()
df = spark.read.csv('sales.csv', header=True, inferSchema=True)

result = (
    df
    .filter(F.col('year') == 2024)
    .withColumn('revenue', F.col('quantity') * F.col('price'))
    .groupBy('region')
    .agg(F.sum('revenue').alias('revenue'))
    .orderBy(F.desc('revenue'))
    .limit(10)
)

---

Additional Resources

• pandas documentation: https://pandas.pydata.org/docs/
• polars documentation: https://docs.pola.rs/
• PySpark documentation: https://spark.apache.org/docs/latest/api/python/
• polars migration guide: https://docs.pola.rs/user-guide/migration/pandas/
• Performance comparison: https://h2oai.github.io/db-benchmark/

dbt Best Practices

Table of Contents

1. Project Structure 2. Model Layering 3. Naming Conventions 4. Materialization Strategies 5. Testing Patterns 6. Documentation 7. Performance Optimization 8. Common Patterns

---

Project Structure

Recommended Directory Layout

my_dbt_project/
├── dbt_project.yml          # Project configuration
├── profiles.yml             # Connection profiles (gitignored)
├── packages.yml             # dbt packages (dbt-utils, etc.)
├── models/
│   ├── staging/             # Source staging models
│   │   ├── source.yml       # Source definitions
│   │   ├── stg_orders.sql
│   │   └── stg_customers.sql
│   ├── intermediate/        # Business logic
│   │   └── int_orders_joined.sql
│   ├── marts/               # Final analytics models
│   │   ├── schema.yml       # Tests and documentation
│   │   ├── fct_orders.sql   # Fact tables
│   │   └── dim_customers.sql # Dimension tables
│   └── _models.yml          # Model-level configs
├── macros/                  # Custom SQL macros
│   └── custom_tests.sql
├── tests/                   # Singular tests
│   └── assert_positive_revenue.sql
├── seeds/                   # CSV reference data
│   └── country_codes.csv
├── snapshots/               # SCD Type 2 snapshots
│   └── dim_customers_snapshot.sql
└── analyses/                # Ad-hoc analyses
    └── customer_analysis.sql

---

Model Layering

Three-Layer Architecture

Layer 1: Staging (models/staging/)

Purpose: Light touch on source data

Responsibilities:

• Rename columns for consistency
• Cast data types
• Basic filtering (remove test data, null IDs)
• 1:1 relationship with source tables

Materialization: View or Ephemeral

Example:

-- models/staging/stg_orders.sql

with source as (
    select * from {{ source('ecommerce', 'raw_orders') }}
),

renamed as (
    select
        -- IDs
        order_id,
        customer_id,

        -- Timestamps (standardize naming)
        created_at as order_created_at,
        updated_at as order_updated_at,

        -- Metrics (cast to correct types)
        cast(total_amount as decimal(18,2)) as order_amount,

        -- Dimensions (clean and standardize)
        lower(trim(status)) as order_status

    from source
    where order_id is not null
)

select * from renamed

Layer 2: Intermediate (models/intermediate/)

Purpose: Complex business logic

Responsibilities:

• Join multiple staging models
• Apply business rules
• Create reusable building blocks
• NOT exposed to end users

Materialization: Ephemeral (preferred) or View

Naming: Prefix with int_

Example:

-- models/intermediate/int_orders_joined.sql

with orders as (
    select * from {{ ref('stg_orders') }}
),

customers as (
    select * from {{ ref('stg_customers') }}
),

products as (
    select * from {{ ref('stg_products') }}
),

joined as (
    select
        o.order_id,
        o.customer_id,
        c.customer_email,
        c.customer_segment,
        o.order_created_at,
        o.order_status,
        p.product_name,
        p.category,
        o.order_amount,

        -- Business logic
        case
            when o.order_amount >= 1000 then 'high_value'
            when o.order_amount >= 500 then 'medium_value'
            else 'standard'
        end as order_value_tier

    from orders o
    left join customers c on o.customer_id = c.customer_id
    left join products p on o.product_id = p.product_id
)

select * from joined

Layer 3: Marts (models/marts/)

Purpose: Business-facing analytics models

Responsibilities:

• Fact tables (events, transactions)
• Dimension tables (customers, products)
• Aggregated metrics
• Wide denormalized tables for BI tools

Materialization: Table or Incremental

Naming: fct_ (facts), dim_ (dimensions)

Example:

-- models/marts/fct_orders.sql

with orders as (
    select * from {{ ref('int_orders_joined') }}
),

aggregated as (
    select
        order_id,
        customer_id,
        order_created_at,
        order_status,
        sum(order_amount) as total_order_amount,
        count(*) as line_item_count

    from orders
    group by 1, 2, 3, 4
)

select * from aggregated

---

Naming Conventions

Model Names

Staging models: stg_<source>_<entity>.sql

• Examples: stg_salesforce_accounts.sql, stg_stripe_payments.sql

Intermediate models: int_<entity>_<verb>.sql

• Examples: int_orders_joined.sql, int_customers_pivoted.sql

Fact tables: fct_<entity>.sql

• Examples: fct_orders.sql, fct_sessions.sql

Dimension tables: dim_<entity>.sql

• Examples: dim_customers.sql, dim_products.sql

Column Names

IDs: Suffix with _id

• Good: customer_id, order_id
• Bad: customer, order

Booleans: Prefix with is_ or has_

• Good: is_active, has_purchased
• Bad: active, purchased

Timestamps: Suffix with _at

• Good: created_at, updated_at
• Bad: creation_date, update_time

Dates: Suffix with _date

• Good: order_date, signup_date
• Bad: order_day, signup

Counts: Suffix with _count

• Good: order_count, session_count
• Bad: orders, sessions

---

Materialization Strategies

View

Use when:

• Staging models (always views or ephemeral)
• Rarely queried models
• Fast queries (<1 second)

Config:

{{
  config(
    materialized='view'
  )
}}

Pros: Always up-to-date, no storage cost Cons: Query runs every time model is referenced

Table

Use when:

• Frequently queried models (>10x/day)
• Expensive queries (>30 seconds)
• Dimension tables with full refresh

Config:

{{
  config(
    materialized='table'
  )
}}

Pros: Fast query performance Cons: Stale data between runs, storage cost

Incremental

Use when:

• Large fact tables (millions of rows)
• Append-only data (events, logs)
• Data updates frequently

Config:

{{
  config(
    materialized='incremental',
    unique_key='order_id',
    on_schema_change='fail'
  )
}}

select
    order_id,
    customer_id,
    order_created_at,
    total_amount
from {{ ref('stg_orders') }}

{% if is_incremental() %}
    -- Only process new records
    where order_created_at > (select max(order_created_at) from {{ this }})
{% endif %}

Pros: Fast runs, efficient storage Cons: More complex, risk of data drift

Ephemeral

Use when:

• Intermediate models used by only one downstream model
• Want to avoid persisting intermediate tables

Config:

{{
  config(
    materialized='ephemeral'
  )
}}

Pros: No storage, always up-to-date Cons: Cannot query directly, recomputed in every downstream model

---

Testing Patterns

Generic Tests

Built-in tests for common validations:

# models/marts/schema.yml

version: 2

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null

      - name: customer_id
        tests:
          - not_null
          - relationships:
              to: ref('dim_customers')
              field: customer_id

      - name: order_status
        tests:
          - accepted_values:
              values: ['pending', 'confirmed', 'shipped', 'delivered', 'cancelled']

      - name: total_amount
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              inclusive: true

Singular Tests

Custom SQL tests for complex validations:

-- tests/assert_order_amount_matches_items.sql

-- Test fails if any rows are returned
select
    order_id,
    sum(item_amount) as calculated_total,
    order_amount as reported_total
from {{ ref('fct_orders') }}
group by order_id, order_amount
having abs(sum(item_amount) - order_amount) > 0.01

Custom Generic Tests (Macros)

-- macros/test_is_recent.sql

{% test is_recent(model, column_name, days=7) %}

select *
from {{ model }}
where {{ column_name }} < current_date - interval '{{ days }}' day

{% endtest %}

Usage:

models:
  - name: fct_orders
    columns:
      - name: order_created_at
        tests:
          - is_recent:
              days: 30

---

Documentation

Model Documentation

# models/marts/schema.yml

models:
  - name: fct_orders
    description: "Order-level fact table with one row per order"
    columns:
      - name: order_id
        description: "Unique identifier for each order"
      - name: customer_id
        description: "Foreign key to dim_customers"
      - name: total_amount
        description: "Total order amount including tax and shipping"

Inline Documentation

-- models/marts/fct_orders.sql

{{
  config(
    materialized='incremental',
    unique_key='order_id'
  )
}}

-- This model calculates order-level metrics by aggregating line items.
-- It runs incrementally to process only new orders since the last run.

with orders as (
    select * from {{ ref('stg_orders') }}
),

-- Join order items to calculate totals
order_items as (
    select * from {{ ref('stg_order_items') }}
),

aggregated as (
    select
        o.order_id,
        o.customer_id,
        o.order_created_at,
        sum(oi.item_amount) as total_amount,
        count(*) as item_count
    from orders o
    inner join order_items oi on o.order_id = oi.order_id
    group by 1, 2, 3
)

select * from aggregated

{% if is_incremental() %}
    where order_created_at > (select max(order_created_at) from {{ this }})
{% endif %}

Generate Documentation

dbt docs generate
dbt docs serve

Access at http://localhost:8080

---

Performance Optimization

1. Use CTEs for Readability

-- Good: Clear, maintainable
with orders as (
    select * from {{ ref('stg_orders') }}
),

customers as (
    select * from {{ ref('stg_customers') }}
),

joined as (
    select
        o.order_id,
        c.customer_name,
        o.total_amount
    from orders o
    left join customers c on o.customer_id = c.customer_id
)

select * from joined

2. Limit Data Early

-- Good: Filter before joins
with orders as (
    select * from {{ ref('stg_orders') }}
    where order_status != 'cancelled'  -- Filter early
),

customers as (
    select * from {{ ref('stg_customers') }}
    where is_active = true  -- Filter early
),

joined as (
    select * from orders o
    left join customers c on o.customer_id = c.customer_id
)

select * from joined

3. Use Incremental Models for Large Tables

{{
  config(
    materialized='incremental',
    unique_key='event_id',
    partition_by={'field': 'event_date', 'data_type': 'date'}
  )
}}

select * from {{ ref('stg_events') }}

{% if is_incremental() %}
    where event_date > (select max(event_date) from {{ this }})
{% endif %}

4. Partition Large Tables (BigQuery, Snowflake)

{{
  config(
    materialized='table',
    partition_by={
      'field': 'order_date',
      'data_type': 'date',
      'granularity': 'day'
    },
    cluster_by=['customer_id', 'region']
  )
}}

select * from {{ ref('stg_orders') }}

---

Common Patterns

Pattern 1: Slowly Changing Dimensions (SCD Type 2)

-- snapshots/dim_customers_snapshot.sql

{% snapshot dim_customers_snapshot %}

{{
    config(
      target_schema='snapshots',
      unique_key='customer_id',
      strategy='timestamp',
      updated_at='updated_at'
    )
}}

select * from {{ ref('stg_customers') }}

{% endsnapshot %}

Run with: dbt snapshot

Pattern 2: Surrogate Keys

-- models/intermediate/int_orders_with_surrogate.sql

select
    {{ dbt_utils.generate_surrogate_key(['order_id', 'line_item_id']) }} as order_line_key,
    order_id,
    line_item_id,
    product_id,
    quantity
from {{ ref('stg_order_items') }}

Pattern 3: Pivot Tables

-- models/intermediate/int_sales_pivoted.sql

select
    customer_id,
    sum(case when product_category = 'Electronics' then sales_amount else 0 end) as electronics_sales,
    sum(case when product_category = 'Clothing' then sales_amount else 0 end) as clothing_sales,
    sum(case when product_category = 'Food' then sales_amount else 0 end) as food_sales
from {{ ref('stg_sales') }}
group by customer_id

Or use dbt_utils:

{{ dbt_utils.pivot(
    'product_category',
    dbt_utils.get_column_values(ref('stg_sales'), 'product_category'),
    agg='sum',
    cmp='=',
    prefix='',
    suffix='_sales',
    then_value='sales_amount'
) }}

Pattern 4: Deduplication

-- models/staging/stg_orders_deduplicated.sql

with source as (
    select * from {{ source('raw', 'orders') }}
),

deduplicated as (
    select *,
        row_number() over (
            partition by order_id
            order by updated_at desc
        ) as row_num
    from source
)

select * from deduplicated
where row_num = 1

---

Macros and Packages

Install Packages

# packages.yml

packages:
  - package: dbt-labs/dbt_utils
    version: 1.1.1
  - package: calogica/dbt_expectations
    version: 0.10.3

Install with: dbt deps

Common Macros

dbt_utils.star:

select
    {{ dbt_utils.star(from=ref('stg_orders'), except=['_loaded_at']) }},
    current_timestamp() as transformed_at
from {{ ref('stg_orders') }}

dbt_utils.union_relations:

{{ dbt_utils.union_relations(
    relations=[
        ref('stg_orders_2023'),
        ref('stg_orders_2024')
    ]
) }}

---

Additional Resources

• Official dbt Docs: https://docs.getdbt.com/
• dbt Style Guide: https://github.com/dbt-labs/corp/blob/main/dbt_style_guide.md
• dbt Utils Package: https://github.com/dbt-labs/dbt-utils
• dbt Discourse Community: https://discourse.getdbt.com/

ETL vs ELT Transformation Patterns

Table of Contents

1. Overview 2. ETL Pattern Deep Dive 3. ELT Pattern Deep Dive 4. Hybrid Approaches 5. Decision Matrix 6. Architecture Patterns 7. Performance Considerations 8. Cost Analysis

---

Overview

ETL (Extract, Transform, Load) and ELT (Extract, Load, Transform) represent fundamentally different approaches to data transformation in modern data architecture.

Historical Context

ETL Era (1990s-2010s):

• Limited warehouse compute power
• Expensive storage (optimize before loading)
• Transformation on dedicated servers
• Tools: Informatica, Talend, DataStage

ELT Era (2010s-present):

• Cloud warehouse revolution (Snowflake, BigQuery, Redshift)
• Elastic compute scaling
• Cheap storage (load everything, transform later)
• Tools: dbt, Dataform, warehouse-native transformations

---

ETL Pattern Deep Dive

Architecture

Source Systems → ETL Server → Data Warehouse
                    ↓
            [Transform Logic]

When to Use ETL

1. Regulatory Compliance Requirements

Use case: Healthcare (HIPAA), Finance (PCI-DSS)

Why: Sensitive data must be redacted/masked before touching warehouse.

Example:

# ETL transformation to mask PII before loading
import pandas as pd
import hashlib

def mask_pii(df):
    # Hash SSN before loading to warehouse
    df['ssn_hash'] = df['ssn'].apply(
        lambda x: hashlib.sha256(str(x).encode()).hexdigest()
    )
    df = df.drop(columns=['ssn'])  # Remove original
    return df

# Transform before load
raw_data = extract_from_source()
clean_data = mask_pii(raw_data)
load_to_warehouse(clean_data)  # Only clean data hits warehouse

2. Legacy System Integration

Use case: Mainframe data, on-prem systems without cloud access

Why: Target warehouse cannot directly access source systems.

Pattern: Batch extraction to staging area → Transform → Load

3. Real-Time Streaming Transformations

Use case: IoT sensors, clickstream analytics, fraud detection

Why: Need immediate transformation before storage.

Tools: Apache Flink, Kafka Streams, AWS Kinesis Analytics

Example:

# Real-time ETL with Kafka Streams
from kafka import KafkaConsumer, KafkaProducer
import json

consumer = KafkaConsumer('raw-events', bootstrap_servers=['localhost:9092'])
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])

for message in consumer:
    raw_event = json.loads(message.value)

    # Transform in-flight
    transformed_event = {
        'user_id': raw_event['uid'],
        'event_type': raw_event['type'].lower(),
        'timestamp': parse_timestamp(raw_event['ts']),
        'metadata': clean_metadata(raw_event['meta'])
    }

    # Load to clean topic
    producer.send('clean-events', json.dumps(transformed_event).encode())

4. Resource-Constrained Warehouses

Use case: Small warehouse plans, cost optimization

Why: Offload compute to cheaper transformation servers.

ETL Advantages

1. Data Security: Sensitive data never reaches warehouse in raw form 2. Clean Warehouse: Only production-ready data stored 3. Consistent Transformations: Single transformation run for all downstream use cases 4. Network Efficiency: Smaller payloads sent to warehouse (pre-filtered)

ETL Disadvantages

1. Slower at Scale: Sequential processing bottleneck 2. Infrastructure Overhead: Need dedicated ETL servers 3. Less Flexible: Changing logic requires ETL server updates 4. No Raw Data Access: Cannot re-transform historical data without re-extraction

---

ELT Pattern Deep Dive

Architecture

Source Systems → Data Warehouse → BI/Analytics
                      ↓
              [Transform in Warehouse]

When to Use ELT

1. Modern Cloud Data Warehouses

Use case: Snowflake, BigQuery, Databricks, Redshift

Why: Leverage massive parallel processing (MPP) architecture.

Example with dbt:

-- models/marts/fct_orders.sql
-- Runs INSIDE warehouse (Snowflake, BigQuery, etc.)

with orders as (
    select * from {{ source('raw', 'orders') }}  -- Raw data
),

cleaned as (
    select
        order_id,
        customer_id,
        cast(order_date as date) as order_date,
        cast(total_amount as decimal(18,2)) as total_amount,
        lower(trim(status)) as order_status
    from orders
    where order_id is not null
)

select * from cleaned

Performance: Warehouse auto-scales compute based on query complexity.

2. Analytics Engineering Workflows

Use case: Rapid model iteration, A/B testing transformations

Why: SQL analysts can modify transformations without engineering support.

Pattern: Raw data → SQL transformations → Business logic iteration

Example:

-- Analyst can modify this dbt model directly
-- models/marts/customer_segmentation.sql

select
    customer_id,
    case
        when lifetime_value >= 10000 then 'VIP'
        when lifetime_value >= 5000 then 'High Value'
        when lifetime_value >= 1000 then 'Medium Value'
        else 'Low Value'
    end as customer_segment
from {{ ref('customer_metrics') }}

Change thresholds → dbt run → See results immediately.

3. Schema-on-Read Requirements

Use case: Exploratory data analysis, data science

Why: Keep raw data available for unforeseen questions.

Pattern: Load everything → Transform as needed

Example:

-- Raw JSON data in BigQuery
select json_extract_scalar(raw_data, '$.user.email') as email
from `raw.events`
where json_extract_scalar(raw_data, '$.event_type') = 'purchase'

4. Large Dataset Processing

Use case: Petabyte-scale data (logs, events, clickstreams)

Why: Warehouse MPP architecture handles parallelism automatically.

Performance comparison:

• ETL: Single server processes 1TB → 8 hours
• ELT: Warehouse with 100 nodes processes 1TB → 10 minutes

ELT Advantages

1. Scalability: Leverage warehouse parallelism (auto-scaling) 2. Flexibility: Re-transform historical data anytime 3. Speed: Faster for large datasets (MPP vs single server) 4. Democratization: SQL analysts can build transformations 5. Raw Data Preservation: Keep original data for future analysis

ELT Disadvantages

1. Security Risks: Raw data with PII/PHI in warehouse 2. Messy Warehouse: Without governance, accumulates cruft 3. Cost: Warehouse compute can be expensive for complex transformations 4. Requires Modern Stack: Needs cloud warehouse with MPP

---

Hybrid Approaches

Pattern 1: ETL for Sensitive + ELT for Analytics

Use case: Healthcare, finance, any PII-heavy data

Architecture:

Source → ETL (cleanse PII) → Warehouse → ELT (analytics) → Reports

Example:

# Step 1: ETL for PII cleansing
import pandas as pd

def cleanse_pii(df):
    # Mask email domains
    df['email'] = df['email'].str.replace(r'@.*', '@masked.com', regex=True)

    # Hash SSN
    df['ssn_hash'] = df['ssn'].apply(lambda x: hash_ssn(x))
    df = df.drop(columns=['ssn'])

    # Redact addresses
    df['city'] = df['address'].apply(lambda x: extract_city(x))
    df = df.drop(columns=['address'])

    return df

# Load clean data to warehouse
clean_data = cleanse_pii(raw_data)
load_to_warehouse(clean_data)

-- Step 2: ELT for analytics (dbt in warehouse)
-- models/marts/customer_behavior.sql

select
    customer_id,
    email,  -- Already masked
    count(*) as order_count,
    sum(total_amount) as lifetime_value
from {{ ref('stg_orders') }}
group by 1, 2

Pattern 2: Real-Time ETL + Batch ELT

Use case: Streaming events + historical batch processing

Architecture:

Stream → Kafka → Flink (ETL) → Warehouse
Batch Files → S3 → Warehouse → dbt (ELT)

Example:

# Real-time ETL with Flink
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
table_env = StreamTableEnvironment.create(env)

# Stream processing (ETL)
table_env.execute_sql("""
    CREATE TABLE kafka_source (
        event_id STRING,
        event_type STRING,
        user_id STRING,
        timestamp BIGINT
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'raw-events'
    )
""")

# Transform and load to warehouse
table_env.execute_sql("""
    INSERT INTO snowflake_sink
    SELECT
        event_id,
        LOWER(event_type) as event_type,
        user_id,
        TO_TIMESTAMP(timestamp) as event_timestamp
    FROM kafka_source
""")

-- Batch ELT with dbt (historical aggregations)
-- models/marts/daily_event_summary.sql

select
    date_trunc('day', event_timestamp) as event_date,
    event_type,
    count(*) as event_count
from {{ ref('stg_events') }}  -- Includes both stream and batch
group by 1, 2

Pattern 3: Multi-Stage ETL-ELT Pipeline

Architecture:

Source → Light ETL (basic cleansing) → Warehouse → Heavy ELT (analytics)

When to use: Balance between security and flexibility

Example:

# Light ETL: Basic cleansing only
def light_etl(df):
    # Remove nulls
    df = df.dropna(subset=['order_id', 'customer_id'])

    # Standardize types
    df['order_date'] = pd.to_datetime(df['order_date'])
    df['total_amount'] = pd.to_numeric(df['total_amount'])

    # Mask PII
    df['email'] = df['email'].str.replace(r'@.*', '@masked.com', regex=True)

    return df

load_to_warehouse(light_etl(raw_data))

-- Heavy ELT: Complex analytics in warehouse
-- models/marts/customer_cohort_analysis.sql

with first_purchase as (
    select
        customer_id,
        min(order_date) as cohort_month
    from {{ ref('stg_orders') }}
    group by 1
),

monthly_revenue as (
    select
        o.customer_id,
        date_trunc('month', o.order_date) as order_month,
        sum(o.total_amount) as monthly_revenue
    from {{ ref('stg_orders') }} o
    group by 1, 2
)

select
    fp.cohort_month,
    mr.order_month,
    count(distinct mr.customer_id) as active_customers,
    sum(mr.monthly_revenue) as cohort_revenue
from first_purchase fp
inner join monthly_revenue mr on fp.customer_id = mr.customer_id
group by 1, 2

---

Decision Matrix

---

Architecture Patterns

Pattern 1: Pure ETL Architecture

┌─────────────┐
│   Sources   │
│ (APIs, DBs) │
└──────┬──────┘
       │
       ▼
┌─────────────────────┐
│   ETL Server        │
│  (Python, Airflow)  │
│                     │
│ - Extract           │
│ - Clean             │
│ - Aggregate         │
│ - Enrich            │
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│  Data Warehouse     │
│  (Clean Data Only)  │
└──────┬──────────────┘
       │
       ▼
┌─────────────┐
│  BI Tools   │
└─────────────┘

Best for: Legacy systems, compliance-heavy industries

Pattern 2: Pure ELT Architecture

┌─────────────┐
│   Sources   │
│ (APIs, DBs) │
└──────┬──────┘
       │
       ▼
┌─────────────────────┐
│  Data Warehouse     │
│  (Raw + Clean)      │
│                     │
│  ┌──────────────┐   │
│  │ Raw Schema   │   │
│  └──────┬───────┘   │
│         │           │
│         ▼           │
│  ┌──────────────┐   │
│  │ dbt Models   │   │
│  │ (Transform)  │   │
│  └──────┬───────┘   │
│         │           │
│         ▼           │
│  ┌──────────────┐   │
│  │ Analytics    │   │
│  │ Schema       │   │
│  └──────────────┘   │
└──────┬──────────────┘
       │
       ▼
┌─────────────┐
│  BI Tools   │
└─────────────┘

Best for: Modern data teams, cloud-native companies

Pattern 3: Hybrid (Lambda) Architecture

┌─────────────┐
│   Sources   │
└──────┬──────┘
       │
   ┌───┴───┐
   │       │
   ▼       ▼
┌────┐  ┌─────────────┐
│ETL │  │ Raw Load    │
│PII │  │ (Batch)     │
└─┬──┘  └──────┬──────┘
  │            │
  │            ▼
  │     ┌─────────────────┐
  └────►│  Data Warehouse │
        │                 │
        │  ┌───────────┐  │
        │  │ ELT (dbt) │  │
        │  └───────────┘  │
        └────────┬────────┘
                 │
                 ▼
          ┌─────────────┐
          │  BI Tools   │
          └─────────────┘

Best for: Organizations with both compliance and flexibility needs

---

Performance Considerations

ETL Performance Bottlenecks

1. Single Server Limitation: Cannot parallelize beyond server cores 2. Network Transfer: Extract → Transform server → Load (two hops) 3. Memory Constraints: Large datasets require batching

Optimization strategies:

• Batch processing (chunk large datasets)
• Parallel processing (multiprocessing, Spark)
• Incremental extraction (only new/changed data)

ELT Performance Advantages

1. MPP Architecture: Warehouse auto-parallelizes queries across nodes 2. Data Locality: Transformation happens where data is stored (no network transfer) 3. Elastic Scaling: Warehouse scales compute based on query complexity

Example performance comparison:

Dataset: 500 million rows, 200 GB

---

Cost Analysis

ETL Cost Breakdown

1. Infrastructure: EC2/VM instances running 24/7 2. Data Transfer: Egress charges for moving data 3. Maintenance: Engineer time managing servers 4. Licensing: Commercial ETL tools (Informatica, Talend)

Example monthly cost (mid-sized company):

• ETL servers: $2,000/month
• Data transfer: $500/month
• Tool licensing: $5,000/month
• Total: $7,500/month

ELT Cost Breakdown

1. Warehouse Compute: Pay-per-query or reserved capacity 2. Storage: Cheap (store raw + transformed) 3. Tooling: dbt Core (free), dbt Cloud (optional)

Example monthly cost (mid-sized company):

• Warehouse compute (dbt runs): $1,500/month
• Storage: $500/month
• dbt Cloud: $0-$500/month (optional)
• Total: $2,000-$2,500/month

Savings: 60-70% cost reduction with ELT

---

Conclusion

Default Recommendation (2025)

Use ELT unless one of these conditions applies:

1. Regulatory requirement for pre-load PII redaction → Hybrid (ETL + ELT) 2. Legacy systems without cloud warehouse → ETL 3. Real-time streaming with immediate transformation → Streaming ETL + Batch ELT

Migration Path: ETL → ELT

Phase 1: Add ELT alongside existing ETL

• Keep ETL for critical pipelines
• Build new models with dbt (ELT)
• Validate data matches

Phase 2: Gradually migrate ETL pipelines to ELT

• Start with low-risk, low-complexity models
• Test thoroughly
• Decommission ETL jobs after validation

Phase 3: Optimize ELT workflows

• Implement incremental models
• Add data quality tests
• Monitor performance and cost

Timeline: 6-12 months for full migration

---

Additional Resources

• dbt Best Practices: https://docs.getdbt.com/guides/best-practices
• Snowflake ELT Guide: https://www.snowflake.com/guides/what-elt
• BigQuery Data Transformation: https://cloud.google.com/bigquery/docs/best-practices-transformations
• Databricks Delta Lake: https://docs.databricks.com/delta/index.html

Incremental Loading Strategies

Table of Contents

• Overview
• dbt Incremental Strategies
• 1. Append Strategy (Default)
• 2. Merge Strategy
• 3. Insert Overwrite Strategy
• Python Incremental Patterns
• State Tracking with Checkpoints
• Watermark-Based Loading
• Best Practices
• Common Issues

Overview

Incremental loading processes only new or changed data since the last run, improving performance and reducing costs for large datasets.

dbt Incremental Strategies

1. Append Strategy (Default)

Insert new rows only, no updates to existing rows.

{{
  config(
    materialized='incremental',
    unique_key='event_id'
  )
}}

select
    event_id,
    user_id,
    event_type,
    event_timestamp
from {{ ref('stg_events') }}

{% if is_incremental() %}
    where event_timestamp > (select max(event_timestamp) from {{ this }})
{% endif %}

Use for: Immutable event streams, logs, clickstreams

2. Merge Strategy

Update existing rows and insert new rows (upsert).

{{
  config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge',
    merge_update_columns=['order_status', 'updated_at']
  )
}}

select
    order_id,
    customer_id,
    order_status,
    total_amount,
    updated_at
from {{ ref('stg_orders') }}

{% if is_incremental() %}
    where updated_at > (select max(updated_at) from {{ this }})
{% endif %}

Use for: Mutable data with updates (orders, customer profiles)

3. Insert Overwrite Strategy

Replace data for specific partitions only.

{{
  config(
    materialized='incremental',
    unique_key='date',
    incremental_strategy='insert_overwrite',
    partition_by={'field': 'date', 'data_type': 'date'}
  )
}}

select
    date_trunc('day', event_timestamp) as date,
    count(*) as event_count
from {{ ref('stg_events') }}
group by 1

{% if is_incremental() %}
    where date_trunc('day', event_timestamp) >= current_date - interval '7' day
{% endif %}

Use for: Partitioned data, reprocessing specific dates

Python Incremental Patterns

State Tracking with Checkpoints

import polars as pl
from datetime import datetime

def incremental_load():
    # Load last checkpoint
    last_timestamp = read_checkpoint('last_load_timestamp')

    # Extract only new data
    new_data = (
        pl.scan_parquet('s3://bucket/data/*.parquet')
        .filter(pl.col('created_at') > last_timestamp)
        .collect()
    )

    # Process and load
    transform_and_load(new_data)

    # Save new checkpoint
    save_checkpoint('last_load_timestamp', datetime.now())

Watermark-Based Loading

def load_with_watermark():
    # Get high watermark from target table
    high_watermark = get_max_value('target_table', 'updated_at')

    # Extract data above watermark
    new_data = extract_from_source(f"WHERE updated_at > '{high_watermark}'")

    # Merge into target
    merge_data('target_table', new_data, on='id')

Best Practices

1. Always use unique_key for merge strategies 2. Partition large tables for efficient overwrites 3. Test full-refresh periodically to validate incremental logic 4. Monitor data drift between incremental and full loads 5. Use lookback windows to catch late-arriving data

-- Lookback window for late data
{% if is_incremental() %}
    where updated_at > (select max(updated_at) - interval '3' day from {{ this }})
{% endif %}

Common Issues

Issue: Data drift between incremental and full refresh Solution: Schedule periodic full refreshes (dbt run --full-refresh)

Issue: Late-arriving data missed Solution: Use lookback window (process last N days each run)

Issue: Duplicate data from multiple runs Solution: Use unique_key with merge strategy

Pipeline Orchestration Patterns

Table of Contents

1. Airflow Patterns 2. Dagster Patterns 3. Prefect Patterns 4. Tool Comparison 5. Common Orchestration Patterns 6. Best Practices

---

Airflow Patterns

Basic DAG Structure

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'email': ['team@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'execution_timeout': timedelta(hours=1)
}

with DAG(
    dag_id='example_pipeline',
    default_args=default_args,
    description='Example data pipeline',
    schedule_interval='0 2 * * *',  # Daily at 2 AM
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['production', 'etl']
) as dag:

    extract_task = PythonOperator(
        task_id='extract_data',
        python_callable=extract_function
    )

    transform_task = BashOperator(
        task_id='transform_data',
        bash_command='dbt run --models staging'
    )

    load_task = PythonOperator(
        task_id='load_data',
        python_callable=load_function
    )

    # Define dependencies
    extract_task >> transform_task >> load_task

XCom for Task Communication

def extract_data(**context):
    data = fetch_from_api()
    row_count = len(data)

    # Push to XCom
    context['ti'].xcom_push(key='row_count', value=row_count)
    context['ti'].xcom_push(key='max_timestamp', value=data['timestamp'].max())

    return row_count

def validate_data(**context):
    # Pull from XCom
    row_count = context['ti'].xcom_pull(key='row_count', task_ids='extract_data')

    if row_count < 100:
        raise ValueError(f"Insufficient data: only {row_count} rows")

Dynamic Task Generation

from airflow.models import DagBag

def create_processing_tasks():
    regions = ['us-east', 'us-west', 'eu-central', 'ap-southeast']

    for region in regions:
        PythonOperator(
            task_id=f'process_{region}',
            python_callable=process_region,
            op_kwargs={'region': region}
        )

TaskFlow API (Airflow 2.0+)

from airflow.decorators import dag, task
from datetime import datetime

@dag(
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False
)
def etl_pipeline():

    @task
    def extract():
        return fetch_data_from_api()

    @task
    def transform(data: dict):
        return clean_and_transform(data)

    @task
    def load(data: dict):
        save_to_warehouse(data)

    # Automatic XCom handling
    data = extract()
    transformed = transform(data)
    load(transformed)

dag = etl_pipeline()

---

Dagster Patterns

Asset-Based Workflow

from dagster import asset, AssetExecutionContext
import polars as pl

@asset
def raw_orders(context: AssetExecutionContext):
    """Extract raw orders from source system"""
    df = pl.read_csv('s3://bucket/raw/orders.csv')
    context.log.info(f"Extracted {len(df)} orders")
    return df

@asset
def clean_orders(context: AssetExecutionContext, raw_orders: pl.DataFrame):
    """Clean and standardize order data"""
    df = raw_orders.filter(pl.col('order_id').is_not_null())
    context.log.info(f"Cleaned {len(df)} orders")
    return df

@asset
def order_metrics(context: AssetExecutionContext, clean_orders: pl.DataFrame):
    """Calculate order-level metrics"""
    df = clean_orders.group_by('customer_id').agg([
        pl.col('order_amount').sum().alias('total_spent'),
        pl.col('order_id').count().alias('order_count')
    ])
    return df

dbt Integration

from dagster import Definitions
from dagster_dbt import DbtCliResource, dbt_assets

@dbt_assets(manifest=dbt_manifest_path)
def my_dbt_assets(context, dbt: DbtCliResource):
    yield from dbt.cli(['build'], context=context).stream()

defs = Definitions(
    assets=[my_dbt_assets],
    resources={
        'dbt': DbtCliResource(project_dir='/path/to/dbt')
    }
)

Asset Groups and Schedules

from dagster import define_asset_job, ScheduleDefinition

# Define job for specific assets
daily_analytics_job = define_asset_job(
    name='daily_analytics',
    selection=['raw_orders', 'clean_orders', 'order_metrics']
)

# Schedule the job
daily_schedule = ScheduleDefinition(
    job=daily_analytics_job,
    cron_schedule='0 2 * * *'  # Daily at 2 AM
)

---

Prefect Patterns

Flow and Task Decorators

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def extract_data(source: str):
    """Extract data from source with caching"""
    return fetch_from_source(source)

@task(retries=3, retry_delay_seconds=60)
def transform_data(data: dict):
    """Transform data with automatic retries"""
    return apply_transformations(data)

@task
def load_data(data: dict, target: str):
    """Load data to target"""
    save_to_target(data, target)

@flow(name='etl-pipeline')
def etl_flow(source: str, target: str):
    """Main ETL flow"""
    raw_data = extract_data(source)
    transformed_data = transform_data(raw_data)
    load_data(transformed_data, target)

# Run flow
if __name__ == '__main__':
    etl_flow(source='api', target='warehouse')

Dynamic Task Mapping

from prefect import flow, task

@task
def process_file(file_path: str):
    """Process a single file"""
    return transform_file(file_path)

@flow
def batch_processing():
    """Process multiple files in parallel"""
    files = ['file1.csv', 'file2.csv', 'file3.csv']

    # Map task to multiple inputs (parallel execution)
    results = process_file.map(files)

    return results

Subflows

@flow
def staging_flow(table_name: str):
    """Subflow for staging a single table"""
    extract_task(table_name)
    validate_task(table_name)
    return f"{table_name} staged"

@flow
def main_pipeline():
    """Main flow calling multiple subflows"""
    tables = ['orders', 'customers', 'products']

    for table in tables:
        staging_flow(table)  # Call subflow

    aggregate_all_tables()

---

Tool Comparison

When to Choose Airflow

Pros:

• Battle-tested at massive scale (10,000+ DAGs)
• 5,000+ provider packages (integrations)
• Managed services (AWS MWAA, GCP Cloud Composer, Astronomer)
• Large community and extensive documentation

Cons:

• Complex setup and maintenance
• Dynamic workflows require custom code
• Heavier resource requirements

Best for:

• Enterprise production environments
• Complex static workflows
• Teams needing proven stability

When to Choose Dagster

Pros:

• Asset-based paradigm (data-aware)
• Native dbt integration
• Built-in data lineage and testing
• Excellent developer experience

Cons:

• Smaller community than Airflow
• Fewer integrations
• Newer tool (less battle-tested)

Best for:

• dbt-heavy workflows
• ML pipelines
• Data quality focus
• Modern data teams

When to Choose Prefect

Pros:

• Pythonic API (decorators, not classes)
• Dynamic workflows (runtime task generation)
• Cloud-native architecture
• Rich observability

Cons:

• Smallest community of the three
• Fewer integrations than Airflow
• Some features require Prefect Cloud

Best for:

• Dynamic workflows
• Cloud-first companies
• Teams preferring Python-native approach

---

Common Orchestration Patterns

Pattern 1: Linear Pipeline

# Airflow
extract >> transform >> validate >> load

# Dagster (automatic from dependencies)
@asset
def step2(step1): pass

# Prefect
@flow
def pipeline():
    a = step1()
    b = step2(a)
    c = step3(b)

Pattern 2: Fan-Out / Fan-In

# Airflow
extract >> [transform_a, transform_b, transform_c] >> aggregate

# Dagster
@asset
def aggregate(transform_a, transform_b, transform_c):
    return combine_all([transform_a, transform_b, transform_c])

# Prefect
@flow
def fan_out_in():
    data = extract()
    results = [transform_a(data), transform_b(data), transform_c(data)]
    aggregate(results)

Pattern 3: Conditional Execution

# Airflow BranchOperator
from airflow.operators.python import BranchPythonOperator

def choose_branch(**context):
    if condition():
        return 'process_full'
    else:
        return 'process_incremental'

branch = BranchPythonOperator(
    task_id='branch',
    python_callable=choose_branch
)

# Prefect
@flow
def conditional_flow():
    data = extract()
    if len(data) > 1000:
        full_process(data)
    else:
        incremental_process(data)

Pattern 4: Error Handling and Retries

# Airflow
task = PythonOperator(
    task_id='task',
    python_callable=func,
    retries=3,
    retry_delay=timedelta(minutes=5),
    retry_exponential_backoff=True
)

# Dagster
@asset(retry_policy=RetryPolicy(max_retries=3))
def my_asset():
    return process_data()

# Prefect
@task(retries=3, retry_delay_seconds=[60, 120, 300])
def my_task():
    return process_data()

---

Best Practices

1. Idempotency

Ensure tasks produce the same result when run multiple times:

# Good: Idempotent (truncate then insert)
def load_data():
    truncate_table('target_table')
    insert_data('target_table', data)

# Bad: Not idempotent (duplicates on retry)
def load_data():
    insert_data('target_table', data)  # Duplicates if run twice

2. Incremental Processing

Process only new/changed data:

@task
def extract_incremental(**context):
    # Get last successful run timestamp
    last_run = context['prev_ds']

    # Extract only new data
    data = fetch_data(f"WHERE updated_at > '{last_run}'")
    return data

3. Data Quality Checks

Integrate validation into pipeline:

@task
def validate_data(df):
    # Check row count
    assert len(df) > 0, "Empty dataset"

    # Check required columns
    required_cols = ['order_id', 'customer_id', 'amount']
    assert all(col in df.columns for col in required_cols)

    # Check data types
    assert df['amount'].dtype == 'float64'

    return df

4. Monitoring and Alerting

# Airflow: Email/Slack on failure
from airflow.providers.slack.operators.slack import SlackWebhookOperator

notify_failure = SlackWebhookOperator(
    task_id='notify_failure',
    http_conn_id='slack_webhook',
    message='Pipeline failed!',
    trigger_rule='one_failed'
)

# Dagster: Asset checks
from dagster import asset_check

@asset_check(asset=my_asset)
def check_row_count(asset_df):
    if len(asset_df) < 100:
        return AssetCheckResult(passed=False, description="Too few rows")
    return AssetCheckResult(passed=True)

5. Backfill Strategies

# Airflow: Catchup for historical runs
with DAG(
    dag_id='daily_pipeline',
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=True  # Run for all past dates
) as dag:
    pass

# Dagster: Backfill via CLI
# dagster asset backfill --from 2024-01-01 --to 2024-12-31

---

Additional Resources

• Airflow documentation: https://airflow.apache.org/docs/
• Dagster documentation: https://docs.dagster.io/
• Prefect documentation: https://docs.prefect.io/
• Airflow best practices: https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html

SQL Window Functions Guide

Table of Contents

• Overview
• Basic Syntax
• Common Window Functions
• 1. Ranking Functions
• 2. Aggregate Functions
• 3. LAG and LEAD
• Advanced Patterns
• Percent Change Calculation
• First and Last Values
• Percentiles
• Frame Specifications
• ROWS vs RANGE
• Best Practices

Overview

Window functions perform calculations across a set of table rows related to the current row, without grouping rows into a single output row.

Basic Syntax

<window_function>() OVER (
    PARTITION BY <columns>
    ORDER BY <columns>
    ROWS/RANGE BETWEEN <start> AND <end>
)

Common Window Functions

1. Ranking Functions

ROW_NUMBER()

Assigns unique sequential numbers to rows.

select
    order_id,
    customer_id,
    order_date,
    row_number() over (
        partition by customer_id
        order by order_date
    ) as customer_order_number
from orders

RANK() and DENSE_RANK()

select
    product_id,
    sales_amount,
    rank() over (order by sales_amount desc) as rank,
    dense_rank() over (order by sales_amount desc) as dense_rank
from product_sales

2. Aggregate Functions

Running Totals

select
    order_date,
    daily_revenue,
    sum(daily_revenue) over (
        order by order_date
        rows between unbounded preceding and current row
    ) as cumulative_revenue
from daily_sales

Moving Averages

select
    order_date,
    daily_revenue,
    avg(daily_revenue) over (
        order by order_date
        rows between 6 preceding and current row
    ) as revenue_7day_ma
from daily_sales

3. LAG and LEAD

Access previous or next row values.

select
    order_date,
    daily_revenue,
    lag(daily_revenue, 1) over (order by order_date) as prev_day_revenue,
    lead(daily_revenue, 1) over (order by order_date) as next_day_revenue,
    daily_revenue - lag(daily_revenue, 1) over (order by order_date) as revenue_change
from daily_sales

Advanced Patterns

Percent Change Calculation

select
    order_date,
    revenue,
    (revenue / lag(revenue, 1) over (order by order_date) - 1) * 100 as pct_change
from daily_sales

First and Last Values

select
    customer_id,
    order_date,
    first_value(order_date) over (
        partition by customer_id
        order by order_date
    ) as first_order_date,
    last_value(order_date) over (
        partition by customer_id
        order by order_date
        rows between unbounded preceding and unbounded following
    ) as last_order_date
from orders

Percentiles

select
    customer_id,
    total_spent,
    percent_rank() over (order by total_spent) as percentile_rank,
    ntile(4) over (order by total_spent) as quartile
from customer_lifetime_value

Frame Specifications

ROWS vs RANGE

ROWS: Physical number of rows RANGE: Logical range based on values

-- ROWS: Last 3 physical rows
sum(amount) over (
    order by date
    rows between 2 preceding and current row
)

-- RANGE: All rows with same date value
sum(amount) over (
    order by date
    range between unbounded preceding and current row
)

Best Practices

1. Always use PARTITION BY when analyzing groups separately 2. Use ORDER BY to define the window frame 3. Specify frame bounds explicitly for clarity 4. Consider performance impact on large datasets 5. Use CTEs to make complex window logic readable

with daily_metrics as (
    select
        date,
        revenue,
        avg(revenue) over (
            order by date
            rows between 6 preceding and current row
        ) as ma_7day
    from sales
)
select * from daily_metrics
where ma_7day > 10000