⚡🌊 Lambda-Kappa Hybrid Architecture¶

from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp

spark = SparkSession.builder \
    .appName("BatchProcessing") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Read from raw zone
raw_data = spark.read.format("delta").load("/mnt/datalake/raw/transactions")

# Apply transformations
refined_data = raw_data \
    .filter(col("status") == "completed") \
    .withColumn("processed_timestamp", current_timestamp()) \
    .select("transaction_id", "customer_id", "amount", "transaction_date", "processed_timestamp")

# Write to refined zone with merge
delta_table = DeltaTable.forPath(spark, "/mnt/datalake/refined/transactions")

delta_table.alias("target") \
    .merge(
        refined_data.alias("source"),
        "target.transaction_id = source.transaction_id"
    ) \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

from delta.tables import DeltaTable
from pyspark.sql.functions import col, when

# Stream processing output
stream_path = "/mnt/datalake/stream/user_events"
# Batch processing output
batch_path = "/mnt/datalake/batch/user_events"
# Unified output
unified_path = "/mnt/datalake/unified/user_events"

# Read both sources
stream_df = spark.read.format("delta").load(stream_path)
batch_df = spark.read.format("delta").load(batch_path)

# Create or get unified table
if DeltaTable.isDeltaTable(spark, unified_path):
    unified_table = DeltaTable.forPath(spark, unified_path)
else:
    stream_df.write.format("delta").save(unified_path)
    unified_table = DeltaTable.forPath(spark, unified_path)

# Merge stream data
unified_table.alias("target") \
    .merge(
        stream_df.alias("source"),
        "target.event_id = source.event_id"
    ) \
    .whenMatchedUpdate(
        condition = "source.event_timestamp > target.event_timestamp",
        set = {
            "event_type": "source.event_type",
            "user_id": "source.user_id",
            "event_timestamp": "source.event_timestamp",
            "source_system": "lit('stream')"
        }
    ) \
    .whenNotMatchedInsert(
        values = {
            "event_id": "source.event_id",
            "event_type": "source.event_type",
            "user_id": "source.user_id",
            "event_timestamp": "source.event_timestamp",
            "source_system": "lit('stream')"
        }
    ) \
    .execute()

# Merge batch data
unified_table.alias("target") \
    .merge(
        batch_df.alias("source"),
        "target.event_id = source.event_id"
    ) \
    .whenNotMatchedInsert(
        values = {
            "event_id": "source.event_id",
            "event_type": "source.event_type",
            "user_id": "source.user_id",
            "event_timestamp": "source.event_timestamp",
            "source_system": "lit('batch')"
        }
    ) \
    .execute()

from pyspark.sql import DataFrame
from typing import Dict, Any

class WorkloadRouter:
    """Route data processing based on workload characteristics."""

    def __init__(self, spark):
        self.spark = spark
        self.routing_rules = {
            "batch": {
                "min_volume": 1000000,
                "max_latency_minutes": 60,
                "compute_type": "spark_batch"
            },
            "stream": {
                "max_latency_seconds": 30,
                "compute_type": "stream_analytics"
            },
            "adhoc": {
                "max_volume": 100000,
                "compute_type": "serverless_sql"
            }
        }

    def route_workload(self, df: DataFrame, metadata: Dict[str, Any]) -> str:
        """Determine appropriate processing path."""
        row_count = df.count()
        latency_requirement = metadata.get("latency_requirement_seconds", 3600)

        # Stream processing for low latency
        if latency_requirement < 60:
            return "stream"

        # Ad-hoc processing for small datasets
        if row_count < 100000:
            return "adhoc"

        # Batch processing for large datasets
        return "batch"

    def process(self, df: DataFrame, metadata: Dict[str, Any]):
        """Process data using appropriate engine."""
        route = self.route_workload(df, metadata)

        if route == "batch":
            return self._process_batch(df, metadata)
        elif route == "stream":
            return self._process_stream(df, metadata)
        else:
            return self._process_adhoc(df, metadata)

    def _process_batch(self, df: DataFrame, metadata: Dict[str, Any]):
        """Batch processing with Spark."""
        output_path = metadata.get("output_path")

        df.write \
            .format("delta") \
            .mode("append") \
            .partitionBy("date") \
            .option("mergeSchema", "true") \
            .save(output_path)

        return {"status": "success", "route": "batch", "records": df.count()}

    def _process_stream(self, df: DataFrame, metadata: Dict[str, Any]):
        """Stream processing with Structured Streaming."""
        output_path = metadata.get("output_path")
        checkpoint_path = metadata.get("checkpoint_path")

        query = df.writeStream \
            .format("delta") \
            .outputMode("append") \
            .option("checkpointLocation", checkpoint_path) \
            .start(output_path)

        return {"status": "streaming", "route": "stream", "query_id": query.id}

    def _process_adhoc(self, df: DataFrame, metadata: Dict[str, Any]):
        """Ad-hoc processing for exploration."""
        df.createOrReplaceTempView("adhoc_view")
        return {"status": "success", "route": "adhoc", "view": "adhoc_view"}

# Usage
router = WorkloadRouter(spark)

# Example: Large batch workload
large_df = spark.read.parquet("/mnt/data/large_dataset")
result = router.process(large_df, {
    "latency_requirement_seconds": 3600,
    "output_path": "/mnt/datalake/batch/output"
})

# Example: Real-time stream
stream_df = spark.readStream.format("delta").load("/mnt/datalake/stream/input")
result = router.process(stream_df, {
    "latency_requirement_seconds": 5,
    "output_path": "/mnt/datalake/stream/output",
    "checkpoint_path": "/mnt/checkpoints/stream1"
})

from pyspark.sql.types import StructType, StructField, StringType, TimestampType, DoubleType

# Define unified schema for batch and stream
unified_schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("event_type", StringType(), False),
    StructField("user_id", StringType(), False),
    StructField("event_timestamp", TimestampType(), False),
    StructField("amount", DoubleType(), True),
    StructField("source_system", StringType(), False),
    StructField("processing_timestamp", TimestampType(), False)
])

# Use schema evolution with Delta Lake
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")

# Monitor batch and stream processing health
from azure.monitor.query import LogsQueryClient
from datetime import timedelta

def monitor_hybrid_pipeline():
    """Monitor both batch and stream processing."""
    metrics = {
        "batch": {
            "last_run": None,
            "records_processed": 0,
            "duration_minutes": 0,
            "status": "unknown"
        },
        "stream": {
            "current_lag": 0,
            "records_per_second": 0,
            "error_rate": 0.0,
            "status": "unknown"
        }
    }

    # Check batch processing
    batch_df = spark.sql("""
        SELECT MAX(processing_timestamp) as last_run,
               COUNT(*) as records_processed
        FROM delta.`/mnt/datalake/refined/batch/transactions`
        WHERE date = current_date()
    """)
    metrics["batch"].update(batch_df.first().asDict())

    # Check stream processing
    stream_df = spark.sql("""
        SELECT COUNT(*) as current_lag
        FROM delta.`/mnt/datalake/stream/input`
        WHERE processing_timestamp IS NULL
    """)
    metrics["stream"]["current_lag"] = stream_df.first()["current_lag"]

    return metrics

# Optimize batch processing with partitioning and Z-ordering
from delta.tables import DeltaTable

# Configure Spark for batch performance
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

# Write with optimal partitioning
df.write \
    .format("delta") \
    .mode("append") \
    .partitionBy("date", "region") \
    .save("/mnt/datalake/refined/transactions")

# Optimize table layout
delta_table = DeltaTable.forPath(spark, "/mnt/datalake/refined/transactions")
delta_table.optimize().executeZOrderBy("customer_id")

# Configure streaming for optimal performance
from pyspark.sql.streaming import DataStreamWriter

stream_query = spark.readStream \
    .format("delta") \
    .option("maxFilesPerTrigger", 1000) \
    .option("ignoreDeletes", "true") \
    .load("/mnt/datalake/stream/input") \
    .writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/checkpoints/stream1") \
    .trigger(processingTime='30 seconds') \
    .start("/mnt/datalake/stream/output")

# Create operational metrics dashboard
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def create_hybrid_dashboard(batch_metrics, stream_metrics):
    """Create operational dashboard for hybrid architecture."""

    fig = make_subplots(
        rows=2, cols=2,
        subplot_titles=('Batch Throughput', 'Stream Latency',
                       'Processing Costs', 'Error Rates')
    )

    # Batch throughput
    fig.add_trace(
        go.Scatter(x=batch_metrics['timestamps'],
                  y=batch_metrics['records_per_hour'],
                  name='Batch Throughput'),
        row=1, col=1
    )

    # Stream latency
    fig.add_trace(
        go.Scatter(x=stream_metrics['timestamps'],
                  y=stream_metrics['latency_ms'],
                  name='Stream Latency'),
        row=1, col=2
    )

    # Processing costs
    fig.add_trace(
        go.Bar(x=['Batch', 'Stream'],
              y=[batch_metrics['cost'], stream_metrics['cost']],
              name='Costs'),
        row=2, col=1
    )

    # Error rates
    fig.add_trace(
        go.Scatter(x=batch_metrics['timestamps'],
                  y=batch_metrics['error_rate'],
                  name='Batch Errors'),
        row=2, col=2
    )

    fig.update_layout(height=800, showlegend=True)
    return fig

# Migration script for batch workloads
def migrate_batch_pipeline(source_config, target_config):
    """Migrate existing batch pipeline to hybrid architecture."""

    # 1. Extract from legacy system
    legacy_df = spark.read.jdbc(
        url=source_config['jdbc_url'],
        table=source_config['table_name'],
        properties=source_config['properties']
    )

    # 2. Transform to Delta format
    legacy_df.write \
        .format("delta") \
        .mode("overwrite") \
        .partitionBy("date") \
        .save(target_config['raw_path'])

    # 3. Apply transformations
    transformed_df = spark.read.format("delta").load(target_config['raw_path'])
    transformed_df = apply_business_rules(transformed_df)

    # 4. Write to refined zone
    transformed_df.write \
        .format("delta") \
        .mode("append") \
        .save(target_config['refined_path'])

    return {"status": "migrated", "record_count": transformed_df.count()}

# Add streaming capabilities
def add_stream_processing(stream_config):
    """Add stream processing to existing batch pipeline."""

    # 1. Set up stream ingestion
    stream_df = spark.readStream \
        .format("eventhubs") \
        .options(**stream_config['eventhub_config']) \
        .load()

    # 2. Apply same transformations as batch
    stream_transformed = apply_business_rules(stream_df)

    # 3. Write to stream output
    query = stream_transformed.writeStream \
        .format("delta") \
        .outputMode("append") \
        .option("checkpointLocation", stream_config['checkpoint']) \
        .start(stream_config['output_path'])

    return query