Lambda & Kappa Architecture

# Batch Layer: Recompute views from all historical data
from pyspark.sql import SparkSession
from datetime import datetime

class BatchLayer:
    """Processes all historical page views to compute accurate counts"""

    def __init__(self):
        self.spark = SparkSession.builder.appName("BatchLayer").getOrCreate()

    def compute_batch_views(self, start_date, end_date):
        """
        Run nightly: Reads all page view logs from S3 and computes
        accurate page view counts per URL per day.
        """
        # Read raw data from immutable storage (S3/HDFS)
        df = self.spark.read.parquet(f"s3://data-lake/page_views/{start_date}/*")

        # Compute aggregations
        batch_view = df.groupBy("url", "date").count()

        # Write to serving layer (could be Cassandra, DynamoDB, etc.)
        batch_view.write \
            .format("org.apache.spark.sql.cassandra") \
            .options(table="page_views_batch", keyspace="analytics") \
            .mode("append") \
            .save()

        print(f"Batch view computed for {start_date} to {end_date}")
        return batch_view

# Run as nightly batch job
batch_layer = BatchLayer()
batch_layer.compute_batch_views("2024-01-01", "2024-01-31")

# Speed Layer: Process streaming data for real-time updates
import json
import boto3
from collections import defaultdict
from datetime import datetime, timedelta, timezone

class SpeedLayer:
    """Processes real-time page views to fill the gap until batch catches up"""

    def __init__(self):
        self.kinesis = boto3.client('kinesis')
        self.dynamodb = boto3.resource('dynamodb')
        self.realtime_views = self.dynamodb.Table('page_views_realtime')

        # In-memory state (could use Redis for distributed scenarios)
        self.window_counts = defaultdict(int)

    def process_stream(self):
        """
        Consume Kinesis stream: Update real-time counts as events arrive.
        These counts are temporary and replaced by batch views.
        """
        response = self.kinesis.get_records(
            ShardIterator=self.get_shard_iterator(),
            Limit=1000
        )

        for record in response['Records']:
            event = json.loads(record['Data'])
            self.process_event(event)

    def process_event(self, event):
        """Increment real-time counter"""
        url = event['url']
        timestamp = datetime.fromisoformat(event['timestamp'])

        # Only count recent events (last 24 hours)
        if datetime.now(timezone.utc) - timestamp < timedelta(hours=24):
            key = f"{url}#{timestamp.date()}"
            self.window_counts[key] += 1

            # Write to DynamoDB for serving layer
            self.realtime_views.update_item(
                Key={'url_date': key},
                UpdateExpression='ADD view_count :inc',
                ExpressionAttributeValues={':inc': 1}
            )

    def cleanup_old_data(self):
        """Remove data older than 24h (batch layer has it now)"""
        cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
        old_keys = [k for k in self.window_counts.keys()
                    if datetime.fromisoformat(k.split('#')[1]) < cutoff.date()]
        for key in old_keys:
            del self.window_counts[key]

# Run continuously
speed_layer = SpeedLayer()
while True:
    speed_layer.process_stream()
    speed_layer.cleanup_old_data()

# Serving Layer: Merge batch and real-time views at query time
import boto3
from datetime import datetime, timedelta, timezone

class ServingLayer:
    """Combines batch views (historical) with real-time views (recent)"""

    def __init__(self):
        self.dynamodb = boto3.resource('dynamodb')
        self.batch_table = self.dynamodb.Table('page_views_batch')
        self.realtime_table = self.dynamodb.Table('page_views_realtime')

    def get_page_views(self, url, start_date, end_date):
        """
        Query API: Returns total page views by merging:
        - Batch views (complete, up to yesterday)
        - Real-time views (approximate, last 24 hours)
        """
        total_views = 0
        realtime_views = 0
        yesterday = datetime.now(timezone.utc).date() - timedelta(days=1)

        # 1. Get batch views (older than 24 hours)
        batch_views = self.query_batch_views(url, start_date, min(end_date, yesterday))
        total_views += batch_views

        # 2. Get real-time views (last 24 hours)
        if end_date > yesterday:
            realtime_views = self.query_realtime_views(url)
            total_views += realtime_views

        return {
            'url': url,
            'total_views': total_views,
            'batch_views': batch_views,
            'realtime_views': realtime_views,
            'as_of': datetime.now(timezone.utc).isoformat()
        }

    def query_batch_views(self, url, start_date, end_date):
        """Query Cassandra/DynamoDB for batch views"""
        response = self.batch_table.query(
            KeyConditionExpression='url = :url AND date BETWEEN :start AND :end',
            ExpressionAttributeValues={
                ':url': url,
                ':start': start_date.isoformat(),
                ':end': end_date.isoformat()
            }
        )
        return sum(item['view_count'] for item in response['Items'])

    def query_realtime_views(self, url):
        """Query DynamoDB for real-time views"""
        today = datetime.now(timezone.utc).date()
        key = f"{url}#{today}"
        response = self.realtime_table.get_item(Key={'url_date': key})
        return response.get('Item', {}).get('view_count', 0)

# API Usage
serving_layer = ServingLayer()
result = serving_layer.get_page_views(
    url="/product/123",
    start_date=datetime(2024, 1, 1),
    end_date=datetime.now(timezone.utc)
)

print(f"Total views: {result['total_views']}")
print(f"  Batch (complete): {result['batch_views']}")
print(f"  Real-time (last 24h): {result['realtime_views']}")

# Kappa Architecture: Single streaming job processes everything
from kafka import KafkaConsumer
import boto3
from collections import defaultdict
from datetime import datetime
import json

class KappaStreamProcessor:
    """
    Single streaming job that processes ALL page views.
    Can run in real-time mode OR replay mode for reprocessing.
    """

    def __init__(self, mode='realtime'):
        self.mode = mode
        self.consumer = KafkaConsumer(
            'page-views',
            bootstrap_servers=['localhost:9092'],
            group_id=f'page-view-processor-{mode}',
            auto_offset_reset='earliest' if mode == 'replay' else 'latest',
            enable_auto_commit=True
        )

        self.dynamodb = boto3.resource('dynamodb')
        self.views_table = self.dynamodb.Table('page_views')

        # State store (checkpointing for fault tolerance)
        self.state = defaultdict(int)

    def process_stream(self):
        """
        Main processing loop: Works for both real-time and replay.
        Same code, same logic, different starting offset.
        """
        print(f"Starting stream processor in {self.mode} mode...")

        for message in self.consumer:
            event = json.loads(message.value.decode('utf-8'))
            self.process_event(event)

            # Checkpoint periodically for fault tolerance
            if message.offset % 1000 == 0:
                self.checkpoint()

    def process_event(self, event):
        """Process a single page view event"""
        url = event['url']
        timestamp = datetime.fromisoformat(event['timestamp'])
        date_key = timestamp.date().isoformat()

        # Update in-memory state
        key = f"{url}#{date_key}"
        self.state[key] += 1

        # Update materialized view (DynamoDB)
        self.views_table.update_item(
            Key={'url_date': key},
            UpdateExpression='ADD view_count :inc',
            ExpressionAttributeValues={':inc': 1},
            ReturnValues='NONE'
        )

    def checkpoint(self):
        """Save state for fault tolerance"""
        # In production, use RocksDB or managed state store
        pass

# Real-time processing
realtime_processor = KappaStreamProcessor(mode='realtime')
realtime_processor.process_stream()

# Reprocessing (when logic changes)
replay_processor = KappaStreamProcessor(mode='replay')
replay_processor.process_stream()  # Reads from beginning

# Reprocessing in Kappa: Deploy new version, replay stream

class ReprocessingManager:
    """
    Manages reprocessing when business logic changes.
    Strategy: Blue-green deployment of streaming jobs.
    """

    def deploy_new_version(self, version):
        """
        1. Deploy new streaming job (v2)
        2. Replay Kafka from beginning into new table
        3. Switch traffic to new table once caught up
        4. Decommission old job (v1)
        """
        print(f"Deploying version {version}...")

        # Step 1: Create new materialized view table
        new_table = self.create_table(f"page_views_v{version}")

        # Step 2: Start new streaming job reading from offset 0
        new_job = KappaStreamProcessor(mode='replay')
        new_job.views_table = new_table
        new_job.process_stream()  # Processes entire history

        # Step 3: Wait until caught up to current time
        while not self.is_caught_up(new_job):
            time.sleep(60)

        # Step 4: Switch serving layer to new table (atomic swap)
        self.update_serving_layer(new_table)

        # Step 5: Decommission old job
        self.stop_old_job()

        print(f"Version {version} deployed successfully!")

    def is_caught_up(self, job):
        """Check if replay has processed up to current time"""
        lag = job.consumer.metrics()['consumer-lag']
        return lag < 1000  # Less than 1000 messages behind

# Usage: Reprocess when logic changes
manager = ReprocessingManager()
manager.deploy_new_version(version=2)  # Replays entire stream