Spark Streaming

Overview

Spark Streaming is the part of Apache Spark that processes live data as it arrives. In its modern form, called Structured Streaming, a stream is treated as a table that keeps growing. You write the same DataFrame or SQL query you would write for a batch job, and Spark runs it again and again on new data, usually in small micro-batches. This gives near-real-time results with a familiar programming model.

The main promise is simplicity and correctness. You describe the query, and Spark handles incremental execution, state, fault recovery and exactly-once style results when sources and sinks support it. Spark Streaming fits event aggregation, monitoring, near-real-time feature updates and continuous loading of lake tables such as Delta Lake. For tabular AI, it keeps feature tables fresh, scores new rows as they arrive, and feeds anomaly detection and continual learning loops.

What it is

There are two streaming APIs in the Spark family:

• DStreams (legacy). The original Spark Streaming API, built on RDDs, which splits a stream into small batches. It is older and mostly replaced.
• Structured Streaming (current). A streaming engine built on the DataFrame and SQL engine. It is the recommended choice for new work.

When people say Spark Streaming today, they usually mean Structured Streaming. This article focuses on it and mentions DStreams only for context.

A helpful analogy is a conveyor belt at a sorting centre. Parcels arrive continuously. Instead of handling each parcel alone, workers take a small pile every few seconds, sort it, update the running totals and put the results on the shelf. Customers see totals that are only a few seconds old. The rules of sorting are written once and applied to every pile.

What it can do

• Read from sources such as message queues, files arriving in a folder, and table change feeds, including Apache Kafka topics.
• Apply filters, joins, aggregations, windows and custom logic.
• Write to sinks such as tables, files, message queues and custom writers.
• Keep state across batches, for example running counts per key.
• Handle late data using event-time windows and watermarks.
• Recover from failures using checkpoints.
• Use the same code for batch and streaming, with small changes.

Processing modes

• Micro-batch. The default. Spark triggers small batch jobs at an interval or as fast as it can. Latency is usually from hundreds of milliseconds to seconds.
• Available-now. Processes everything currently available, then stops. Useful for incremental batch jobs that run on a schedule.
• Continuous (experimental or limited). Processes records with lower latency for simple operations, with fewer supported features.

Where it came from

Spark Streaming appeared early in Spark's history as a way to apply the batch engine to live data. The key idea was discretised streams: cut the stream into small time slices and run a normal Spark job on each slice. This reused Spark's fault tolerance and gave a single engine for batch and streaming. It was a different idea from event-at-a-time engines such as Apache Flink.

As users built real pipelines, they found the RDD-based API awkward for event-time logic and end-to-end guarantees. Structured Streaming was introduced later with a cleaner model: the stream is an unbounded table, the query is incremental, and the engine takes care of state, ordering and recovery. It joined the DataFrame and SQL optimiser, and added features such as watermarks, stream-stream joins, arbitrary stateful processing and the available-now trigger.

Streaming also became tightly linked with table formats. When a table format records every commit, a table can act as a streaming source or sink, which unified streaming and batch storage. Exact feature timing varies by version, so check the documentation of your release.

How it works

Step 1: Define the input as an unbounded table

You read a stream with readStream. The result is a DataFrame, but it represents data that keeps arriving. Conceptually, each new record is a new row appended to an input table.

Step 2: Write the query

You apply ordinary transformations: select, filter, join, groupBy, window. Nothing special is needed for the query logic. The same expression could run on a static DataFrame.

Step 3: Choose an output mode

The output mode says what is written each time the result changes:

• Append. Only new final rows are written. Good for pass-through and for windowed results once the window closes.
• Update. Only rows that changed since the last trigger are written. Good for running aggregates.
• Complete. The full result table is written every time. Works only for small aggregated results.

Step 4: Choose a trigger

The trigger decides when to process new data: a fixed interval such as every ten seconds, as soon as possible, or once with available-now. Shorter intervals cut latency but raise overhead and can create many small output files.

Step 5: Start the query with a checkpoint

You call writeStream with a checkpoint location, then start. The checkpoint stores progress, such as which source offsets were processed, and state data. It is the memory of the stream.

An everyday comparison is a bookmark. If you close the book and reopen it later, the bookmark tells you where you were. Without it, you either start over or lose your place.

Step 6: Each micro-batch runs

On each trigger, Spark asks the source for new data since the last offset, plans an incremental batch, runs it on the cluster and writes results to the sink. Then it records progress in the checkpoint.

Step 7: State is kept for stateful operations

Aggregations over time, deduplication and stream-stream joins need memory of past data. Spark stores this state in a state store, backed by memory with changes logged to the checkpoint. State can grow without bound unless you limit it with windows and watermarks.

Step 8: Event time and watermarks handle late data

Events have an event time, when they happened, which differs from the processing time, when Spark sees them. Windows are defined on event time. A watermark is a rule such as "I will wait up to ten minutes for late events". Spark tracks the maximum event time seen and drops state for windows older than the watermark minus the delay. This bounds state and defines when a window result is final.

Think of a school bus that waits ten minutes for latecomers before leaving. After that, it goes, and later arrivals must take another ride, which here means their data may be dropped from that window.

Step 9: Recovery after failure

If a job crashes, you restart it with the same checkpoint. Spark reads the last committed offsets and state, and replays any incomplete batch. With replayable sources and idempotent or transactional sinks, results are not duplicated. This is the basis of end-to-end exactly-once behaviour.

Step 10: Monitoring

Each batch reports metrics such as input rows per second, processing rate, batch duration and state size. Watching them tells you whether the stream keeps up. If processing time per batch is longer than the trigger interval, lag grows.

Key concepts and terms

A worked example

This example reads JSON events from a queue, counts events per five-minute window, and writes to a Delta table. Hosts and paths are fake.

Read the stream

from pyspark.sql import SparkSession, functions as F, types as T

spark = SparkSession.builder.appName("events_stream").getOrCreate()

schema = T.StructType([
    T.StructField("event_id", T.StringType()),
    T.StructField("user_id", T.StringType()),
    T.StructField("event_type", T.StringType()),
    T.StructField("amount", T.DoubleType()),
    T.StructField("event_ts", T.TimestampType()),
])

raw = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker.example.internal:9092")
    .option("subscribe", "events")
    .option("startingOffsets", "latest")
    .load()
)

events = (
    raw.select(F.from_json(F.col("value").cast("string"), schema).alias("e"))
       .select("e.*")
)

The queue source gives binary key and value columns. We cast the value to a string and parse it into columns.

Deduplicate and aggregate with a watermark

agg = (
    events
    .withWatermark("event_ts", "10 minutes")
    .dropDuplicates(["event_id", "event_ts"])
    .groupBy(
        F.window("event_ts", "5 minutes"),
        "event_type",
    )
    .agg(
        F.count("*").alias("events"),
        F.sum("amount").alias("amount_sum"),
    )
    .select(
        F.col("window.start").alias("window_start"),
        F.col("window.end").alias("window_end"),
        "event_type", "events", "amount_sum",
    )
)

The watermark tells Spark to keep state for events up to ten minutes late. Including the event time in the deduplication keys lets Spark drop old keys as the watermark moves.

Write with a checkpoint

query = (
    agg.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "s3://example.internal/chk/events_5m")
    .trigger(processingTime="30 seconds")
    .start("s3://example.internal/lake/gold/events_5m")
)

query.awaitTermination()

In append mode with a watermark, a window is written once, after the watermark passes the end of the window. Results are final, and downstream readers never see a half-counted window.

Use foreachBatch for custom logic

def upsert(batch_df, batch_id):
    batch_df.createOrReplaceTempView("updates")
    batch_df.sparkSession.sql("""
        MERGE INTO gold.user_totals t
        USING (SELECT user_id, SUM(amount) AS amt FROM updates GROUP BY user_id) s
        ON t.user_id = s.user_id
        WHEN MATCHED THEN UPDATE SET t.total = t.total + s.amt
        WHEN NOT MATCHED THEN INSERT (user_id, total) VALUES (s.user_id, s.amt)
    """)

(events.writeStream
    .foreachBatch(upsert)
    .option("checkpointLocation", "s3://example.internal/chk/user_totals")
    .start())

foreachBatch gives you each micro-batch as a normal DataFrame. This is how you merge into tables or write to several sinks. A retry can run the same batch twice, so the merge logic should be idempotent or use the batch id to skip duplicates. The simple add shown here would double count on a replay, so in production keep a record of processed batch ids, or recompute totals from the source rows instead.

Incremental batch with available-now

(spark.readStream.format("delta").load("s3://example.internal/lake/bronze/events")
    .writeStream
    .format("delta")
    .option("checkpointLocation", "s3://example.internal/chk/silver_events")
    .trigger(availableNow=True)
    .start("s3://example.internal/lake/silver/events")
    .awaitTermination())

This processes all new data since the last run and then stops, so a scheduler can run it every hour. You get streaming's bookkeeping with batch's cost profile.

A numeric picture of a watermark

Say the watermark delay is 10 minutes and the largest event time seen is 12:20. The watermark is 12:10. A window from 12:00 to 12:05 ended before the watermark, so it is final and its state can be dropped. An event with time 12:07 that arrives now is still within the allowed lateness for the window 12:05 to 12:10 and is counted. An event with time 12:03 arriving now belongs to a closed window and is ignored. With a delay of 10 minutes, state for at most about 10 minutes of windows is kept, instead of growing forever.

Strengths and limitations

Strengths

• One API for batch and streaming. Less code and fewer bugs from duplicate logic.
• Strong guarantees. Checkpoints and transactional or idempotent sinks give exactly-once style results.
• Rich operators. Windows, joins, deduplication and arbitrary state.
• Scale. Runs on the same cluster engine as batch jobs.
• Lake integration. Works naturally with table formats that support streaming reads and writes.
• Incremental batch. Available-now trigger reduces cost for non-urgent pipelines.
• Ecosystem. Many connectors, plus access from PySpark, Scala and SQL.

Limitations

• Latency floor. Micro-batches have overhead. Single-digit millisecond latency is not realistic.
• State management. Large state needs tuning, and bad watermarks cause unbounded growth.
• Small files. Frequent triggers create many small files unless you compact.
• Operational care. Checkpoints must be preserved, and some query changes break compatibility with old checkpoints.
• Late data trade-offs. A watermark drops events later than the delay.
• Limited continuous mode. Fewer operators are supported.
• Resource use. Long-running clusters cost money even when traffic is low unless you scale them.
• Debugging. Time-dependent behaviour is harder to test than batch.

How it compares with alternatives

Many teams use a message log for transport, Spark for aggregation and loading, and a table format for storage. If your need is sub-second reaction to single events, a true event-at-a-time engine is a better fit. If your need is fresh tables every few seconds or minutes, Structured Streaming is often simpler.

Role in tabular foundation models

Tabular foundation models learn from rows and columns, cope with mixed types and missing values, and can predict from a few example rows in context. Spark Streaming matters because real tables are not static. New rows arrive all the time, and models need to see them, score them and sometimes adapt to them.

Fresh feature tables

Streaming jobs turn raw events into feature rows. For example, a job can keep per-customer counts and sums in five-minute windows and write them to a table in the gold layer. A tabular model such as TabPFN, TabICL or TabSTAR then reads recent rows as context or as query rows. Because the streaming job uses the same logic as the batch backfill, training and serving features match.

Row and column structure

Streaming data usually starts as nested events. A streaming job parses them into a flat table with typed columns, which is the shape tabular models need. A schema check at the edge, with bad records sent to a quarantine table, keeps malformed rows out of model inputs.

Mixed types and missing values

Events often arrive with missing fields. A streaming job can count nulls per column per window, add indicator columns and keep nulls intact. This lets a model, or an imputation step, handle them. Imputed values should be written to new columns or tables, so the original event remains visible.

No column names

Event sources sometimes send positional fields or short codes. A streaming job can map them to columns using a lookup. For testing a model's reliance on names, a job can emit a second table with generic column names. Keep the mapping table versioned.

Pre-training and evaluation data

Pre-training data is usually built in batch, but streaming can collect fresh real tables and examples continuously into a bronze table for later curation. For evaluation, a streaming job can store the model's input, prediction and later outcome side by side, producing a labelled set for monitoring. Read about how these models are described on the model research and model architecture pages.

In-context learning on live data

With in-context learning, a model predicts new rows from labelled example rows. A streaming job can keep a rolling sample of recent labelled rows, for example with deduplication and a bounded buffer, and attach it as context when scoring new rows in foreachBatch. This allows the model to adapt to recent patterns without retraining. Keep the context bounded and stable, and log which context rows were used.

Continual learning

For continual learning, the stream decides what the model sees next. A job can read only new rows, update statistics, trigger a small update or refresh the context set, and record the table version processed. Because checkpoints record progress, the loop can restart after a crash without skipping or repeating data. Concept drift can be tracked by comparing feature distributions per window.

Anomaly detection

Streaming anomaly detection is a natural fit. A job can score each new row with a model, compare window aggregates with history, and write alerts to a table or queue. Typical checks are volume drops, unusual values and new category levels. See anomaly detection. Because Spark can mix batch and stream, the same detector can be tested on history and then run live.

Sector and semantic classification

When new records arrive, such as new companies or new columns, a streaming job can call a classifier to assign a sector or semantic type, store the label with a confidence score and route low-confidence rows to review. Labels are versioned along with the model version. For ideas on measuring such systems, see the model benchmarks page.

Practical limits

Foundation models can be heavy. Scoring each micro-batch needs a loaded model and enough memory, so teams often use a model service or load the model once per task. If latency needs are below what micro-batches deliver, scoring may need to move into a separate low-latency service, with Spark used for the aggregated features.

Practical guidance and best practices

Design

• Decide the latency you really need. Seconds or minutes is often enough.
• Use event time, not processing time, for windows.
• Set a watermark based on measured lateness, not guesses.
• Keep state small: use windows, TTLs and deduplication keys with event time.
• Prefer append mode for final results and update mode for dashboards.
• Separate ingestion (bronze) from cleaning (silver) from serving (gold).

Reliability

• Always set a stable checkpoint location, one per query.
• Never share a checkpoint between queries.
• Use replayable sources and idempotent or transactional sinks.
• Make foreachBatch logic idempotent, for example by recording batch ids.
• Test restart behaviour: stop the job, start it and verify no gaps or duplicates.
• Plan for schema changes at the source with a tolerant parser and a quarantine path.

Performance

• Choose a trigger interval that matches the processing time per batch.
• Limit input per batch with rate options so a backlog does not crash the job.
• Keep partitions balanced, and watch for skewed keys.
• Compact small output files on a schedule.
• Avoid heavy Python per-row work, and prefer built-in functions or batched pandas functions.
• Tune the number of shuffle partitions for streaming, where batches are small.

Operations

• Monitor input rate, processing rate, batch duration and state size.
• Alert when lag grows or when a query stops.
• Use automatic restarts with backoff.
• Keep a runbook for resetting a checkpoint safely.
• Record the code version with each checkpoint so changes are traceable.

Testing

• Use a memory sink or test sources to feed known events.
• Test late, duplicate and out-of-order events on purpose.
• Compare streaming output against a batch computation over the same data.
• Test with an empty stream and with a burst.

Security and governance

• Keep credentials in secret stores, never in code.
• Encrypt data in transit and at rest.
• Drop personal data as early as possible.
• Register output tables in a data catalog with owner and freshness.

Window types

A tumbling window is like a calendar day: each event falls in exactly one. A sliding window is like a rolling seven-day average: each event appears in several windows. A session window is like a conversation: it lasts until there is a long silence.

sliding = events.groupBy(
    F.window("event_ts", "10 minutes", "1 minute"), "event_type"
).count()

Overlapping windows multiply state and output rows, so use the smallest overlap that serves the need.

Stream-stream and stream-static joins

A stream-static join enriches each event with data from a table that changes slowly, such as a customer dimension. It is cheap because the static side needs no state. A stream-stream join matches events from two live streams, for example clicks and purchases. It needs state for both sides, so you should add a time condition and watermarks, otherwise Spark must keep every row forever.

joined = clicks.alias("c").join(
    purchases.alias("p"),
    F.expr("""
        c.user_id = p.user_id AND
        p.event_ts BETWEEN c.event_ts AND c.event_ts + INTERVAL 30 MINUTES
    """),
    "inner",
)

The time range tells Spark how long it must remember a click while waiting for a matching purchase.

Sizing and capacity thinking

A simple model helps with planning. If events arrive at 5,000 per second and each micro-batch is 10 seconds, each batch holds about 50,000 events. If a batch takes 6 seconds to process, the job has headroom of 4 seconds. If a batch takes 12 seconds, the job falls behind by 2 seconds every batch, and lag grows without limit. The goal is a processing time well below the trigger interval, with room for bursts. Measure on real traffic, not just on averages, because peaks decide whether you fall behind.

State size is the second planning number. Estimate the number of keys, the size of each state record and the window length. A million keys with 200 bytes each is a manageable 200 MB. A hundred million keys with long windows can become hundreds of gigabytes and needs a larger cluster or a different design, such as pre-aggregating upstream.

Schema evolution in streams

Producers change events over time: new fields, renamed fields, changed types. A robust stream expects this.

• Parse with a permissive schema and keep unknown fields in a raw column.
• Route records that fail parsing to a quarantine table with the error.
• Add new columns to the target table on purpose, not automatically.
• Version event schemas, and agree changes with producers.
• Monitor the share of rejected records. A jump signals a producer change.

Backfills and reprocessing

Sooner or later you must recompute history after a bug or a logic change. A safe approach is to keep raw events in a bronze table, run a batch job over the affected time range into a new table, compare it with the streaming output, then switch readers. Avoid replaying history through the live stream with the same checkpoint, because it mixes old and new state. If you must reset a stream, start a new checkpoint and define the starting point explicitly.

A pre-launch checklist

1. The latency requirement is written down and realistic for micro-batches.
2. Event time and watermark are defined and justified by measured lateness.
3. Checkpoint location is unique, durable and documented.
4. Sink writes are idempotent or transactional.
5. Bad records go to a quarantine path.
6. Metrics and alerts cover lag, batch duration and state size.
7. Restart and failure tests passed.
8. A compaction job exists for the output table.
9. A backfill plan exists.
10. Owners and on-call contacts are listed in the catalog.

Common mistakes

1. Using processing time for windows. Results change when the system is slow or restarts.
2. No watermark on stateful queries. State grows until the job runs out of memory.
3. A watermark that is too short. Valid late data is dropped silently.
4. Deleting or sharing checkpoints. The job loses its place or corrupts another query.
5. Non-idempotent foreachBatch. A retried batch double counts.
6. Tiny triggers with big overhead. Many small files and unstable jobs result.
7. Complete output mode on large results. Each batch rewrites everything.
8. Changing the query shape and reusing an old checkpoint. Some changes make old state unreadable.
9. Ignoring backlog. A source that produces faster than the job can process leads to ever-growing lag.
10. Skipping schema validation. One bad record can fail the whole batch.
11. Treating streaming as magic real time. Micro-batches add latency, so confirm the requirement.
12. Forgetting compaction. Read performance falls as small files accumulate.

Frequently asked questions

What is Spark Streaming?

Spark Streaming is Spark's capability to process live data. Today it mainly means Structured Streaming, where a stream is treated as a table that keeps growing and your DataFrame or SQL query runs incrementally on new data. It supports windows, joins, state and fault recovery.

What is the difference between Spark Streaming and Structured Streaming?

Spark Streaming, also called DStreams, is the older API built on RDDs that cuts a stream into small batches. Structured Streaming is the newer engine built on DataFrames and SQL, with event-time support, watermarks, and stronger guarantees. New projects should use Structured Streaming.

Is Spark Streaming real time?

It is near real time. Micro-batch processing typically gives latencies from hundreds of milliseconds to a few seconds, depending on the job. That is enough for dashboards, monitoring and many feature pipelines. If you need millisecond reactions to single events, an event-at-a-time engine such as Apache Flink is usually better.

What is a watermark?

A watermark is a threshold that says how late data is allowed to be. Spark tracks the largest event time seen and subtracts the delay to get the watermark. Windows and state older than that are finalised and removed. It balances completeness against memory use and result delay.

What is a checkpoint and why do I need it?

A checkpoint stores a stream's progress and state, such as processed offsets and aggregation state. If the job stops or fails, it resumes from the checkpoint without losing or repeating data. Each query needs its own stable checkpoint location, and deleting it resets the stream.

Does Spark Streaming guarantee exactly-once processing?

It can provide end-to-end exactly-once results when the source is replayable, the engine tracks offsets in checkpoints, and the sink is idempotent or transactional. Table formats with atomic commits make this practical. With sinks that cannot meet these conditions, you may get at-least-once behaviour and need deduplication.

How do I handle late or out-of-order events?

Define windows on event time and set a watermark that matches how late your data typically is. Events within the allowed lateness update their windows, and later events are dropped from stateful results. For stricter needs, store raw events in a bronze table and recompute corrections in a batch job.

Can I run batch and streaming code together?

Yes. Because the API is shared, much logic can be reused. You can backfill history with a batch job and then start the stream from a known point, or use the available-now trigger to run incremental batches on a schedule. Make sure the two paths produce the same results by testing them on the same data.

What sources and sinks does it support?

Common sources include message queues such as Apache Kafka, files landing in a folder and table change feeds. Common sinks include table formats like Delta Lake, files, message queues and custom writers through foreachBatch. Exact connectors depend on your platform and version.

Why is my streaming job falling behind?

Usually the processing time per batch exceeds the trigger interval, or the input rate rose. Causes include skewed keys, heavy state, slow sinks, small files and slow Python functions. Check the batch duration, input and processing rates, and state size, then fix the biggest bottleneck or add resources.

Glossary

• Append mode: Output mode that writes only new final rows.
• Available-now trigger: A trigger that processes all available data and then stops.
• Backpressure: Limiting the input rate so the job can keep up.
• Checkpoint: Stored progress and state used to recover a stream.
• Complete mode: Output mode that writes the whole result each time.
• DStream: The legacy discretised stream API.
• Event time: The time an event actually occurred.
• Exactly-once: Each input record affects the output a single time.
• foreachBatch: A hook to run custom batch code for each micro-batch.
• Micro-batch: A small batch job run on each trigger.
• Offset: A position within a source, such as a place in a log.
• Processing time: The time Spark handles an event.
• Sink: The destination of a stream's output.
• Source: The origin of a stream's input.
• State store: Storage for state kept between batches.
• Trigger: The rule that decides when to run the next batch.
• Update mode: Output mode that writes only changed rows.
• Watermark: A bound on lateness that lets Spark finalise windows and drop old state.
• Window: A time range used to group events for aggregation.

Summary

Spark Streaming, in its Structured Streaming form, lets you process live data with the same DataFrame and SQL code you use for batch. Spark treats the stream as a growing table, runs the query incrementally in micro-batches, keeps state, uses event-time windows and watermarks for late data, and recovers from failure with checkpoints.

It is a strong choice for near-real-time aggregation, continuous loading of lake tables and fresh feature pipelines, and a weaker choice for millisecond latency. For tabular foundation models, it keeps feature tables current, supports rolling context for in-context learning, feeds continual learning loops, and scores rows for imputation, anomaly detection and classification as they arrive. Use PySpark for the code, Delta Lake for reliable storage, and design watermarks, checkpoints and idempotent writes with care.

Related

It belongs to Batch and streaming processing in my stack, under Data engineering and lakehouse. Others in the same group: