Data Processing Pipelines

import pandas as pd

# EXTRACT: Read from source
df = pd.read_csv('sales_data.csv')
# Raw: 10,000 rows with duplicates, nulls, bad data

# TRANSFORM: Clean and enrich
df = df.drop_duplicates()               # Remove duplicates
df = df.dropna(subset=['customer_id'])  # Remove nulls
df['amount'] = df['amount'].abs()       # Fix negative amounts
df['date'] = pd.to_datetime(df['date']) # Parse dates
df = df[df['amount'] > 0]               # Filter valid transactions

# Enrich with customer data
customers = pd.read_sql("SELECT * FROM customers", db_conn)
df = df.merge(customers, on='customer_id', how='left')

# LOAD: Write cleaned data to warehouse
df.to_sql('clean_sales', warehouse_conn, if_exists='append')

-- EXTRACT & LOAD: Use Fivetran/Airbyte to load raw data
-- Raw table already exists in warehouse: raw_sales

-- TRANSFORM: dbt model (runs in warehouse)
-- models/staging/stg_sales.sql
WITH source AS (
    SELECT * FROM raw_sales
),

cleaned AS (
    SELECT DISTINCT
        customer_id,
        ABS(amount) AS amount,
        CAST(date AS DATE) AS date
    FROM source
    WHERE customer_id IS NOT NULL
      AND amount > 0
)

SELECT
    c.*,
    cust.name AS customer_name,
    cust.segment
FROM cleaned c
LEFT JOIN dim_customers cust
    ON c.customer_id = cust.customer_id

Time:     00:00        01:00        02:00        03:00
          │            │            │            │
Data:     ████████     ████████     ████████     ████████
          Collect      Collect      Collect      Collect
                       │
                       ▼
                    Process
                    All Data
                    From Last Hour

# Airflow DAG - runs daily at 2 AM
from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator
from datetime import datetime, timedelta

def extract_daily_sales():
    """Extract yesterday's sales"""
    query = """
        SELECT * FROM sales
        WHERE date = CURRENT_DATE - 1
    """
    df = pd.read_sql(query, source_db)
    df.to_parquet('/data/raw/sales_{{ ds }}.parquet')

def transform_and_load():
    """Transform and load to warehouse"""
    df = pd.read_parquet('/data/raw/sales_{{ ds }}.parquet')

    # Aggregate by customer
    daily_summary = df.groupby('customer_id').agg({
        'amount': 'sum',
        'quantity': 'sum'
    }).reset_index()

    daily_summary.to_sql('daily_sales_summary', warehouse,
                         if_exists='append')

dag = DAG(
    'daily_sales_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule='0 2 * * *',  # 2 AM daily
)

extract = PythonOperator(task_id='extract', python_callable=extract_daily_sales, dag=dag)
transform = PythonOperator(task_id='transform', python_callable=transform_and_load, dag=dag)

extract >> transform

Time:     00:00:00  00:00:01  00:00:02  00:00:03
          │         │         │         │
Data:     █ ──► Process
             █ ──► Process
                █ ──► Process
                   █ ──► Process

          Continuous flow, each event processed immediately

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'sales_events',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

# Process each event as it arrives
for message in consumer:
    event = message.value

    # Transform in real-time
    processed = {
        'customer_id': event['customer_id'],
        'amount': abs(float(event['amount'])),
        'timestamp': event['timestamp'],
        'category': categorize_purchase(event['amount'])
    }

    # Write to real-time dashboard database
    write_to_db(processed)

    # Check for fraud in real-time
    if event['amount'] > 10000:
        send_fraud_alert(event)