Batch vs Streaming

# Load entire historical dataset - typical batch approach
import pandas as pd
import numpy as np

# Read 12 months of e-commerce data at once
df = pd.read_csv('dataplexa_ecommerce.csv')

# Check the size of our batch dataset
print(f"Total records: {len(df):,}")
print(f"Date range: {df['date'].min()} to {df['date'].max()}")

# Process entire year's data to find seasonal trends
df['date'] = pd.to_datetime(df['date'])
df['month'] = df['date'].dt.month

# Aggregate revenue by month - batch calculation
monthly_revenue = df.groupby('month')['revenue'].sum() / 100000

# Show top 3 revenue months
top_months = monthly_revenue.nlargest(3)
print("Top revenue months (₹ Lakhs):")
print(top_months)

# Simulate streaming transaction processing
from datetime import datetime
import time

# Function to process each transaction as it arrives
def process_streaming_transaction(order_data):
    # Check transaction in real-time - no waiting for batch
    transaction_time = datetime.now()
    
    # Flag high-risk transactions immediately
    if order_data['revenue'] > 50000:  # Transactions above ₹50k
        risk_score = "HIGH"
    else:
        risk_score = "LOW"
        
    return {
        'order_id': order_data['order_id'],
        'processed_at': transaction_time,
        'risk_score': risk_score
    }

# Simulate 5 incoming transactions - streaming style
sample_transactions = [
    {'order_id': 12001, 'revenue': 75000},  # High risk
    {'order_id': 12002, 'revenue': 2500},   # Low risk  
    {'order_id': 12003, 'revenue': 125000}, # High risk
    {'order_id': 12004, 'revenue': 890},    # Low risk
    {'order_id': 12005, 'revenue': 67000}   # High risk
]

# Process each transaction immediately as it arrives
for transaction in sample_transactions:
    result = process_streaming_transaction(transaction)
    print(f"Order {result['order_id']}: {result['risk_score']} risk")
    time.sleep(0.05)  # Simulate small processing delay

# Hybrid approach - batch + streaming for different needs
batch_cost_per_gb = 0.50   # ₹0.50 per GB processed
stream_cost_per_hour = 25  # ₹25 per hour for streaming infrastructure

# Calculate cost for processing 100GB of transaction data
daily_data_volume = 100  # GB

# Batch: Process once at night
batch_daily_cost = daily_data_volume * batch_cost_per_gb
print(f"Batch processing cost: ₹{batch_daily_cost} per day")

# Streaming: Process 24/7
stream_daily_cost = 24 * stream_cost_per_hour  
print(f"Stream processing cost: ₹{stream_daily_cost} per day")

# Cost difference
cost_difference = stream_daily_cost - batch_daily_cost
print(f"Streaming costs ₹{cost_difference} more per day ({cost_difference/batch_daily_cost*100:.0f}% increase)")

# Decision framework - categorize your use cases
use_cases = [
    {"task": "Daily sales report", "max_delay": "24 hours", "recommended": "batch"},
    {"task": "Fraud detection", "max_delay": "100 ms", "recommended": "stream"},
    {"task": "Customer segmentation", "max_delay": "1 week", "recommended": "batch"},
    {"task": "Product recommendations", "max_delay": "2 seconds", "recommended": "stream"},
    {"task": "Inventory planning", "max_delay": "1 day", "recommended": "batch"}
]

# Categorize by processing type
batch_tasks = [task for task in use_cases if task["recommended"] == "batch"]
stream_tasks = [task for task in use_cases if task["recommended"] == "stream"]

print(f"Batch processing tasks: {len(batch_tasks)}")
print(f"Stream processing tasks: {len(stream_tasks)}")

print("\nBatch tasks:")
for task in batch_tasks:
    print(f"- {task['task']} (delay OK: {task['max_delay']})")
    
print("\nStream tasks:")  
for task in stream_tasks:
    print(f"- {task['task']} (delay OK: {task['max_delay']})")