Build Your First Python Data Pipeline

Published August 2026 · Data Science · ~1,800 words

Every data science project eventually needs the same thing: a reliable way to move data from its raw form to something useful. You read a file, check for problems, clean it up, save the result, and track what happened.

This is a data pipeline. And building one in Python is a practical skill that pays off in every project — from a weekend analysis to a production system.

Featured image: articles/083/featured-image.svg

Architecture diagram: articles/083/pipeline-architecture.svg

The Five Stages

Every data pipeline has five stages:

Let's build each stage with working code.

Stage 1: Extract

Extracting data means reading it from its source into a pandas DataFrame:

import pandas as pd

def extract_csv(path):
    """Read CSV and return DataFrame."""
    df = pd.read_csv(path)
    print(f'[EXTRACT] Read {len(df):,} rows from {path}')
    return df

def extract_api(url):
    """Read JSON from API endpoint."""
    import requests
    response = requests.get(url)
    response.raise_for_status()
    return pd.DataFrame(response.json())

def extract_database(query, connection_string):
    """Query database directly into DataFrame."""
    from sqlalchemy import create_engine
    engine = create_engine(connection_string)
    return pd.read_sql(query, engine)

The key principle: extraction should do one thing — get the data. Do not clean or transform it here.

Stage 2: Validate

Validation catches problems before they propagate:

def validate(df):
    """Check data quality and return issues list."""
    issues = []
    
    # Check for null values in required columns
    for col in ['product', 'date', 'quantity']:
        nulls = df[col].isna().sum()
        if nulls > 0:
            issues.append(f'{nulls} rows with null {col}')
    
    # Check for invalid values
    if (df['quantity'] <= 0).any():
        issues.append(f'{(df["quantity"] <= 0).sum()} rows with non-positive quantity')
    
    if (df['unit_price'] <= 0).any():
        issues.append(f'{(df["unit_price"] <= 0).sum()} rows with non-positive price')
    
    # Check date format
    invalid_dates = df['date'].apply(
        lambda d: pd.to_datetime(d, errors='coerce') is pd.NaT
    ).sum()
    if invalid_dates > 0:
        issues.append(f'{invalid_dates} rows with invalid dates')
    
    status = 'PASS' if not issues else 'WARN'
    print(f'[VALIDATE] {status} - {len(df):,} rows checked')
    for issue in issues:
        print(f'  WARNING: {issue}')
    
    return df, issues

Validation does not fix data — it reports what is wrong. This separation is important because you may want different handling strategies for different types of issues.

Stage 3: Transform

Transforming cleans the data and adds useful derived columns:

def transform(df):
    """Clean data and add derived columns."""
    initial = len(df)
    
    # Remove invalid rows
    df = df.dropna(subset=['product', 'date'])
    df = df[df['quantity'] > 0]
    df = df[df['unit_price'] > 0]
    
    # Validate and parse dates
    df['date'] = pd.to_datetime(df['date'], errors='coerce')
    df = df.dropna(subset=['date'])
    
    # Normalize text
    df['region'] = df['region'].str.title()
    df['category'] = df['category'].replace('', 'Uncategorized')
    
    # Fill defaults
    df['customer_email'] = df['customer_email'].fillna('unknown@example.com')
    
    # Add derived columns
    df['total'] = (df['quantity'] * df['unit_price']).round(2)
    df['month'] = df['date'].dt.to_period('M').astype(str)
    df['day_of_week'] = df['date'].dt.day_name()
    df['is_large_order'] = df['total'] > 5000
    
    dropped = initial - len(df)
    print(f'[TRANSFORM] {initial:,} -> {len(df):,} rows ({dropped} dropped)')
    return df

Transform rules to follow:

• Remove invalid data before computing derived columns
• Normalize text early (case, whitespace)
• Fill defaults explicitly — do not rely on pandas auto-fill
• Document every derived column's purpose

Stage 4: Load

Loading saves the clean data to its destination:

import json
from datetime import datetime

def load(df, output_dir):
    """Save clean data and generate summary."""
    # Save to Parquet (compressed, fast, type-preserving)
    parquet_path = f'{output_dir}/clean_data.parquet'
    df.to_parquet(parquet_path, index=False, compression='zstd')
    
    # Generate summary
    summary = {
        'total_rows': len(df),
        'total_revenue': round(df['total'].sum(), 2),
        'avg_order_value': round(df['total'].mean(), 2),
        'unique_products': int(df['product'].nunique()),
        'date_range': f"{df['date'].min().date()} to {df['date'].max().date()}",
        'generated_at': datetime.now().isoformat(),
    }
    
    summary_path = f'{output_dir}/summary.json'
    with open(summary_path, 'w') as f:
        json.dump(summary, f, indent=2)
    
    print(f'[LOAD] Saved {len(df):,} rows to {parquet_path}')
    return summary

Loading best practices:

• Use Parquet for analytical data — compressed, typed, fast
• Generate a summary document — future you will thank present you
• Include a timestamp — know when the data was generated
• Handle load failures gracefully — do not lose clean data

Stage 5: Monitor

Monitoring tracks what happened at each stage:

import time

class PipelineMonitor:
    def __init__(self):
        self.steps = []
        self.start_time = time.perf_counter()
    
    def log_step(self, name, rows_in, rows_out, duration_ms, status='ok'):
        self.steps.append({
            'step': name,
            'rows_in': rows_in,
            'rows_out': rows_out,
            'duration_ms': round(duration_ms, 1),
            'status': status,
        })
    
    def report(self):
        total = (time.perf_counter() - self.start_time) * 1000
        print(f'\n=== PIPELINE REPORT ===')
        print(f'Total time: {total:.1f} ms')
        for s in self.steps:
            print(f"  [{s['status'].upper():>6}] {s['step']:<15} "
                  f"{s['rows_in']} -> {s['rows_out']}  "
                  f"{s['duration_ms']:>8.1f} ms")
        return self.steps

Why monitoring matters:

• Debugging — know which step failed or is slow
• Row tracking — see where data was lost
• Performance — identify bottlenecks
• Auditing — prove what happened and when

The Complete Pipeline

Putting it all together:

import pandas as pd
import time

def run_pipeline(input_csv, output_dir):
    """Run the complete data pipeline."""
    monitor = PipelineMonitor()
    
    # 1. Extract
    t0 = time.perf_counter()
    raw = pd.read_csv(input_csv)
    monitor.log_step('EXTRACT', 0, len(raw),
                      (time.perf_counter() - t0) * 1000)
    
    # 2. Validate
    t0 = time.perf_counter()
    validated, issues = validate(raw)
    monitor.log_step('VALIDATE', len(raw), len(raw),
                      (time.perf_counter() - t0) * 1000,
                      status='warn' if issues else 'ok')
    
    # 3. Transform
    t0 = time.perf_counter()
    clean = transform(validated)
    monitor.log_step('TRANSFORM', len(raw), len(clean),
                      (time.perf_counter() - t0) * 1000)
    
    # 4. Load
    t0 = time.perf_counter()
    summary = load(clean, output_dir)
    monitor.log_step('LOAD', len(clean), len(clean),
                      (time.perf_counter() - t0) * 1000)
    
    # 5. Monitor
    monitor.report()
    
    return summary

# Run it
if __name__ == '__main__':
    summary = run_pipeline('raw_data.csv', './output')
    print(summary)

Real Pipeline Output

When we ran this pipeline on 10,000 synthetic sales rows:

The validation step found 6 types of quality issues: missing products, negative quantities, negative prices, invalid dates, empty categories, and missing emails. The transform step removed 1,009 invalid rows and added derived columns.

Common Pipeline Mistakes

Scaling the Pipeline

The same five stages work at every scale:

The architecture stays the same. Only the tools change.

Best Practices

Further Reading

• Parquet vs CSV — why Parquet is the right output format
• DuckDB: SQL on Your Laptop — query pipeline outputs with SQL
• Why ML Models Fail in Production — pipelines prevent many production failures
• Model Drift Explained — monitoring pipeline outputs over time
• Data Leakage in ML — validation catches leakage sources
• Train/Val/Test Sets — pipeline-aware data splitting
• Local Python Docker Workspace — reproducible pipeline environment

Conclusion

A data pipeline is not complicated. It is five steps: extract, validate, transform, load, monitor. Each step has a clear purpose, a clear input, and a clear output.

The discipline of separating these stages — and logging what happens at each one — is what separates a reliable data project from a fragile one. The code is not hard. The architecture is not complex. The value is in the structure.