Batch vs Streaming vs Micro-Batch

Batch Processing — The Foundation

T=06:00  Pipeline starts (triggered by cron or Airflow)
         Run parameters: run_date = 2026-03-17

T=06:01  Extract phase:
         Read all orders WHERE date = '2026-03-17'
         → 48,234 rows fetched from PostgreSQL replica
         → Written to S3 Bronze as Parquet

T=06:08  Transform phase (dbt):
         dbt run --select models/silver/orders.sql
         → 48,234 rows cleaned and typed, 47 rows rejected (to DLQ)
         → Written to silver.orders partition date=2026-03-17

T=06:14  Aggregate phase (dbt Gold):
         dbt run --select models/gold/daily_revenue.sql
         → Computes SUM, COUNT, AVG per store per category

T=06:17  PIPELINE EXITS — run complete. Duration: 17 minutes.
         Next run: T+24h at 06:00

WHAT HAPPENS BETWEEN T=06:17 AND T=06:00 NEXT DAY:
The pipeline does not exist as a running process — no compute resources
are consumed. Data written between runs accumulates, waiting for next
batch. A customer who placed an order at 08:00 PM won't see it in
analytics until 06:17 AM the NEXT day (22+ hour latency).

THIS IS FINE when the business question is "what was yesterday's revenue?"
THIS IS NOT FINE when the question is "is there fraud happening RIGHT NOW?"

Streaming Processing — Continuous, Event-Driven

EVENT TIME:      When the event actually happened (timestamp in the payload)
PROCESSING TIME: When the streaming system processed the event

These diverge whenever: the network is slow, the device was offline
(mobile app buffered events, flushed on reconnect), the system was under
load (consumer fell behind), or the event source has retries.

EXAMPLE:
  A FreshCart delivery agent marks an order "delivered" at 11:58 PM on a
  bad network connection. The event reaches Kafka at 12:03 AM.
    Event time:      2026-03-17 23:58:00  (when the tap happened)
    Processing time: 2026-03-18 00:03:00  (when Kafka received it)

  If your pipeline counts "deliveries on 2026-03-17":
    Using event time:      counts this delivery correctly for March 17
    Using processing time: counts this delivery for March 18 — WRONG

  ALWAYS use event time for business metrics. Only use processing time
  for system metrics (consumer lag, throughput).

TUMBLING WINDOWS (non-overlapping, fixed size):
  [00:00–01:00] [01:00–02:00] [02:00–03:00]
  Each event belongs to exactly one window.
  Use for: hourly/daily aggregates, session-independent metrics.

SLIDING WINDOWS (overlapping, fixed size, moves by step):
  [00:00–01:00] [00:30–01:30] [01:00–02:00]  (30-min step)
  Each event may belong to multiple windows.
  Use for: moving averages, rolling metrics.

SESSION WINDOWS (variable size, bounded by inactivity gap):
  [events...gap > 30min...][events...gap > 30min...][events...]
  Window closes when no events arrive for the gap duration.
  Use for: user session analysis, visit duration.

A watermark is the streaming system's current estimate of the maximum
event time it has seen, minus an allowed lateness:

  watermark = max_event_time_seen - allowed_lateness

Purpose: tell the system when it is safe to close a window and produce a
result, even if late events might still arrive.

Example:
  allowed_lateness = 5 minutes
  max event time seen = 23:58:00 → watermark = 23:53:00

  Window [23:00–24:00] is not closed yet — events up to 5 min late may
  still arrive. Window [22:00–23:00] IS closed — no events older than
  23:53 can legitimately arrive.

  Too small watermark → windows close early → late events dropped
  Too large watermark → windows close late → higher latency, more memory

EVENT SOURCE          MESSAGE BROKER      STREAM PROCESSOR     SINK
Payment service  →    Kafka topic:    →   Flink/Spark      →  Cassandra
(produces events)     payments.v1         Streaming           (real-time store)
                                                           →  S3 Parquet
                                                              (data lake landing)

KAFKA CONSUMER GROUP MECHANICS:
  - Multiple consumer instances in a group share the topic partitions
  - Each partition is consumed by exactly one consumer at a time
  - Offset tracks position: which events have been processed
  - Auto-commit offset vs manual commit after successful processing

# Manual offset commit (recommended for correctness):
consumer.poll(timeout_ms=1000)
for message in records:
    process(message.value)
    write_to_sink(message.value)   # write BEFORE committing offset
consumer.commit()                  # commit AFTER successful write
# If write fails: do not commit → message reprocessed on next poll → at-least-once

# Auto-commit (default, simpler, less safe):
# Offset committed on a timer regardless of whether processing succeeded.
# Risk: offset committed before write completes → message lost on crash.
consumer.enable_auto_commit = True  # do NOT use for financial data

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, sum as spark_sum
from pyspark.sql.types import StructType, StringType, DecimalType, TimestampType

spark = SparkSession.builder.appName('payment_stream').getOrCreate()

payment_schema = StructType() \
    .add('payment_id', StringType()) \
    .add('amount', DecimalType(10, 2)) \
    .add('store_id', StringType()) \
    .add('event_time', TimestampType())

raw_stream = spark.readStream \
    .format('kafka') \
    .option('kafka.bootstrap.servers', 'kafka:9092') \
    .option('subscribe', 'payments.v1') \
    .option('startingOffsets', 'latest') \
    .load()

payments = raw_stream \
    .select(from_json(col('value').cast('string'), payment_schema).alias('data')) \
    .select('data.*') \
    .withWatermark('event_time', '5 minutes')

# Aggregate: revenue per store per 1-hour tumbling window:
hourly_revenue = payments \
    .groupBy(window('event_time', '1 hour'), 'store_id') \
    .agg(spark_sum('amount').alias('hourly_revenue'))

query = hourly_revenue.writeStream \
    .outputMode('update') \
    .format('delta') \
    .option('checkpointLocation', 's3://freshcart-lake/checkpoints/hourly_revenue') \
    .trigger(processingTime='30 seconds') \
    .start('s3://freshcart-lake/silver/hourly_revenue')

Micro-Batch — Small Batches on a Short Interval

MICRO-BATCH (Spark Structured Streaming, default):
  t=0s:   Collect all Kafka messages arrived in last 30 seconds
  t=0.5s: Process as one Spark batch job (bounded)
  t=2.3s: Write results to sink, commit Kafka offsets
  t=30s:  Collect next batch... repeat

  Latency: ~30 seconds. Throughput: very high. State: per-batch checkpoint.
  Weakness: minimum latency = trigger interval (can't go below ~1s practically)

TRUE STREAMING (Apache Flink):
  Event arrives → Immediately processed → Output emitted. No batch boundary.

  Latency: milliseconds (10–200ms typical end-to-end).
  State: maintained continuously in a distributed state store (RocksDB).
  Weakness: more complex to operate, more expensive, harder to debug.

CHOOSING:
  Need < 1 second latency?       → True streaming (Flink)
  1 second – 15 minutes latency? → Micro-batch (Spark Structured Streaming)
  > 15 minutes latency?          → Batch (standard Spark or dbt)
  Most "real-time" dashboards?   → Micro-batch with 5-minute trigger

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
stream = spark.readStream.format('kafka') \
    .option('kafka.bootstrap.servers', 'kafka:9092') \
    .option('subscribe', 'orders.v1').load()

# Fixed interval micro-batch (most common):
query = stream.writeStream \
    .trigger(processingTime='5 minutes') \
    .format('delta') \
    .option('checkpointLocation', 's3://freshcart-lake/checkpoints/orders_stream') \
    .start('s3://freshcart-lake/bronze/orders_stream')

# Once trigger — process all available data right now, then stop
# (useful for backfill or scheduled runs that want streaming semantics):
query = stream.writeStream.trigger(once=True).format('delta') \
    .start('s3://freshcart-lake/bronze/orders_stream')

# Available-now trigger (Spark 3.3+) — process all available data in batches:
query = stream.writeStream.trigger(availableNow=True).format('delta') \
    .start('s3://freshcart-lake/bronze/orders_stream')

# Continuous processing (experimental — approaches true streaming):
query = stream.writeStream.trigger(continuous='1 second') \
    .format('console').start()

import time
from datetime import datetime, timedelta, timezone

def run_micro_batch(interval_seconds: int = 300) -> None:
    """Run a micro-batch loop that processes N minutes of data repeatedly."""
    while True:
        batch_end   = datetime.now(timezone.utc)
        batch_start = batch_end - timedelta(seconds=interval_seconds)

        try:
            rows = extract_from_source(batch_start, batch_end)
            transformed = transform(rows)
            upsert_to_destination(transformed)   # upsert is critical here
            print(f'Batch [{batch_start.isoformat()} - {batch_end.isoformat()}]: {len(rows)} rows')
        except Exception as e:
            print(f'Batch failed: {e}')
            # Do NOT advance the time window — retry same window next loop

        time.sleep(interval_seconds)   # wait before next batch

# Run with a delay to let late data arrive.
# The March 17 batch runs at 6 AM March 18 — 6+ hours after midnight.
# Any order timestamped March 17 but arriving late has 6 hours to arrive.
# Simple. Correct. No special logic needed.

# For very late data (hours or days late), run a correction batch:
# 0 10 * * * python3 pipeline.py --date yesterday --mode correction
# This reprocesses yesterday with all data that has arrived since the
# daily run. Upsert semantics make this safe.

# If micro-batch interval is 5 minutes, query with 10-minute lookback:
def extract_with_overlap(batch_end, overlap_minutes=10):
    batch_start = batch_end - timedelta(minutes=5 + overlap_minutes)
    return db.query(
        "SELECT * FROM events WHERE event_time BETWEEN %s AND %s",
        (batch_start, batch_end),
    )
# This reads events from 15 minutes ago to now. Events that arrived late
# (up to 10 minutes late) are included. Upsert at destination handles the
# re-processing of already-seen events.

from pyspark.sql.functions import window, sum as spark_sum

# Define allowed lateness of 10 minutes:
payments_with_watermark = payments.withWatermark('event_time', '10 minutes')
#   ↑ Any event more than 10 minutes behind the watermark is considered late

hourly_revenue = payments_with_watermark \
    .groupBy(window('event_time', '1 hour'), 'store_id') \
    .agg(spark_sum('amount').alias('revenue'))

# Output mode matters for late data:
# 'append':   Output only when window is finalised (after watermark passes)
#              → Lowest memory, highest latency, correct
# 'update':   Output whenever window data changes (including late updates)
#              → Lower latency, more updates to downstream
# 'complete': Output all windows on every trigger — only for small datasets

query = hourly_revenue.writeStream \
    .outputMode('update') \
    .format('delta') \
    .option('checkpointLocation', 's3://checkpoints/hourly_revenue') \
    .trigger(processingTime='1 minute') \
    .start('s3://freshcart-lake/silver/hourly_revenue_stream')

With update mode and upsert at destination (Delta MERGE): window results
are updated as late events arrive, and the final result is correct after
the watermark passes. Downstream consumers see intermediate updates and
must handle them.

THE LATE DATA DECISION TREE:
  99th percentile latency < 5 min:  allowed_lateness = 10 min
  99th percentile latency < 30 min: allowed_lateness = 60 min
  Data can be hours late (mobile app offline): use batch with correction run

Rule: if you cannot bound your data lateness to < 1 hour, streaming with
watermarks becomes very expensive — a 1-hour watermark on 1-minute windows
means holding 60 windows open in memory simultaneously. Use batch or
micro-batch + correction instead.

QUESTION 1: What is the maximum acceptable data latency for this use case?
  < 1 second:  True streaming only (Flink / Kafka Streams)
  1s – 5 min:  Micro-batch (Spark Structured Streaming, 30s trigger)
  5 – 60 min:  Micro-batch (5–15 minute trigger) or fast batch
  > 1 hour:    Batch (daily, hourly, or whatever interval fits)

  COMMON MISTAKE: picking streaming because "real-time" sounds better.
  Ask: what decision or action requires this latency? If the answer is "a
  dashboard refresh" — does the user genuinely need sub-second updates,
  or would 5-minute updates serve equally well?

QUESTION 2: How complex is the transformation logic?
  Simple filtering/typing, no joins:       All three work fine
  Joins to slowly-changing dimensions:     Batch or micro-batch (easier state)
  Aggregations over large time windows:    Batch (all data available at once)
  Pattern detection across event sequence: Streaming (Flink CEP)
  ML model inference per event:            Streaming (low-latency requirement)

  RULE: if the transformation requires data from multiple time periods or
  large lookups, streaming state management becomes complex and expensive.
  Batch makes this trivial.

QUESTION 3: How late can source events arrive?
  < 5 minutes late:   Micro-batch with 10-min lookback overlap
  5–60 minutes late:  Streaming with 60-min watermark OR micro-batch + correction
  Hours late (mobile offline data, delayed batch feeds): Batch only

QUESTION 4: How complex can the operations model be?
  Small team, no streaming expertise: Batch (always)
  Team familiar with Spark Streaming: Micro-batch
  Dedicated streaming engineers:       True streaming if latency requires it

QUESTION 5: What is the data volume?
  < 1 GB/day:       Batch on a single machine (Pandas, not even Spark needed)
  1 GB – 1 TB/day:  Batch Spark or micro-batch Spark
  > 1 TB/day:       Micro-batch or streaming depending on latency needs
  > 10 TB/day:      Almost certainly micro-batch or streaming

PRACTICAL ROUTING TABLE:
  Finance report (daily revenue, costs):    BATCH
  Operations dashboard (last 15 min):       MICRO-BATCH (5 min)
  Real-time fraud detection:                 STREAMING (Flink)
  Customer segmentation (weekly):            BATCH
  Live delivery tracking:                    STREAMING
  Hourly data quality check:                 MICRO-BATCH (10 min)
  Payment gateway health monitoring:         STREAMING

LAMBDA ARCHITECTURE (2012–2020, now declining):
  Source → Batch Layer (Spark, nightly) → Batch views (accurate, slow)
         → Speed Layer (Storm/Flink)    → Real-time views (fast, approximate)
                                         ↓
                               Serving Layer (merge both)
  Problems: two codebases doing the same logic, subtle batch/stream result
  differences, operational complexity of maintaining two stacks.

KAPPA ARCHITECTURE (2014–present, now dominant):
  Source → Kafka (persistent event log, replayed for reprocessing)
         → Single streaming pipeline (Flink or Spark Streaming)
         → Serving layer (no separate batch layer)
  Reprocessing: replay Kafka from the beginning with a new consumer group.
  Benefits: single codebase, one source of truth, natural reprocessing.
  Limitation: requires Kafka to retain data long enough for full
  reprocessing (expensive at high data volumes).

MODERN HYBRID (2022–present, most practical):
  Source → Kafka (days of retention for streaming)
         → S3/ADLS (years of retention as Parquet, Delta Lake)
  Streaming path: Kafka → Flink/Spark Streaming → real-time sink (low latency)
  Batch path:     S3 → Spark batch → warehouse (historical accuracy, bulk)
  Both write to the same Delta Lake tables (upserts reconcile any differences)

FreshCart's CTO asks you to design the data processing architecture for three specific business requirements. Here is how a senior data engineer applies the decision framework to each:

Requirement 1 — Finance department needs a daily P&L report by 7 AM. The report covers all transactions from the previous day, reconciled against bank settlement data. It needs to be exact — finance cannot act on approximate data.

Decision: Batch, daily at 4 AM. Exact financial data requires all transactions to have settled before processing. A 4 AM run gives 4 hours for any late-arriving transactions to appear in the source before the report is needed at 7 AM. Streaming would add complexity without helping — finance does not act on real-time P&L, only on the final daily figure.

Requirement 2 — Operations team needs a dashboard showing store performance for the last 30 minutes, refreshed every 5 minutes.Store managers use this to respond to issues mid-day (slow service, high order cancellation rate).

Decision: Micro-batch, 5-minute trigger. 5-minute latency is perfectly adequate for a manager responding to operational issues. A Spark Structured Streaming job reads from Kafka (where order events are published as they occur), aggregates the last 30 minutes of data every 5 minutes, and writes to a Delta table that the dashboard queries. This is far simpler than true streaming and adequate for the use case.

Requirement 3 — Risk team needs real-time payment fraud detection — block suspicious transactions before they complete.A transaction that requires investigation must be flagged within 2 seconds of the payment event.

Decision: True streaming, Apache Flink. The 2-second latency requirement eliminates both batch and micro-batch. Flink is chosen over Spark Streaming because Flink's native event time processing and sub-second latency per record meet the requirement. The Flink job consumes from Kafka, computes a velocity score (transactions per card in last 60 seconds), and publishes a fraud signal back to Kafka where the payment service reads it before completing the transaction.

Three requirements, three different processing models. All three run in the same data platform, sharing the same Kafka cluster and the same Delta Lake. This is the architecture of a mature, fit-for-purpose data platform — not one model for everything, but the right model for each need.