Data Pipelines and ETL: From Raw Data to Insights

Data Pipelines and ETL: From Raw Data to Insights

What is ETL?

ETL stands for Extract, Transform, Load - the three-stage process of building data pipelines:

• Extract: Retrieve data from source systems (APIs, databases, files)
• Transform: Clean, validate, and reshape data into useful format
• Load: Move processed data to target system (warehouse, database)

ETL Pipeline Architecture

Data Sources (APIs, DB, Files) ↓ [EXTRACT] ↓ [TRANSFORM] - Clean, validate, enrich, aggregate ↓ [LOAD] ↓
Data Warehouse / Analytics DB / Data Lake ↓ [ANALYZE] - Reports, Dashboards, ML models

Stage 1: Extract

import requests
import json
import csv
from datetime import datetime

# Extract from REST API
def extract_from_api(api_url, params=None): """Fetch data from API""" try: response = requests.get(api_url, params=params, timeout=10) response.raise_for_status() return response.json() except requests.RequestException as e: print(f"Error extracting from API: {e}") return None

# Extract from CSV file
def extract_from_csv(file_path): """Read CSV file into list of dictionaries""" data = [] try: with open(file_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) data = list(reader) except Exception as e: print(f"Error reading CSV: {e}") return data

# Extract from JSON file
def extract_from_json(file_path): """Read JSON file""" try: with open(file_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"Error reading JSON: {e}") return []

# Extract from database
from pymongo import MongoClient

def extract_from_mongodb(connection_string, db_name, collection_name, query=None): """Extract from MongoDB""" client = MongoClient(connection_string) db = client[db_name] collection = db[collection_name] return list(collection.find(query or {}))

# Example: Extract Census data from India
census_data = extract_from_api( 'https://api.data.gov.in/census/2021', params={'state': 'Maharashtra', 'limit': 1000}
)

Stage 2: Transform

import pandas as pd
from datetime import datetime
import re

def clean_and_validate_data(data): """Remove duplicates, handle missing values, validate""" # Remove duplicates df = pd.DataFrame(data) df = df.drop_duplicates() # Handle missing values df['email'] = df['email'].fillna('') df['phone'] = df['phone'].fillna('N/A') df['age'] = df['age'].fillna(df['age'].median())  # Fill with median # Remove rows where critical fields are missing df = df.dropna(subset=['name', 'id']) return df.to_dict('records')

def standardize_data(data): """Convert data to consistent format""" for record in data: # Standardize names to title case if 'name' in record: record['name'] = record['name'].title().strip() # Standardize phone to 10 digits (India format) if 'phone' in record: phone = re.sub(r'D', '', record['phone']) record['phone'] = phone[-10:] if len(phone) >= 10 else phone # Standardize email if 'email' in record: record['email'] = record['email'].lower().strip() # Convert dates if 'date_of_birth' in record: try: record['date_of_birth'] = datetime.strptime( record['date_of_birth'], '%d-%m-%Y' ).isoformat() except: record['date_of_birth'] = None return data

def enrich_data(data): """Add calculated fields""" for record in data: # Calculate age from date of birth if 'date_of_birth' in record and record['date_of_birth']: try: dob = datetime.fromisoformat(record['date_of_birth']) age = (datetime.now() - dob).days // 365 record['age'] = age except: pass # Add processing timestamp record['processed_at'] = datetime.utcnow().isoformat() # Add data quality score score = 0 required_fields = ['name', 'email', 'phone', 'age'] for field in required_fields: if field in record and record[field]: score += 25 record['data_quality_score'] = score return data

def validate_data(data): """Check data meets requirements, return valid and invalid""" valid = [] invalid = [] for record in data: errors = [] # Validate email format if 'email' in record: if record['email'] and not re.match(r'^[w.-]+@[w.-]+.w+$', record['email']): errors.append('Invalid email') # Validate age if 'age' in record: try: age = int(record['age']) if age < 0 or age > 150: errors.append('Age out of range') except: errors.append('Age not numeric') if errors: record['validation_errors'] = errors invalid.append(record) else: valid.append(record) return valid, invalid

# Complete transform example
def transform_pipeline(raw_data): """Execute complete transformation""" print("Starting transformation...") # Step 1: Clean cleaned = clean_and_validate_data(raw_data) print(f"Cleaned data: {len(cleaned)} records") # Step 2: Standardize standardized = standardize_data(cleaned) # Step 3: Enrich enriched = enrich_data(standardized) # Step 4: Validate valid, invalid = validate_data(enriched) print(f"Valid: {len(valid)}, Invalid: {len(invalid)}") return valid, invalid

Stage 3: Load

def load_to_csv(data, file_path): """Save to CSV file""" if not data: return df = pd.DataFrame(data) df.to_csv(file_path, index=False, encoding='utf-8') print(f"Loaded {len(data)} records to {file_path}")

def load_to_json(data, file_path): """Save to JSON file""" with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) print(f"Loaded {len(data)} records to {file_path}")

def load_to_mongodb(data, connection_string, db_name, collection_name): """Save to MongoDB""" client = MongoClient(connection_string) db = client[db_name] collection = db[collection_name] if data: # Use insert_many for batch insert result = collection.insert_many(data) print(f"Loaded {len(result.inserted_ids)} records to MongoDB")

def load_to_postgres(data, connection_string, table_name): """Save to PostgreSQL""" import psycopg2 from io import StringIO conn = psycopg2.connect(connection_string) cursor = conn.cursor() # Convert to CSV-like format for COPY df = pd.DataFrame(data) # Create table if not exists (simplified) cursor.execute(f"TRUNCATE TABLE {table_name}") # Insert data for _, row in df.iterrows(): cursor.execute( f"INSERT INTO {table_name} VALUES (%s, %s, %s, ...)", tuple(row) ) conn.commit() cursor.close() conn.close() print(f"Loaded {len(data)} records to PostgreSQL")

Real Example: Election Results Pipeline (India)

class ElectionResultsPipeline: """ETL pipeline for Indian election results""" def __init__(self): self.raw_data = [] self.cleaned_data = [] self.results_db = 'election_results' def extract(self, election_commission_url): """Extract from Election Commission of India API""" print("Extracting election data...") # Fetch data from ECI response = requests.get(election_commission_url) self.raw_data = response.json()['results'] print(f"Extracted {len(self.raw_data)} constituencies") return self.raw_data def transform(self): """Transform raw election data""" print("Transforming election data...") transformed = [] for constituency in self.raw_data: # Extract and standardize record = { 'constituency_id': constituency['code'], 'constituency_name': constituency['name'].title(), 'state': constituency['state'], 'total_voters': int(constituency['total_voters']), 'voter_turnout': float(constituency['turnout_percent']), 'results': [] } # Process candidate results for candidate in constituency['candidates']: record['results'].append({ 'candidate_name': candidate['name'].title(), 'party': candidate['party'], 'votes': int(candidate['votes']), 'vote_share': float(candidate['vote_share']), 'status': candidate['status']  # Won/Lost/Trailing }) # Sort by votes (winner first) record['results'].sort(key=lambda x: x['votes'], reverse=True) # Add calculated fields total_votes_polled = sum(r['votes'] for r in record['results']) record['total_votes_polled'] = total_votes_polled record['winning_party'] = record['results'][0]['party'] record['winning_margin'] = ( record['results'][0]['votes'] - record['results'][1]['votes'] if len(record['results']) > 1 else 0 ) transformed.append(record) self.cleaned_data = transformed print(f"Transformed {len(transformed)} constituencies") return transformed def load(self, output_file): """Load to database/file""" print("Loading election results...") # Load to JSON with open(output_file, 'w') as f: json.dump(self.cleaned_data, f, indent=2) print(f"Loaded {len(self.cleaned_data)} results to {output_file}") def run_pipeline(self, api_url, output_file): """Execute complete ETL pipeline""" self.extract(api_url) self.transform() self.load(output_file)

# Usage
pipeline = ElectionResultsPipeline()
pipeline.run_pipeline( 'https://api.eci.gov.in/election/2024/results', 'election_results_2024.json'
)

Batch vs Streaming Pipelines

# BATCH PIPELINE - Process all data at once
def batch_pipeline(input_file, output_file, batch_size=1000): """Process data in batches""" data = extract_from_csv(input_file) # Process in batches for i in range(0, len(data), batch_size): batch = data[i:i+batch_size] transformed_batch = transform_pipeline(batch) load_to_csv(transformed_batch, f"{output_file}_part_{i}.csv")

# STREAMING PIPELINE - Process data continuously
from kafka import KafkaConsumer, KafkaProducer
import json

def streaming_pipeline(): """Process data as it arrives""" # Consume from Kafka topic consumer = KafkaConsumer( 'raw_data_topic', bootstrap_servers=['localhost:9092'], value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) # Produce to results topic producer = KafkaProducer( bootstrap_servers=['localhost:9092'], value_serializer=lambda v: json.dumps(v).encode('utf-8') ) for message in consumer: raw_record = message.value # Transform immediately cleaned = clean_and_validate_data([raw_record])[0] transformed = enrich_data([cleaned])[0] # Load immediately producer.send('results_topic', value=transformed)

Data Quality and Monitoring

class PipelineMonitor: def __init__(self): self.stats = { 'extracted': 0, 'valid': 0, 'invalid': 0, 'loaded': 0, 'errors': [] } def log_extraction(self, count): self.stats['extracted'] = count def log_validation(self, valid_count, invalid_count): self.stats['valid'] = valid_count self.stats['invalid'] = invalid_count def log_error(self, error_msg): self.stats['errors'].append({ 'timestamp': datetime.utcnow().isoformat(), 'message': error_msg }) def log_load(self, count): self.stats['loaded'] = count def get_report(self): success_rate = (self.stats['valid'] / self.stats['extracted'] * 100) if self.stats['extracted'] > 0 else 0 return { 'extracted': self.stats['extracted'], 'processed': self.stats['valid'] + self.stats['invalid'], 'loaded': self.stats['loaded'], 'success_rate': f"{success_rate:.2f}%", 'errors': len(self.stats['errors']) }

Practice Problems

1. Design an ETL pipeline to process Census 2021 data from CSV to MongoDB
2. Create a data cleaning function that handles 10 common data quality issues
3. Build a streaming pipeline using Kafka to process live election results
4. Write validation rules for a student enrollment database
5. Compare batch vs streaming approaches for stock market price updates

Key Takeaways

• ETL pipelines move data through Extract → Transform → Load stages
• Extraction retrieves data from diverse sources (APIs, files, databases)
• Transformation cleans, validates, standardizes, and enriches data
• Loading moves processed data to target systems efficiently
• Data quality is critical - validate and monitor at each stage
• Batch pipelines process accumulated data periodically
• Streaming pipelines process data continuously as it arrives
• Monitoring and error handling ensure pipeline reliability