Batch vs Streaming: Lambda, Kappa, and Architecture Trade-offs

Batch vs Streaming: Choosing the Right Processing Paradigm

The fundamental question in data engineering is whether to process data in batches or as a continuous stream.

The Batch-Streaming Spectrum

Processing Paradigms:

Key Considerations:

1. Latency requirements — how fast do you need results?
2. Data volume — how much data are you processing?
3. Consistency needs — exact-once or at-least-once?
4. Operational maturity — can your team manage streaming infrastructure?

Key Insight: The choice is not binary but a design decision based on latency requirements, data volume, consistency needs, and operational maturity. Micro-batch systems (Spark Structured Streaming, Flink with mini-batch) occupy the middle ground.

Lambda vs Kappa Architecture

Processing Spectrum

Architecture Diagram

Batch vs Streaming Comparison

Production Code

Lambda Architecture: Batch + Streaming Merge

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)


class LambdaArchitecture:
    """
    Lambda Architecture implementation: batch layer + speed layer + serving layer.
    """

    def __init__(self, spark: SparkSession):
        self.spark = spark

    # ------------------------------------------------------
    # BATCH LAYER: Comprehensive, accurate results
    # ------------------------------------------------------
    def batch_layer(self, input_path: str, batch_output_path: str, execution_date: str):
        """Run batch processing for a complete dataset."""
        logger.info(f"Running batch layer for {execution_date}")

        # Read complete dataset for the day
        raw_df = (
            self.spark.read
            .parquet(f"{input_path}/date={execution_date}")
        )

        # Comprehensive aggregation
        batch_result = (
            raw_df
            .groupBy("product_id", "region")
            .agg(
                F.count("*").alias("total_orders"),
                F.sum("amount").alias("total_revenue"),
                F.countDistinct("customer_id").alias("unique_customers"),
                F.avg("amount").alias("avg_order_value"),
                F.percentile_approx("amount", 0.5).alias("median_order_value"),
            )
        )

        # Write batch view
        (
            batch_result
            .write
            .mode("overwrite")
            .parquet(f"{batch_output_path}/date={execution_date}")
        )

        logger.info(f"Batch layer completed for {execution_date}")
        return batch_result

    # ------------------------------------------------------
    # SPEED LAYER: Real-time approximations
    # ------------------------------------------------------
    def speed_layer(self, kafka_bootstrap: str, topic: str, speed_output_path: str):
        """Run streaming processing for real-time approximations."""
        logger.info("Starting speed layer streaming query")

        raw_stream = (
            self.spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", kafka_bootstrap)
            .option("subscribe", topic)
            .option("startingOffsets", "latest")
            .load()
        )

        # Parse and aggregate
        parsed = (
            raw_stream
            .select(F.from_json(F.col("value").cast("string"), self._get_schema()).alias("data"))
            .select("data.*")
            .withWatermark("event_time", "5 minutes")
            .groupBy(
                F.window("event_time", "5 minutes"),
                "product_id",
                "region",
            )
            .agg(
                F.count("*").alias("recent_orders"),
                F.sum("amount").alias("recent_revenue"),
            )
        )

        query = (
            parsed.writeStream
            .format("parquet")
            .option("path", speed_output_path)
            .option("checkpointLocation", f"{speed_output_path}/_checkpoint")
            .trigger(processingTime="1 minute")
            .outputMode("append")
            .start()
        )

        return query

    # ------------------------------------------------------
    # SERVING LAYER: Merge batch + speed views
    # ------------------------------------------------------
    def serving_layer(
        self,
        batch_output_path: str,
        speed_output_path: str,
        merged_output_path: str,
        execution_date: str,
    ):
        """Merge batch and speed views into a unified serving view."""
        logger.info("Merging batch and speed views")

        # Read batch view (complete, accurate)
        batch_view = (
            self.spark.read
            .parquet(f"{batch_output_path}/date={execution_date}")
        )

        # Read speed view (recent, approximate)
        speed_view = (
            self.spark.read
            .parquet(speed_output_path)
            .filter(F.col("window_start") >= F.lit(execution_date))
        )

        # Merge: prefer batch for historical, speed for recent
        merged = (
            batch_view.alias("b")
            .join(
                speed_view.alias("s"),
                (F.col("b.product_id") == F.col("s.product_id")) &
                (F.col("b.region") == F.col("s.region")),
                "full_outer",
            )
            .select(
                F.coalesce(F.col("s.product_id"), F.col("b.product_id")).alias("product_id"),
                F.coalesce(F.col("s.region"), F.col("b.region")).alias("region"),
                F.coalesce(F.col("s.recent_orders"), F.col("b.total_orders")).alias("total_orders"),
                F.coalesce(F.col("s.recent_revenue"), F.col("b.total_revenue")).alias("total_revenue"),
            )
        )

        (
            merged.write
            .mode("overwrite")
            .parquet(f"{merged_output_path}/date={execution_date}")
        )

        logger.info(f"Serving layer merged for {execution_date}")
        return merged

    def _get_schema(self):
        from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType
        return StructType([
            StructField("product_id", StringType()),
            StructField("region", StringType()),
            StructField("amount", DoubleType()),
            StructField("event_time", TimestampType()),
        ])

Kappa Architecture: Replay-Based Reprocessing

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
import logging

logger = logging.getLogger(__name__)


class KappaArchitecture:
    """
    Kappa Architecture: process all data through streaming, reprocess by replaying.
    """

    def __init__(self, spark: SparkSession):
        self.spark = spark

    def stream_processor(self, kafka_bootstrap: str, topic: str, output_path: str, version: str):
        """
        Deploy a streaming processor. To reprocess, deploy a new version
        and replay from the beginning of the Kafka topic.
        """
        logger.info(f"Deploying stream processor version: {version}")

        # Read from Kafka (replayable log)
        raw_stream = (
            self.spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", kafka_bootstrap)
            .option("subscribe", topic)
            .option("startingOffsets", "earliest")  # For reprocessing
            .option("group.id", f"processor-{version}")
            .load()
        )

        # Version-specific transformation logic
        processed = self._apply_transformations(raw_stream, version)

        # Write to versioned output
        query = (
            processed.writeStream
            .format("parquet")
            .option("path", f"{output_path}/version={version}")
            .option("checkpointLocation", f"{output_path}/_checkpoint/{version}")
            .trigger(processingTime="5 minutes")
            .outputMode("append")
            .start()
        )

        return query

    def _apply_transformations(self, stream, version: str):
        """Apply version-specific transformations."""
        parsed = (
            stream
            .select(F.from_json(F.col("value").cast("string"), self._get_schema()).alias("data"))
            .select("data.*")
        )

        if version == "v1":
            # Original logic
            return parsed.select(
                "event_id", "user_id", "amount",
                F.col("timestamp").alias("event_time"),
            )

        elif version == "v2":
            # Updated logic: added fraud detection
            return parsed.select(
                "event_id", "user_id", "amount",
                F.col("timestamp").alias("event_time"),
            ).withColumn(
                "is_fraud",
                F.when(F.col("amount") > 10000, True).otherwise(False),
            )

        return parsed

    def reprocess(self, kafka_bootstrap: str, topic: str, output_path: str, new_version: str):
        """
        Reprocess historical data by deploying a new version and replaying.
        Old version continues serving until new version catches up.
        """
        logger.info(f"Starting reprocessing with version {new_version}")

        # Deploy new version (reads from beginning of topic)
        new_query = self.stream_processor(
            kafka_bootstrap, topic, output_path, new_version
        )

        # Wait for new version to catch up to current time
        # (monitored via consumer lag or watermark)
        logger.info("New version deployed. Monitor consumer lag for catch-up completion.")

        return new_query

    def _get_schema(self):
        from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType
        return StructType([
            StructField("event_id", StringType()),
            StructField("user_id", StringType()),
            StructField("amount", DoubleType()),
            StructField("timestamp", TimestampType()),
        ])

Best Practices

1. Start with batch if your team is new to data engineering. Add streaming only when latency requirements demand it.
2. Use micro-batch streaming (Spark Structured Streaming) as a middle ground — near-real-time with batch-like simplicity.
3. Choose Kappa over Lambda when possible. Replayable event logs (Kafka) eliminate the need for dual codebases.
4. If using Lambda, share transformation logic between batch and speed layers via a common library to prevent code divergence.
5. Set appropriate watermarks in streaming systems to bound state growth while maintaining completeness.
6. Monitor consumer lag as a key SLA metric. Lag exceeding retention period = data loss.
7. Design for reprocessing from day one. Store raw data in replayable formats (Kafka, append-only Parquet).
8. Use idempotent processing so that reprocessing (batch re-run or stream replay) produces identical results.
9. Document the latency-completeness trade-off for each pipeline and get stakeholder agreement on acceptable approximations.
10. Evaluate operational cost — streaming requires 24/7 monitoring, state management, and expertise that batch does not.

Architecture Pattern Comparison

See Also

• 022 - Spark Structured Streaming - Micro-batch streaming implementation
• 019 - Apache Kafka: Topics, Producers, and Consumers - Event log for Kappa architecture
• 021 - Apache Spark: RDDs, DataFrames, and the Catalyst Optimizer - Batch processing engine
• 024 - Data Ingestion Patterns - Ingestion for both paradigms
• 016 - ETL vs ELT - Transformation paradigm selection