etl-pipelines

name	etl-pipelines
description	Extract, Transform, Load (ETL) pipeline design and implementation for moving data from source systems, transforming it to meet requirements, and loading it into target systems for analysis.
category	data-science
keywords	["etl-pipelines","data pipelines","data engineering","data transformation","data integration","workflow orchestration","apache airflow","dag"]
difficulty	intermediate
related_skills	["pandas","big-data","data-visualization"]

ETL Pipelines

What I do

I provide frameworks and patterns for building robust data pipelines that extract data from various sources, transform it to meet business requirements, and load it into target systems. I ensure data quality, enable scheduled execution, and provide monitoring and error handling capabilities for production data workflows.

When to use me

• Moving data from operational databases to data warehouses
• Consolidating data from multiple sources
• Creating data lakes from raw data
• Building feature stores for machine learning
• Scheduling regular data refreshes
• Implementing data quality checks
• Building real-time streaming pipelines
• Creating data aggregation workflows

Core Concepts

Pipeline Components

• Extract: Pull data from source systems (databases, APIs, files)
• Transform: Clean, validate, enrich, aggregate data
• Load: Write data to destination systems
• Orchestrate: Manage dependencies and scheduling

Data Quality

• Schema validation: Verify structure matches expectations
• Data type checks: Ensure correct types
• Range checks: Validate value ranges
• Uniqueness checks: Detect duplicate records
• Referential integrity: Verify relationships

Orchestration

• DAGs: Directed Acyclic Graphs for dependency management
• Task dependencies: Define execution order
• Failure handling: Retries, alerts, dead letter queues
• Backfilling: Re-processing historical data

Tools

• Apache Airflow: Workflow orchestration
• Prefect/Dagster: Modern orchestration frameworks
• Luigi: Spotify's pipeline framework
• dbt: Data transformation tool

Code Examples (Python)

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging

# Basic ETL implementation
class BasicETL:
    def __init__(self, source_conn, target_conn):
        self.source = source_conn
        self.target = target_conn
        self.logger = logging.getLogger(__name__)
    
    def extract(self, query, params=None):
        """Extract data from source."""
        self.logger.info(f"Extracting data with query: {query[:100]}...")
        df = pd.read_sql(query, self.source, params=params)
        self.logger.info(f"Extracted {len(df)} rows")
        return df
    
    def transform(self, df):
        """Transform data."""
        self.logger.info(f"Transforming {len(df)} rows...")
        
        # Clean column names
        df.columns = [c.strip().lower().replace(' ', '_') for c in df.columns]
        
        # Handle missing values
        df = df.fillna({
            : ,
            : ,
            : pd.NaT
        })
        
        
        df[] = pd.to_datetime(df[], errors=)
        df[] = pd.to_numeric(df[], errors=)
        
        
        df = df.drop_duplicates(subset=[])
        
        
        df[] = datetime.now()
        
        
        df = df[df[] >= ]
        
         df
    
     ():
        
        .logger.info()
        df.to_sql(table_name, .target, if_exists=if_exists, index=)
        .logger.info()
    
     ():
        
        :
            df = .extract(extract_query)
            df = .transform(df)
             transform_func:
                df = transform_func(df)
            .load(df, target_table)
             
         Exception  e:
            .logger.error()
            


 :

     ():
        
         col, dtype  expected_schema.items():
             col   df.columns:
                 ValueError()
             df[col].dtype != dtype:
                 TypeError()
    

     ():
        
         min_val     (df[col] < min_val).():
            invalid = (df[col] < min_val).()
             ValueError()
         max_val     (df[col] > max_val).():
            invalid = (df[col] > max_val).()
             ValueError()
    

     ():
        
        duplicates = df[col].duplicated().()
         duplicates > :
             ValueError()
    

     ():
        
         col  required_cols:
            completeness = df[col].notna().mean()
             completeness < threshold:
                 ValueError()


 :
     ():
        .source = source
        .target = target
    
     ():
        
        :
            query = 
            result = pd.read_sql(query, .target)
             result[].iloc[]  (result) >   
        :
             
    
     ():
        
        log_df = pd.DataFrame([{
            : table_name,
            : rows_processed,
            : status,
            : datetime.now()
        }])
        log_df.to_sql(, .target, if_exists=, index=)
    
     ():
        
        last_run = .get_last_processed(table_name)
        
         last_run:
            query = extract_query_template.(watermark=)
        :
            query = extract_query_template.(watermark=)
        
        df = pd.read_sql(query, .source)
        
         (df) == :
            .logger.info()
            
        
        
        df = .transform(df)
        
        
        .load(df, table_name, if_exists=)
        
        
        .log_etl(table_name, (df), )
        
         df





Transform single record.Process streaming records.

Best Practices

1. Design for failure: Assume any component can fail; implement retries, logging, and alerts.

2. Incremental over full reload: Process only new/changed data when possible for efficiency.

3. Idempotent operations: Running the pipeline multiple times should produce the same result.

4. Separate concerns: Keep extraction, transformation, and loading logic independent.

5. Metadata management: Track data lineage, schema versions, and processing timestamps.

6. Monitor data quality: Implement validation at each stage with alerting.

7. Handle schema evolution: Plan for schema changes without breaking pipelines.

8. Document data contracts: Define clear interfaces between pipeline stages.

Common Patterns

Pattern 1: Resilient Batch ETL

class ResilientETL:
    def __init__(self, config):
        self.config = config
        self.max_retries = 3
        self.backoff_factor = 2
    
    def run_with_retry(self, task_name, task_func, **kwargs):
        """Run task with exponential backoff retry."""
        for attempt in range(self.max_retries):
            try:
                return task_func(**kwargs)
            except Exception as e:
                if attempt == self.max_retries - 1:
                    self.send_alert(task_name, e)
                    raise
                sleep_time = self.backoff_factor ** attempt
                self.logger.warning(f"Task {task_name} failed, retrying in {sleep_time}s")
                sleep(sleep_time)
    
    def checkpoint(self, stage, state):
        """Save pipeline state for recovery."""
        checkpoint_df = pd.DataFrame([{
            'stage': stage,
            'state': json.dumps(state),
            'timestamp': datetime.now()
        }])
        checkpoint_df.to_sql('pipeline_checkpoints', 
                           self.checkpoint_db, 
                           if_exists=)
    
     ():
        
        query = 
         pd.read_sql(query, .checkpoint_db).iloc[][]

Pattern 2: Data Quality Framework

class DataQualityFramework:
    def __init__(self):
        self.checks = []
    
    def add_check(self, name, check_func, severity='error'):
        """Add quality check."""
        self.checks.append({'name': name, 'func': check_func, 'severity': severity})
    
    def run_checks(self, df, context):
        """Execute all quality checks."""
        results = []
        for check in self.checks:
            try:
                passed = check['func'](df, context)
                results.append({
                    'check': check['name'],
                    'passed': passed,
                    'severity': check['severity']
                })
            except Exception as e:
                results.append({
                    'check': check['name'],
                    'passed': False,
                    'error': str(e),
                    'severity': check['severity']
                })
        
        # Raise on error severity failures
        errors = [r for r in results if not r[]  r[] == ]
         errors:
             DataQualityError(errors)
        
         results
    

     ():
        
        framework = DataQualityFramework()
        
        framework.add_check(, 
             df, ctx: df.isnull().().() == )
        
        framework.add_check(,
             df, ctx: (df.columns) == (ctx[]))
        
        framework.add_check(,
             df, ctx: (df[ctx[]] >= ).().())
        
        framework.add_check(,
             df, ctx: df.duplicated().() == )
        
         framework

Pattern 3: Feature Engineering Pipeline

class FeatureEngineeringPipeline:
    def __init__(self, feature_store):
        self.feature_store = feature_store
        self.transformations = []
    
    def add_feature(self, name, source_col, transform_func):
        """Add feature transformation."""
        self.transformations.append({
            'name': name,
            'source': source_col,
            'func': transform_func
        })
    
    def build_features(self, df, timestamp_col='timestamp'):
        """Build all features for a dataset."""
        features_df = df.copy()
        
        for feat in self.transformations:
            features_df[feat['name']] = feat['func'](features_df[feat['source']])
        
        # Add temporal features
        features_df['hour'] = features_df[timestamp_col].dt.hour
        features_df['day_of_week'] = features_df[timestamp_col].dt.dayofweek
        features_df['is_weekend'] = features_df['day_of_week'].isin([5, 6]).astype(int)
        
        # Add rolling features
        features_df = self.add_rolling_features(features_df, 'value', '1h', '24h')
        
        return features_df
    
    def ():
        
         window  windows:
            df[] = df.groupby()[col].transform(
                 x: x.rolling(window, min_periods=).mean()
            )
         df
    
     ():
        
        .feature_store.save(
            entity_id=entity_id,
            features=features_df.to_dict(orient=),
            timestamp=timestamp
        )