Data Pipeline Skill

Design, build, and troubleshoot data pipelines that reliably move and transform data. Cover ingestion patterns, transformation logic, error handling, idempotency, orchestration, and monitoring.

Workflow

Step 1: Define Pipeline Requirements

Gather these details before designing:

1. Source(s): What systems produce the data? (APIs, databases, files, streams)
2. Destination: Where does the data need to land? (Warehouse, lake, application DB)
3. Volume: How much data per run? (KBs, MBs, GBs, TBs)
4. Frequency: Real-time, hourly, daily, weekly?
5. Freshness SLA: How stale can the data be? (Minutes, hours, days)
6. Quality requirements: What data quality checks are needed?
7. Dependencies: What other pipelines or systems must complete first?
8. Error tolerance: Can the pipeline skip bad records or must it fail on any error?

Step 2: Choose the Architecture Pattern

Step 3: Design the Pipeline Stages

Stage 1: Extraction / Ingestion

# API ingestion pattern
def extract_from_api(endpoint, params, pagination_key='next_cursor'):
    """Extract data from a paginated API."""
    all_records = []
    cursor = None

    while True:
        if cursor:
            params[pagination_key] = cursor
        response = requests.get(endpoint, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()
        all_records.extend(data['results'])

        cursor = data.get('next_cursor')
        if not cursor:
            break

    return all_records

# Database extraction with incremental loading
def extract_incremental(connection, table, watermark_column, last_watermark):
    """Extract only new/changed records since last run."""
    query = f"""
        SELECT *
        FROM {table}
        WHERE {watermark_column} > %(last_watermark)s
        ORDER BY {watermark_column}
    """
    df = pd.read_sql(query, connection, params={'last_watermark': last_watermark})
    new_watermark = df[watermark_column].max()
    return df, new_watermark

Stage 2: Validation

def validate_extracted_data(df, expectations):
    """Validate data meets expectations before processing."""
    issues = []

    # Row count check
    if len(df) < expectations.get('min_rows', 0):
        issues.append(f"Row count {len(df)} below minimum {expectations['min_rows']}")

    # Required columns
    missing_cols = set(expectations.get('required_columns', [])) - set(df.columns)
    if missing_cols:
        issues.append(f"Missing columns: {missing_cols}")

    # Null check on critical columns
    for col in expectations.get('not_null_columns', []):
        null_count = df[col].isnull().sum()
        if null_count > 0:
            issues.append(f"Column {col} has {null_count} nulls")

    # Freshness check
    if 'date_column' in expectations:
        max_date = df[expectations['date_column']].max()
        if max_date < expectations.get('min_date'):
            issues.append(f"Data only goes to {max_date}, expected at least {expectations['min_date']}")

    if issues:
        raise DataValidationError(issues)

    return True

Stage 3: Transformation

# Transformation pattern with audit trail
def transform(df):
    """Apply business transformations with logging."""
    initial_count = len(df)

    # Clean
    df = df.dropna(subset=['customer_id'])
    print(f"Dropped {initial_count - len(df)} rows with null customer_id")

    # Transform
    df['revenue_usd'] = df.apply(convert_currency, axis=1)
    df['fiscal_quarter'] = df['order_date'].apply(get_fiscal_quarter)

    # Deduplicate
    before_dedup = len(df)
    df = df.drop_duplicates(subset=['order_id'], keep='last')
    print(f"Removed {before_dedup - len(df)} duplicate orders")

    # Validate output
    assert len(df) > 0, "Transformation produced zero rows"
    assert df['revenue_usd'].notna().all(), "Null revenue after currency conversion"

    return df

Stage 4: Loading

# Idempotent load pattern
def load_idempotent(df, table, partition_column, partition_value, engine):
    """Load data idempotently by replacing the partition."""
    with engine.begin() as conn:
        # Delete existing data for this partition
        conn.execute(
            f"DELETE FROM {table} WHERE {partition_column} = %(val)s",
            {'val': partition_value}
        )
        # Insert new data
        df.to_sql(table, conn, if_exists='append', index=False)
        print(f"Loaded {len(df)} rows into {table} for {partition_column}={partition_value}")

Step 4: Implement Error Handling

import time
from functools import wraps

def retry_with_backoff(max_retries=3, base_delay=1, max_delay=60):
    """Retry decorator with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except (ConnectionError, TimeoutError) as e:
                    if attempt == max_retries:
                        raise
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
                    time.sleep(delay)
        return wrapper
    return decorator

# Dead letter queue pattern for bad records
def process_with_dlq(records, process_fn, dlq_table):
    """Process records, routing failures to a dead letter queue."""
    successes, failures = [], []
    for record in records:
        try:
            result = process_fn(record)
            successes.append(result)
        except Exception as e:
            record['error'] = str(e)
            record['failed_at'] = datetime.utcnow()
            failures.append(record)

    if failures:
        pd.DataFrame(failures).to_sql(dlq_table, engine, if_exists='append')
        print(f"Routed {len(failures)} records to DLQ")

    return successes

Step 5: Implement Idempotency

Ensure the pipeline can be re-run safely:

1. Partition-based replace: Delete and re-insert data for a specific partition (date, batch ID).
2. Upsert/Merge: Use MERGE or INSERT ON CONFLICT to handle duplicates.
3. Watermark tracking: Store the last processed timestamp/ID and resume from there.
4. Atomic operations: Use database transactions so partial loads don't corrupt data.

-- MERGE pattern (Snowflake/BigQuery)
MERGE INTO target_table t
USING staging_table s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET
    t.amount = s.amount,
    t.updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT (id, amount, created_at, updated_at)
VALUES (s.id, s.amount, s.created_at, s.updated_at);

Step 6: Set Up Monitoring and Alerting

Track these metrics for every pipeline:

# Pipeline monitoring pattern
def log_pipeline_metrics(pipeline_name, metrics):
    """Log pipeline run metrics for monitoring."""
    record = {
        'pipeline': pipeline_name,
        'run_id': str(uuid.uuid4()),
        'started_at': metrics['started_at'],
        'completed_at': datetime.utcnow(),
        'duration_seconds': (datetime.utcnow() - metrics['started_at']).total_seconds(),
        'rows_extracted': metrics['rows_extracted'],
        'rows_loaded': metrics['rows_loaded'],
        'rows_rejected': metrics['rows_rejected'],
        'status': 'success',
    }
    # Write to monitoring table or send to observability system
    pd.DataFrame([record]).to_sql('pipeline_runs', engine, if_exists='append')

Output Format

When designing a pipeline, deliver:

1. Architecture diagram (text-based) showing data flow
2. Stage-by-stage specification with code or pseudocode
3. Error handling strategy with retry logic and DLQ
4. Idempotency approach explaining how re-runs are safe
5. Monitoring plan with metrics and alert thresholds
6. Runbook for common failure scenarios

Examples

Example 1: Daily API-to-Warehouse Pipeline

Extract from REST API with pagination, validate schema, transform and clean, load into Snowflake with partition-based idempotency, orchestrate with Airflow.

Example 2: Real-Time Event Processing

Consume from Kafka, validate schema with Avro, enrich with dimension lookups, write to warehouse and real-time dashboard, handle late-arriving events.

Example 3: Database Replication

CDC from PostgreSQL using Debezium, write to Kafka, consume into warehouse, handle schema evolution, maintain referential integrity.

Best Practices

1. Make it idempotent. Every pipeline must be safe to re-run without creating duplicates or corruption.
2. Fail loudly. Silent failures are worse than crashes. Alert on anomalies, not just errors.
3. Log everything. Record row counts, durations, and error details for every run.
4. Validate at boundaries. Check data quality after extraction and before loading.
5. Use incremental loads. Full refreshes do not scale. Load only new/changed data.
6. Handle schema evolution. Expect source schemas to change. Detect and adapt or alert.
7. Separate extraction from transformation. Land raw data first, then transform. This makes debugging easier.
8. Test with production-scale data. A pipeline that works with 100 rows may fail with 100 million.
9. Document dependencies. Make upstream and downstream dependencies explicit and visible.
10. Plan for backfills. Design pipelines so you can re-process historical data when logic changes.