Data Pipelines¶

Data pipelines move and transform data from source to destination. This sheet covers architecture patterns, orchestration, monitoring, error handling, testing, and production best practices.

Table of Contents¶

Installation / Setup
Core Syntax
Common Operations
Practical Examples
Common Patterns
Best Practices
Common Mistakes
Troubleshooting
Related Cheat Sheets
Official Documentation

Installation / Setup¶

# Common pipeline tools
pip install apache-airflow    # orchestration
pip install great-expectations  # data quality
pip install dbt-core          # transformation
pip install pandas            # single-machine
pip install pyspark           # distributed

Core Syntax¶

Pipeline Architecture Patterns¶

Pattern	Description	When to Use
Batch	Process data in scheduled batches	Daily/hourly reporting
Streaming	Process data in real-time	Low-latency requirements
Lambda	Batch + streaming layers	Flexibility + real-time
Kappa	Streaming only	Simplicity; event-sourced
Medallion	Bronze/Silver/Gold layers	Lakehouse architecture

Medallion Architecture:
Bronze: Raw data (ingested)
Silver: Cleaned, validated
Gold: Aggregated, business-ready

Pipeline Components¶

Source → Ingestion → Storage → Processing → Serving → Destination

Components:
- Source: databases, APIs, files, streams
- Ingestion: connectors, CDC, webhooks
- Storage: data lake, warehouse, lakehouse
- Processing: batch, streaming, transformation
- Serving: APIs, dashboards, exports
- Destination: databases, files, applications

DAG (Directed Acyclic Graph)¶

# Pipeline as a DAG
from airflow import DAG
from airflow.operators.python import PythonOperator

with DAG('my_pipeline', schedule_interval='@daily') as dag:
    extract = PythonOperator(task_id='extract', python_callable=extract)
    transform = PythonOperator(task_id='transform', python_callable=transform)
    load = PythonOperator(task_id='load', python_callable=load)
    validate = PythonOperator(task_id='validate', python_callable=validate)

    extract >> transform >> load >> validate

Common Operations¶

Orchestration¶

# Airflow DAG
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def extract(**context):
    # Extract data
    pass

def transform(**context):
    # Transform data
    pass

def load(**context):
    # Load data
    pass

with DAG(
    'etl_pipeline',
    start_date=datetime(2026, 1, 1),
    schedule_interval='@daily',
    catchup=False,
) as dag:
    t1 = PythonOperator(task_id='extract', python_callable=extract)
    t2 = PythonOperator(task_id='transform', python_callable=transform)
    t3 = PythonOperator(task_id='load', python_callable=load)

    t1 >> t2 >> t3

Monitoring¶

import time
import logging
from datetime import datetime

class PipelineMonitor:
    def __init__(self, pipeline_name):
        self.pipeline_name = pipeline_name
        self.start_time = None
        self.metrics = {}

    def start(self):
        self.start_time = time.time()
        logging.info(f"Pipeline {self.pipeline_name} started")

    def record_metric(self, name, value):
        self.metrics[name] = value

    def finish(self, success=True):
        duration = time.time() - self.start_time
        self.metrics['duration_seconds'] = duration
        self.metrics['success'] = success
        self.metrics['end_time'] = datetime.now().isoformat()

        logging.info(f"Pipeline {self.pipeline_name} finished: {self.metrics}")

        # Send to monitoring system
        self.send_to_monitoring()

    def send_to_monitoring(self):
        # Send metrics to Prometheus, Datadog, etc.
        pass

Error Handling¶

from airflow.models import Variable
from airflow.utils.email import send_email

def handle_failure(context):
    """Handle pipeline failure."""
    # Log error
    logging.error(f"Pipeline failed: {context}")

    # Send alert
    send_email(
        to=['team@example.com'],
        subject=f"Pipeline failed: {context['dag'].dag_id}",
        html_content=str(context)
    )

# Use in DAG
default_args = {
    'on_failure_callback': handle_failure,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

Testing¶

import pytest
import pandas as pd

def test_extract():
    """Test extraction."""
    df = extract_data()
    assert len(df) > 0
    assert 'id' in df.columns

def test_transform():
    """Test transformation."""
    df = extract_data()
    result = transform_data(df)
    assert result.isnull().sum().sum() == 0
    assert len(result) > 0

def test_load():
    """Test loading."""
    df = transform_data(extract_data())
    count = load_data(df)
    assert count == len(df)

# Integration test
def test_pipeline():
    """Test full pipeline."""
    df = run_pipeline()
    assert len(df) > 0
    assert validate_data(df)

Practical Examples¶

Example 1 — Batch pipeline¶

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

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

def extract_orders(**context):
    import pandas as pd
    df = pd.read_sql("SELECT * FROM orders WHERE date = %s", conn, params=[context['ds']])
    return df.to_parquet('/staging/orders.parquet')

def transform_orders(**context):
    import pandas as pd
    df = pd.read_parquet('/staging/orders.parquet')
    df['total'] = df['quantity'] * df['price']
    df.to_parquet('/staging/orders_transformed.parquet')

def load_orders(**context):
    import pandas as pd
    df = pd.read_parquet('/staging/orders_transformed.parquet')
    df.to_sql('orders_daily', conn, if_exists='append', index=False)

with DAG('daily_orders', default_args=default_args, schedule_interval='0 2 * * *') as dag:
    t1 = PythonOperator(task_id='extract', python_callable=extract_orders)
    t2 = PythonOperator(task_id='transform', python_callable=transform_orders)
    t3 = PythonOperator(task_id='load', python_callable=load_orders)

    t1 >> t2 >> t3

Example 2 — Multi-source pipeline¶

def extract_from_source(source):
    def extract():
        if source == 'api':
            return extract_from_api()
        elif source == 'database':
            return extract_from_db()
        elif source == 'file':
            return extract_from_file()
    return extract

sources = ['api', 'database', 'file']

with DAG('multi_source', default_args=default_args) as dag:
    extracts = [
        PythonOperator(task_id=f'extract_{s}', python_callable=extract_from_source(s))
        for s in sources
    ]

    merge = PythonOperator(task_id='merge', python_callable=merge_data)
    load = PythonOperator(task_id='load', python_callable=load_data)

    extracts >> merge >> load

Example 3 — Data quality pipeline¶

import great_expectations as gx

def validate_data():
    context = gx.get_context()
    batch_request = context.sources.pandas_default.read_csv("data.csv")

    suite = context.add_expectation_suite("my_suite")
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="id")
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="amount", min_value=0, max_value=10000
        )
    )

    result = context.run_checkpoint("my_checkpoint")
    assert result.success

Common Patterns¶

Pattern: retry with exponential backoff¶

import random

def retry_with_backoff(func, max_retries=3, base_delay=1):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            time.sleep(delay)

Pattern: circuit breaker¶

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None

    def call(self, func):
        if self.is_open():
            raise Exception("Circuit breaker is open")
        try:
            result = func()
            self.reset()
            return result
        except Exception as e:
            self.record_failure()
            raise

    def is_open(self):
        if self.last_failure_time is None:
            return False
        return (time.time() - self.last_failure_time) < self.recovery_timeout

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

    def reset(self):
        self.failure_count = 0
        self.last_failure_time = None

Pattern: dead letter queue¶

def process_with_dlq(messages):
    dlq = []

    for message in messages:
        try:
            process(message)
        except Exception as e:
            dlq.append({'message': message, 'error': str(e)})

    # Store failed messages for reprocessing
    if dlq:
        store_dlq(dlq)

Best Practices¶

Idempotent tasks — safe to rerun without side effects.
Atomic transactions — all or nothing for data loads.
Monitoring and alerting — know when things fail.
Data lineage — track where data comes from and goes.
Testing — unit tests, integration tests, data quality tests.
Documentation — document schemas, dependencies, and runbooks.
Version control — track pipeline code changes.
Separation of concerns — extract, transform, load as separate steps.

Common Mistakes¶

Mistake	Fix
No error handling	Add try/except and alerting
No monitoring	Track metrics and alert
No testing	Add unit and integration tests
Hardcoding credentials	Use secrets management
No lineage tracking	Add metadata and audit columns
Tight coupling	Make components independent
No rollback strategy	Implement transactions and backups

Troubleshooting¶

Symptom	Cause / Fix
Pipeline fails silently	Add error handling and alerting
Slow performance	Profile and optimize bottlenecks
Data quality issues	Add validation; check sources
Duplicates	Use idempotent loads; upsert
Missing data	Check source connectivity; check logs
Memory errors	Process in batches; increase resources

airflow.md — orchestration with Airflow
etl-elt.md — ETL/ELT patterns
apache-spark.md — distributed processing
delta-lake.md — Delta Lake operations
ci-cd — CI/CD for pipelines

Official Documentation¶

Apache Airflow — https://airflow.apache.org/docs/
Great Expectations — https://docs.greatexpectations.io/
dbt — https://docs.getdbt.com/

Last Updated¶

2026-09-06 — Initial version.

Data Pipelines¶

Table of Contents¶

Installation / Setup¶

Core Syntax¶

Pipeline Architecture Patterns¶

Pipeline Components¶

DAG (Directed Acyclic Graph)¶

Common Operations¶

Orchestration¶

Monitoring¶

Error Handling¶

Testing¶

Practical Examples¶

Example 1 — Batch pipeline¶

Example 2 — Multi-source pipeline¶

Example 3 — Data quality pipeline¶

Common Patterns¶

Pattern: retry with exponential backoff¶

Pattern: circuit breaker¶

Pattern: dead letter queue¶

Best Practices¶

Common Mistakes¶

Troubleshooting¶

Related Cheat Sheets¶

Official Documentation¶

Last Updated¶