Spark Structured Streaming

Spark Structured Streaming is the stream-processing layer of Apache Spark. It represents an unbounded input stream as an incrementally updated table, lets an application describe the computation with the DataFrame or Dataset API, and repeatedly executes that computation as new records arrive.

Structured Streaming is not a separate cluster service. A streaming query runs inside a normal Spark application: the driver plans and coordinates the query, executors run tasks, and a cluster manager such as Kubernetes, YARN, or Spark Standalone supplies resources. The application remains alive while one or more streaming queries are active.

Execution Model #

Structured Streaming uses the Spark SQL engine for both batch and streaming queries. The same expressions, projections, filters, aggregations, and joins therefore participate in SQL analysis, Catalyst optimization, and physical planning.

The default execution mode is micro-batch processing:

1. Spark asks each source which new input positions are available.
2. The driver chooses an input range for the next micro-batch.
3. Executors process that finite range as a distributed Spark job.
4. Stateful operators update their state stores.
5. The sink commits the batch output.
6. Spark records progress in the query checkpoint.

This model normally produces latency from hundreds of milliseconds to seconds. It is a good fit for event pipelines, incremental ETL, monitoring, and continuously refreshed aggregates. Spark also has a continuous processing mode with lower latency and at-least-once semantics, but it supports a smaller set of operations and is not the usual default.

The following terms describe different layers:

One Spark application can run multiple streaming queries. Each query has its own lifecycle and must have a distinct checkpoint location.

The Unbounded-Table Model #

Spark treats the input stream as rows appended to an unbounded input table. A query transforms that input into a result table. Every trigger incorporates newly available rows and updates the result.

Consider a running total by customer:

SELECT customer_id, SUM(amount)
FROM orders
GROUP BY customer_id

Each new order changes one row in the conceptual result table. How those changes reach a sink depends on the output mode:

A stateless query can normally use append mode because an emitted row never changes. An aggregation without a watermark cannot generally append final rows because Spark cannot know that another record for a group will never arrive. Update and complete modes expose changing results, while event-time windows and watermarks can eventually make old aggregate rows final.

Not every output mode is valid for every query and sink. Spark analyzes the complete plan and rejects unsupported combinations when the query starts.

Create and Control a Query #

A streaming read creates a DataFrame whose isStreaming property is true. Transformations remain lazy. Calling writeStream.start() constructs and starts the streaming query:

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.Trigger

val spark = SparkSession.builder()
  .appName("order-stream")
  .getOrCreate()

val input = spark.readStream
  .format("rate")
  .option("rowsPerSecond", 10)
  .load()

val query = input
  .selectExpr("timestamp", "value * 2 AS doubled_value")
  .writeStream
  .queryName("doubled-values")
  .format("console")
  .outputMode("append")
  .trigger(Trigger.ProcessingTime("5 seconds"))
  .start()

query.awaitTermination()

Common trigger policies are:

• Processing-time trigger: start a micro-batch at a configured interval.
• Default trigger: start the next micro-batch as soon as the previous one finishes.
• Available-now trigger: process all data available when the query starts, potentially in multiple batches, and then stop. This is useful for scheduled incremental jobs.
• Continuous trigger: use continuous processing when the query and sink support it.

Stopping the driver stops its queries unless another deployment mechanism restarts the application. A production supervisor should restart a failed driver with the same application code and checkpoint location.

Sources and Sinks #

Sources expose new data and durable positions to Spark. Sinks accept the rows emitted for a completed trigger.

Common sources include:

• Kafka;
• files arriving in a directory;
• Delta Lake and other data-source extensions;
• sockets and the rate source for development; and
• custom Data Source APIs.

Common sinks include:

• files;
• Kafka;
• table formats and databases supplied by connectors;
• foreachBatch for custom per-batch writes;
• foreach for per-row writer logic; and
• console and memory sinks for development.

The source, output mode, checkpoint, and sink together determine the actual recovery and delivery guarantees. The phrase “Structured Streaming is exactly once” is not an unconditional property of every external side effect.

Add the Kafka Connector #

Kafka support is distributed separately from Spark SQL. Match the connector version and Scala binary suffix to the Spark distribution:

spark-submit \
  --packages "org.apache.spark:spark-sql-kafka-0-10_2.13:${SPARK_VERSION}" \
  --class example.OrderPipeline \
  order-pipeline.jar

PySpark applications need the same JVM connector:

spark-submit \
  --packages "org.apache.spark:spark-sql-kafka-0-10_2.13:${SPARK_VERSION}" \
  order_pipeline.py

The artifact name retains 0-10 for historical compatibility; it supports current Kafka brokers rather than only Kafka 0.10. Broker, Spark, Scala, and connector compatibility should still be checked for the deployed versions.

Read from Kafka #

Use subscribe for a fixed topic list, subscribePattern for topics matching a regular expression, or assign for explicit topic partitions. Exactly one of these options is allowed:

val kafkaInput = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker-1:9092,broker-2:9092")
  .option("subscribe", "orders")
  .option("startingOffsets", "earliest")
  .option("maxOffsetsPerTrigger", 500000)
  .load()

Each Kafka record becomes a row with this shape:

Kafka keys and values are bytes. Cast textual data explicitly and parse structured values with a declared schema:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

val orderSchema = new StructType()
  .add("order_id", StringType, nullable = false)
  .add("customer_id", StringType, nullable = false)
  .add("amount", DecimalType(18, 2), nullable = false)
  .add("event_time", TimestampType, nullable = false)

val orders = kafkaInput
  .select(
    col("key").cast("string").as("kafka_key"),
    from_json(col("value").cast("string"), orderSchema).as("event"),
    col("topic"),
    col("partition"),
    col("offset"),
    col("timestamp").as("kafka_timestamp")
  )
  .select("kafka_key", "event.*", "topic", "partition", "offset", "kafka_timestamp")

Do not silently discard malformed records. Preserve enough Kafka metadata to diagnose and replay them, then route invalid rows to an explicit error sink:

val decoded = kafkaInput
  .select(
    col("key"),
    col("value"),
    col("topic"),
    col("partition"),
    col("offset"),
    from_json(col("value").cast("string"), orderSchema).as("event")
  )

val validOrders = decoded.filter(col("event").isNotNull)
val invalidOrders = decoded.filter(col("event").isNull)

The startingOffsets option only controls a newly created query. Once a checkpoint exists, recovery resumes from progress recorded in that checkpoint. Changing startingOffsets does not override it.

Event Time and Watermarks #

Processing time is the time at which Spark handles a record. Event time is a timestamp carried by the record that describes when the event occurred. Network delays, retries, offline devices, and Kafka backlog can make these times differ substantially.

Group by an event-time window when the business question is based on event occurrence:

val revenueByWindow = orders
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window(col("event_time"), "5 minutes"),
    col("customer_id")
  )
  .agg(sum("amount").as("revenue"))

The watermark tells Spark how late data may arrive before the engine can retire old state. Conceptually, it trails the greatest event time observed by the configured delay. It is not the current wall-clock time and it is not a promise that every record within the delay will be accepted immediately.

For a windowed aggregation:

• records earlier than the current watermark may be dropped from that stateful operation;
• Spark retains a window while later matching records may still arrive;
• once the watermark passes the applicable window boundary, Spark can remove its state; and
• append mode can emit the final window result when Spark knows it will no longer update it.

Without a watermark or another bounded-state policy, state for old keys, windows, joins, or deduplication may grow indefinitely.

Keyed and Stateful Processing #

Spark does not expose a long-lived KeyedStream type equivalent to Flink. The closest abstractions are grouped DataFrames or KeyValueGroupedDataset values:

val byCustomer = typedOrders.groupByKey(_.customerId)

Rows with the same grouping key are shuffled to the same state partition. This enables per-key state, but it does not collect an unbounded key group into memory all at once.

Built-in stateful operations include:

• aggregations;
• event-time windows;
• stream-stream joins;
• dropDuplicatesWithinWatermark; and
• arbitrary keyed state transformations.

For new arbitrary stateful logic, transformWithState provides ValueState, ListState, and MapState, state time-to-live policies, processing-time or event-time timers, and optional initial state. A stateful processor handles the rows for one key and may emit zero or more output rows. Older applications may use mapGroupsWithState, flatMapGroupsWithState, or Python’s applyInPandasWithState.

State has operational cost. Choose bounded keys, define expiration semantics, monitor state-store size, and avoid storing entire event histories when an aggregate or compact summary is sufficient.

Streaming Joins #

Structured Streaming supports stream-static and stream-stream joins.

A stream-static join enriches streaming rows from a static DataFrame:

val customers = spark.read.parquet("/warehouse/customers")

val enrichedOrders = orders.join(
  broadcast(customers),
  Seq("customer_id"),
  "left"
)

The static side is read as part of the query plan. It does not automatically become a continuously changing table merely because its underlying files change.

A stream-stream join must retain unmatched rows from one or both inputs until a future match becomes impossible. Bound that state with watermarks and an event-time range:

val matched = orders
  .withWatermark("event_time", "15 minutes")
  .as("o")
  .join(
    payments
      .withWatermark("payment_time", "15 minutes")
      .as("p"),
    expr("""
      o.order_id = p.order_id AND
      p.payment_time >= o.event_time AND
      p.payment_time <= o.event_time + INTERVAL 30 MINUTES
    """),
    "leftOuter"
  )

Join type, watermark placement, and time constraints determine whether Spark can clean up state and when it can emit unmatched outer-join rows.

Checkpoints and Kafka Offsets #

A checkpoint is the durable identity and recovery state of a query. Store it on storage visible to every possible driver and executor, such as HDFS or a suitably configured object store:

val query = revenueByWindow.writeStream
  .format("parquet")
  .option("path", "s3a://analytics/revenue-by-window")
  .option(
    "checkpointLocation",
    "s3a://analytics-checkpoints/revenue-by-window"
  )
  .outputMode("append")
  .start()

The checkpoint contains:

• source offsets processed by each batch;
• batch commit metadata;
• query metadata; and
• state-store snapshots and change data for stateful operators.

For a Kafka source, Spark’s checkpoint is the recovery authority. The query does not depend on Kafka consumer-group committed offsets to decide where a recovered batch starts. Kafka must nevertheless retain source records long enough for the query to catch up or recover; a checkpoint cannot reconstruct records that Kafka has already deleted.

Use one checkpoint directory for exactly one logical query. Reusing a directory for an unrelated query can corrupt semantics or fail query initialization. Many changes to a stateful query are incompatible with an existing checkpoint, including changes to grouping keys, state schemas, join structure, or the number and kind of stateful operators. Treat such a change as a state migration, not a routine redeployment.

Processing and Delivery Guarantees #

Structured Streaming can replay a deterministic micro-batch after failure. Checkpointed source positions and state-store commits prevent that replay from being counted twice inside the query’s managed state. End-to-end guarantees still depend on the sink:

Exactly-once processing does not mean that Spark starts a distributed transaction across every external system. To make a custom sink idempotent, use a stable event identifier, an upsert key, a transactional destination, or the batchId supplied to foreachBatch.

The application logic must also be deterministic under replay. Generating random identifiers or calling an external service from an ordinary transformation can produce a different result when Spark retries the same input.

Write to Kafka #

The Kafka sink requires a value column and accepts optional key, headers, topic, and partition columns. Keys and values must be strings or binary values:

val kafkaOutput = revenueByWindow.select(
  col("customer_id").cast("string").as("key"),
  to_json(
    struct(
      col("customer_id"),
      col("window"),
      col("revenue")
    )
  ).as("value")
)

val query = kafkaOutput.writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker-1:9092,broker-2:9092")
  .option("topic", "customer-revenue")
  .option(
    "checkpointLocation",
    "hdfs:///checkpoints/customer-revenue"
  )
  .outputMode("append")
  .start()

The checkpoint prevents Spark from intentionally advancing past an uncommitted batch, but a Kafka producer retry can still write duplicate records. Give each result a deterministic key or event ID when consumers need deduplication. A compacted topic retains the latest record for a key eventually, but compaction is not immediate and does not prevent consumers from observing duplicates before compaction.

See Apache Kafka for topic, partition, offset, retention, and compaction semantics.

Use foreachBatch Carefully #

foreachBatch exposes each micro-batch as an ordinary batch DataFrame and supplies a monotonically increasing batch ID:

val query = revenueByWindow.writeStream
  .foreachBatch { (batch, batchId) =>
    batch.persist()

    batch.write
      .mode("append")
      .parquet("/warehouse/revenue-history")

    batch.write
      .format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "current_revenue")
      .mode("append")
      .save()

    batch.unpersist()
  }
  .option(
    "checkpointLocation",
    "hdfs:///checkpoints/revenue-multi-sink"
  )
  .start()

This example is not atomic across Parquet and JDBC. A failure after the first write can cause Spark to rerun the batch and repeat that write. Production code should record batchId transactionally in each destination, use idempotent upserts, or avoid pretending that independent sinks form one transaction.

Do not call actions repeatedly without considering recomputation. Persist the batch when several writes reuse it, then unpersist it before the callback returns.

Partitioning and Throughput #

Kafka and Spark have separate notions of partitioning:

• Kafka partitions determine source ordering and the natural units from which records are fetched.
• Spark shuffle partitions determine parallelism after aggregations, joins, and repartitioning.
• A Kafka record key influences Kafka partition placement, but parsing that key does not automatically make the Spark DataFrame partitioned by the same logical key.
• groupBy, groupByKey, joins, and explicit repartition operations may introduce a Spark shuffle.

Important controls include:

• maxOffsetsPerTrigger to cap the total Kafka records selected for a trigger;
• trigger interval to balance scheduling overhead and latency;
• spark.sql.shuffle.partitions or adaptive settings for shuffle parallelism;
• executor count, cores, memory, and locality;
• state-store provider and memory configuration; and
• sink throughput and transaction latency.

If input arrives faster than completed batches consume it, scheduling delay and Kafka lag grow. Increasing the trigger frequency cannot fix a pipeline whose sustained processing capacity is below its input rate. Examine source ingestion, shuffle skew, state size, garbage collection, and sink latency before adding executors blindly.

Monitoring and Operations #

StreamingQuery exposes current status, recent progress, exceptions, and termination:

println(query.status)
println(query.lastProgress)
println(query.exception)

The Spark UI and StreamingQueryListener expose metrics such as:

• input rows per second;
• processed rows per second;
• trigger and batch duration;
• source start and end offsets;
• event-time watermark;
• state row count and state-store memory;
• shuffle and spill statistics; and
• sink commit duration.

Monitor the external systems as well. Kafka consumer lag, broker throttling, retention headroom, destination errors, filesystem commit latency, and checkpoint-storage health are not fully described by Spark executor metrics.

Operational alerts should distinguish:

1. no input because producers are idle;
2. no input because the source is inaccessible;
3. slow but progressing batches;
4. a query that terminated with an exception; and
5. a driver that is no longer running.

Deployment #

Package a Structured Streaming application like any other Spark application, include connector dependencies, and submit it to the chosen cluster manager:

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --packages "org.apache.spark:spark-sql-kafka-0-10_2.13:${SPARK_VERSION}" \
  --class example.OrderPipeline \
  order-pipeline.jar

Cluster mode is generally preferable for production because the cluster manager supervises the driver. Configure an application restart policy, but ensure every restarted instance uses the same durable checkpoint and that only one active instance owns a given query checkpoint.

Graceful shutdown should call StreamingQuery.stop() or stop the Spark session and allow in-flight work to finish where the deployment environment permits it. Abrupt termination remains recoverable when the source, state, checkpoint store, and sink satisfy their recovery contracts.

Structured Streaming and Flink #

Spark Structured Streaming and Flink solve overlapping problems but expose different mental models:

Choose based on workload, latency, state complexity, operational platform, and team expertise rather than treating either engine as universally better.

Common Failure Modes #

Reusing a Checkpoint for a Different Query #

A checkpoint belongs to one logical query. Start incompatible code with a new checkpoint and explicitly plan how historical input or state will be migrated.

Assuming startingOffsets Overrides Recovery #

It applies only when no checkpoint progress exists. Delete or replace a checkpoint only when intentionally starting a new query and after deciding whether reprocessing or data loss is acceptable.

Omitting a Watermark #

Stateful event-time queries can retain state indefinitely. Define lateness requirements from actual producer behavior and monitor dropped late data and state growth.

Claiming Exactly Once for an Arbitrary Sink #

Spark cannot make an ordinary HTTP call, email, JDBC insert, or Kafka producer write globally transactional merely by enabling a checkpoint. Design the destination operation to be idempotent.

Confusing Kafka and Spark Keys #

A Kafka key controls Kafka partitioning. Spark must still group or repartition by the desired application key before keyed aggregation or custom stateful processing.

Treating the Console Sink as Production Storage #

The console and memory sinks are debugging tools. They do not provide a durable serving layer.

Changing Stateful Logic in Place #

State schemas and operator identities are encoded in the checkpoint. Validate whether an upgrade is compatible; otherwise, use a new checkpoint and an explicit bootstrap or replay procedure.

Design Checklist #

Before deploying a query, answer:

1. What identifies an event, and how are duplicates handled?
2. Which timestamp is event time, and how late can valid data arrive?
3. What bounds every stateful operator’s state?
4. Which output mode does the result require?
5. What does the selected sink guarantee during retries?
6. Where is the checkpoint stored, retained, and monitored?
7. How long does the source retain data relative to worst-case recovery time?
8. Which query changes are compatible with the existing checkpoint?
9. What input rate, state size, and sink latency can the deployment sustain?
10. Which metrics and alerts demonstrate both progress and correctness?

References #

• Structured Streaming Programming Guide
• Structured Streaming and Kafka Integration Guide
• transformWithState Programming Guide
• Structured Streaming API Documentation