Spark Structured Streaming is Apache Spark’s declarative stream-processing engine. Instead of asking developers to write record-at-a-time callbacks, it treats a stream as an unbounded table and lets developers express logic with the same DataFrame, Dataset, and SQL APIs used for batch jobs. Spark then runs the query incrementally as new data arrives.
The default execution mode is micro-batch. Spark collects newly available input for a trigger interval, plans a small batch job through Catalyst and Tungsten, executes it across the cluster, updates state, writes output, and records progress in checkpoints. This design gives Spark a strong bridge from batch to streaming: mature SQL optimization, familiar APIs, and exactly-once processing when sources are replayable and sinks are idempotent or transactional.
Spark also introduced continuous processing mode in Spark 2.3. Continuous mode aims for much lower latency by running long-lived tasks that process records continuously instead of launching a Spark job for every trigger. The trade-off is narrower query support and weaker fault-tolerance guarantees: Spark’s official guide describes roughly 1 ms end-to-end latency with at-least-once guarantees, compared with roughly 100 ms best-case latency and exactly-once guarantees for micro-batch.
The architecture is a useful contrast with Flink and Beam. Flink is a native event-at-a-time streaming engine with batch as a special case. Beam is a portable programming model with runner-specific execution. Structured Streaming starts from Spark SQL and makes streaming look like incremental table maintenance. That makes it especially attractive for teams already using Spark for ETL, data lakes, ML feature generation, and SQL analytics.
Structured Streaming models incoming records as rows appended to an input table. A query over that table produces a result table. Every trigger, Spark reads a range of new offsets, incrementally updates the result, and writes changes to a sink. The programmer writes a DataFrame query; the engine decides how to turn that query into repeated incremental execution.
A trigger defines how often Spark checks for new data and starts another micro-batch. A processing-time trigger such as 10 seconds gives the system time to batch records and amortize scheduling overhead. A once or available-now trigger is useful for bounded backfills through the same streaming code path.
Structured Streaming records source offsets, batch IDs, query metadata, and state-store data in a checkpoint location. On restart, Spark uses that checkpoint to know which input offsets were already planned and which output batches may need replay. The exactly-once story depends on replayable sources and sinks that can make retries idempotent.
Continuous mode keeps tasks running and checkpoints progress at a configured interval. It removes much of the micro-batch scheduling cost, but it supports only a limited subset of operations compared with micro-batch. It is best understood as a specialized low-latency mode, not the default way to run arbitrary Structured Streaming queries.
Structured Streaming’s main abstraction is not a callback loop. It is an incrementally maintained query plan. This lets Spark reuse batch query optimization, but it also means unsupported operations are usually operations Spark cannot safely incrementalize for an unbounded input.
In micro-batch mode, Spark repeatedly chooses an offset range from each source, writes that range to checkpoint metadata, executes a Spark job over the range, updates state stores, commits sink output, and records the batch as complete. If a worker crashes, Spark can retry tasks. If the driver restarts, the checkpoint tells the query where to resume.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, sum as sum_
from pyspark.sql.types import StructType, StringType, TimestampType, LongType
spark = SparkSession.builder.appName("orders-stream").getOrCreate()
schema = (
StructType()
.add("order_id", StringType())
.add("customer_id", StringType())
.add("event_time", TimestampType())
.add("amount_cents", LongType())
)
events = (
spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "broker:9092")
.option("subscribe", "orders")
.load()
.select(from_json(col("value").cast("string"), schema).alias("e"))
.select("e.*")
)
hourly = (
events
.withWatermark("event_time", "15 minutes")
.groupBy(window(col("event_time"), "1 hour"), col("customer_id"))
.agg(sum_("amount_cents").alias("amount_cents"))
)
query = (
hourly.writeStream
.format("delta")
.outputMode("update")
.option("checkpointLocation", "s3://checkpoints/orders-hourly")
.trigger(processingTime="30 seconds")
.start("s3://tables/orders-hourly")
)Spark can guarantee that a deterministic query over replayable offsets produces the same committed sink result after failure. That is not the same as exactly-once side effects to arbitrary external services. A foreachBatch sink provides a batch ID so the application can deduplicate writes, but Spark’s documentation warns that foreachBatch is at-least-once by default unless the sink uses that batch ID or another idempotency key.
def write_idempotently(batch_df, batch_id):
(
batch_df
.withColumn("spark_batch_id", lit(batch_id))
.write
.format("jdbc")
.option("url", jdbc_url)
.option("dbtable", "staging_orders")
.mode("append")
.save()
)
# Database-side merge uses (business_key, spark_batch_id) to ignore retries.
run_merge_statement(batch_id)
query = (
events.writeStream
.foreachBatch(write_idempotently)
.option("checkpointLocation", "s3://checkpoints/orders-jdbc")
.start()
)Continuous mode changes the trigger, not the DataFrame API. A supported query can be started with a continuous trigger and checkpoint interval. Spark records progress periodically, and tasks push records through without waiting for the next micro-batch boundary. The mode is attractive for stateless Kafka-to-Kafka transformations where the low-latency target justifies narrower semantics.
continuous = (
spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "broker:9092")
.option("subscribe", "raw-events")
.load()
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "broker:9092")
.option("topic", "clean-events")
.option("checkpointLocation", "s3://checkpoints/clean-events")
.trigger(continuous="1 second")
.start()
)Structured Streaming is a natural fit for landing Kafka, Kinesis, or file-arrival data into Delta, Iceberg, Hudi, or Parquet tables. Micro-batch writes align well with object storage because files are larger, commits are less frequent, and compaction can be scheduled around batch boundaries.
Dashboards that tolerate seconds of latency can use windowed aggregations with watermarks. Spark keeps state for active windows, updates result rows each trigger, and evicts old state once the watermark passes the lateness threshold. This is often simpler than maintaining custom aggregation state in a service.
Many business pipelines enrich live events with relatively static dimension tables. Structured Streaming supports joining streaming DataFrames with static DataFrames, which lets teams reuse data warehouse tables for enrichment while treating the event source as unbounded.
Teams already using Spark ML and DataFrames can compute features from streams with familiar transformations, then reuse related logic for backfills. This reduces training-serving skew when the same query shape can run over historical tables and live events.
Very short trigger intervals reduce waiting time but increase scheduling overhead and sink commit pressure. Very long intervals improve throughput but increase freshness lag and batch size variance. Measure input rate, planning time, processing time, and commit time before locking an interval.
Checkpoint locations are part of the stream’s durable identity. Store them on reliable storage, keep one checkpoint per logical query, and treat incompatible query changes as migrations. Accidentally reusing or deleting a checkpoint can replay old data, skip data, or corrupt state.
Without watermarks, stateful aggregations may need to remember every key forever. A watermark gives Spark permission to evict old window state. Pick the threshold from observed lateness distributions and business tolerance, then publish what happens to records that arrive beyond the bound.
foreachBatch is a practical escape hatch for custom writes, multiple sinks, and unsupported operations, but the application owns semantics. Persist the batch ID, write to staging tables, and merge deterministically so a retried batch cannot duplicate external side effects.