streaming

Two streaming APIs ship with Spark:

• Structured Streaming is the strategic API. It is implemented in sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ and exposed through DataStreamReader / DataStreamWriter on SparkSession.
• DStreams is the legacy micro-batch API in streaming/. It is in maintenance mode: bug fixes only, no new features.

This page focuses on Structured Streaming and gives a short overview of DStreams at the end.

Structured Streaming

Purpose

• Express streaming computations as DataFrame queries that get executed incrementally.
• Provide exactly-once semantics on top of replayable sources and idempotent sinks.
• Maintain stateful operators (aggregations, joins, deduplication) with pluggable durable state stores.

Directory layout

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/
  StreamExecution.scala            - base class for the streaming runtime
  MicroBatchExecution.scala        - default execution mode (default in production)
  ContinuousExecution.scala        - low-latency continuous mode
  StreamingQueryManager.scala      - registry of running queries on a SparkSession
  StreamingQueryListener.scala     - event hooks
  Source.scala / Sink.scala        - V1 streaming source/sink interfaces
  sources/                         - file source, rate source, memory source, ...
  state/                           - the state store framework
  state/HDFSBackedStateStoreProvider.scala
  state/RocksDBStateStoreProvider.scala
  StatefulOperator.scala           - base for stateful exec operators
  WatermarkSupport.scala           - watermark plumbing
  StreamingSymmetricHashJoinExec.scala
  FlatMapGroupsWithStateExec.scala
  TransformWithStateExec.scala     - the new flexible stateful API in Spark 4
  ...

sql/api/src/main/scala/org/apache/spark/sql/streaming/   - public DSL (DataStreamReader/Writer)

Key abstractions

Type	What it is
StreamingQuery (sql/core/.../streaming/StreamingQuery.scala)	The handle returned by start(). Wraps a StreamExecution.
StreamExecution	Per-query execution loop. Owns the offset log and commit log.
MicroBatchExecution	Default mode. Splits time into micro-batches.
ContinuousExecution	Long-running tasks that emit records as they arrive.
Source / MicroBatchStream (V2)	Pluggable input.
Sink / StreamingWrite (V2)	Pluggable output.
StateStore, StateStoreProvider	Durable per-key state.
StreamingQueryManager	Per-SparkSession registry of queries.
StreamingQueryListener	User callback for query progress events.

Micro-batch execution

sequenceDiagram
    participant U as User code
    participant SQM as StreamingQueryManager
    participant MBE as MicroBatchExecution
    participant SRC as Source
    participant ENG as Spark engine
    participant SINK as Sink

    U->>SQM: ds.writeStream.start()
    SQM->>MBE: spawn execution thread
    loop Each batch
        MBE->>SRC: getOffset / planInputPartitions
        MBE->>MBE: build incremental LogicalPlan
        MBE->>ENG: execute (Catalyst optimize + physical plan)
        ENG->>SINK: write rows for this batch
        MBE->>MBE: commit batch (offset log + commit log)
    end

MicroBatchExecution writes two write-ahead logs into the configured checkpoint location:

• offsets/ - the source offsets that bound each batch.
• commits/ - markers indicating each batch was successfully written by the sink.

A query that crashes is restarted from the highest committed batch, replaying from the recorded offsets.

State store

Stateful operators (StreamingAggregationExec, StreamingDeduplicateExec, StreamingSymmetricHashJoinExec, FlatMapGroupsWithStateExec, TransformWithStateExec) keep per-key state across batches. The state-store framework is in sql/core/.../execution/streaming/state/:

• HDFSBackedStateStoreProvider - the original provider; writes deltas and snapshot files to the checkpoint directory.
• RocksDBStateStoreProvider - the recommended provider for large-state queries.

State is partitioned by the operator's grouping key and written in batched commits aligned with the offset log.

Watermarks and output modes

WatermarkSupport (sql/core/.../execution/streaming/WatermarkSupport.scala) tracks the maximum event time seen and applies the user's allowed lateness to decide when state can be evicted.

Output modes:

• Append - emit only new rows whose result is final.
• Update - emit rows that changed since the last batch.
• Complete - re-emit the entire result table each batch (only for full aggregations).

Spark 4: TransformWithState

TransformWithStateExec (sql/core/.../execution/streaming/TransformWithStateExec.scala) introduces a more flexible stateful API that exposes value, list, and map state primitives to user code. It is implemented on top of the same state-store framework but offers TTL, schema evolution, and broader access patterns.

Connectors

• Kafka source/sink: connector/kafka-0-10-sql/.
• Kinesis: connector/kinesis-asl/.
• File source: built in (CSV, JSON, Parquet, Avro, ORC, text).
• Memory and rate sources: built in (for testing).

Legacy DStream API

streaming/ houses the original DStream API. Its core classes are:

• StreamingContext (streaming/.../StreamingContext.scala) - the streaming-specific context; wraps a SparkContext.
• DStream (streaming/.../dstream/DStream.scala) - a sequence of RDDs over time.
• Receiver (streaming/.../receiver/Receiver.scala) - long-running task that pulls data into the cluster.

DStreams are still supported for existing pipelines but Structured Streaming is the recommended API for new work. The classes carry @deprecated-style notes in places but the public API has not been removed.

Integration points

• Reuses sql/core for plan execution and core/ for scheduling.
• The state store interacts with the BlockManager only indirectly - state files live on the configured checkpoint filesystem (HDFS, S3, GCS, ABFS, ...).
• Each running query exposes metrics on the SparkListenerBus and through the StreamingQuery.lastProgress API.
• The Connect server exposes streaming via commands.proto's WriteStreamOperationStart / StreamingQueryCommand messages.

Entry points for modification

• Add a new built-in source: implement MicroBatchStream (V2) under sql/core/.../execution/streaming/sources/.
• Add a new state-store provider: implement StateStoreProvider and register it via spark.sql.streaming.stateStore.providerClass.
• Add a stateful operator: subclass StateStoreWriter / StateStoreReader and emit it from a new SparkPlanner strategy.