Is Structured Streaming actually streaming or micro-batch?

Micro-batch by default — Spark runs a small job every trigger interval. There is a continuous processing mode with millisecond latency, but it supports only a subset of operations. For most workloads micro-batch at a one-second trigger is both simpler and fast enough.

What does a watermark actually do?

It tells Spark how late a record may arrive before it can be ignored. That bound is what lets Spark drop the state for old windows, so memory stays constant instead of growing forever. Without a watermark, a streaming aggregation keeps every window's state indefinitely.

Why does my streaming query fail on restart?

Usually a changed query with an existing checkpoint. The checkpoint stores the query plan alongside offsets, and incompatible changes — adding a stateful operator, changing the aggregation — are rejected. Use a new checkpoint location for an incompatible change, accepting that state restarts.

Can I use complete output mode with a large aggregation?

Only if the result is small. Complete mode rewrites the entire result table on every trigger, so it has to fit in memory and the write cost grows with result size. Use update or append mode for anything with many groups.

Structured Streaming in PySpark

Structured Streaming applies the DataFrame API to data that never ends. The code looks almost identical to a batch job — which is the point — but three concepts have no batch equivalent: output modes, watermarks, and checkpoints.

A stream from files

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = SparkSession.builder.appName("streaming").master("local[*]").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", 4)

schema = StructType([
    StructField("event_time", TimestampType(), True),
    StructField("user",       StringType(),    True),
    StructField("action",     StringType(),    True),
    StructField("amount",     DoubleType(),    True),
])

stream = (
    spark.readStream
         .schema(schema)                 # required — streams cannot infer
         .option("maxFilesPerTrigger", 1)
         .json("data/stream_in")
)

print("is streaming:", stream.isStreaming)

$ python stream_read.py
is streaming: True

A streaming source must have an explicit schema. Spark cannot sample an unbounded input to infer one.

Writing the stream out

query = (
    stream.filter(F.col("amount") > 10)
          .select("event_time", "user", "amount")
          .writeStream
          .outputMode("append")
          .format("console")
          .option("truncate", False)
          .trigger(processingTime="5 seconds")
          .start()
)
query.awaitTermination(30)

Drop a file into data/stream_in/:

mkdir -p data/stream_in && cat > data/stream_in/batch1.json <<'EOF'
{"event_time": "2026-02-14T10:00:05", "user": "alice", "action": "buy",  "amount": 250.0}
{"event_time": "2026-02-14T10:00:12", "user": "bob",   "action": "buy",  "amount": 5.0}
{"event_time": "2026-02-14T10:00:31", "user": "alice", "action": "buy",  "amount": 120.0}
EOF

-------------------------------------------
Batch: 0
-------------------------------------------
+-------------------+-----+------+
|event_time         |user |amount|
+-------------------+-----+------+
|2026-02-14 10:00:05|alice|250.0 |
|2026-02-14 10:00:31|alice|120.0 |
+-------------------+-----+------+

Bob’s £5 was filtered out. Add a second file and a new batch appears without restarting anything — the query is running continuously and picking up new files each trigger.

Output modes

The mode decides what Spark emits each trigger, and not every mode works with every query.

counts = stream.groupBy("user").agg(F.sum("amount").alias("total"))

query = counts.writeStream.outputMode("append").format("console").start()

pyspark.errors.exceptions.captured.AnalysisException:
Append output mode not supported when there are streaming aggregations on streaming
DataFrames/DataSets without watermark;

Append means “only emit rows that will never change”. A running total per user can always change when a new record arrives, so Spark refuses. Use complete:

query = counts.writeStream.outputMode("complete").format("console").start()

-------------------------------------------
Batch: 0
-------------------------------------------
+-----+-----+
| user|total|
+-----+-----+
|alice|370.0|
|  bob|  5.0|
+-----+-----+

-------------------------------------------
Batch: 1
-------------------------------------------
+-----+-----+
| user|total|
+-----+-----+
|alice|420.0|
|  bob| 35.0|
+-----+-----+

The whole result table, rewritten each trigger. Correct here, unusable with a million users.

Mode	Emits	Works with
`append`	only new, final rows	non-aggregating queries, or aggregations with a watermark
`update`	rows that changed this trigger	most aggregations
`complete`	the entire result table	aggregations with a small result only

update is usually the right choice — it emits only what changed, and works without a watermark.

Windows and watermarks

Aggregate over event time rather than arrival time:

windowed = (
    stream
    .withWatermark("event_time", "2 minutes")
    .groupBy(F.window("event_time", "1 minute"), "user")
    .agg(F.sum("amount").alias("total"), F.count("*").alias("events"))
    .select(F.col("window.start").alias("start"),
            F.col("window.end").alias("end"),
            "user", "total", "events")
)

query = windowed.writeStream.outputMode("append").format("console") \
                .option("truncate", False).start()

-------------------------------------------
Batch: 1
-------------------------------------------
+-------------------+-------------------+-----+-----+------+
|start              |end                |user |total|events|
+-------------------+-------------------+-----+-----+------+
|2026-02-14 10:00:00|2026-02-14 10:01:00|alice|370.0|2     |
|2026-02-14 10:00:00|2026-02-14 10:01:00|bob  |5.0  |1     |
+-------------------+-------------------+-----+-----+------+

append works now. The watermark tells Spark that no record more than two minutes late will be accepted, so once the watermark passes a window’s end that window is final and can be emitted — and its state discarded.

Watch a late record get dropped. Send one timestamped well in the past:

cat > data/stream_in/late.json <<'EOF'
{"event_time": "2026-02-14T09:45:00", "user": "carol", "action": "buy", "amount": 999.0}
EOF

-------------------------------------------
Batch: 3
-------------------------------------------
+-----+---+----+-----+------+
|start|end|user|total|events|
+-----+---+----+-----+------+
+-----+---+----+-----+------+

Nothing. That window closed long ago, so the record was silently dropped. Check the progress metrics to see it:

print(query.lastProgress["stateOperators"][0]["numRowsDroppedByWatermark"])

Monitor that number. It is the only signal that you are discarding real data, and a watermark set too tight will quietly throw away records forever.

The trade is explicit: a longer watermark accepts later data and holds more state; a shorter one bounds memory and drops more. There is no setting that does both.

Checkpoints

A console sink needs no checkpoint. Any real sink does:

query = (
    windowed.writeStream
    .outputMode("append")
    .format("parquet")
    .option("path", "data/stream_out")
    .option("checkpointLocation", "data/checkpoint")
    .trigger(processingTime="10 seconds")
    .start()
)

find data/checkpoint -maxdepth 1 -type d

data/checkpoint
data/checkpoint/offsets
data/checkpoint/commits
data/checkpoint/sources
data/checkpoint/state

offsets records what has been read, commits what has been fully processed, state holds window aggregates. Kill the process and restart it with the same checkpoint location and it resumes exactly where it stopped — no reprocessing, no gap.

Without a checkpoint, a restart reprocesses everything or nothing, depending on the source, and stateful queries lose their state entirely.

Changing the query incompatibly is rejected:

pyspark.errors.exceptions.captured.StreamingQueryException:
Cannot start query with ID ... as it has changed:
Streaming aggregation added to the query. This is not supported.

Use a new checkpoint location for such a change and accept that state restarts from empty.

Reading from Kafka

The most common production source:

kafka_stream = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "events")
    .option("startingOffsets", "latest")
    .load()
)

kafka_stream.printSchema()

root
 |-- key: binary (nullable = true)
 |-- value: binary (nullable = true)
 |-- topic: string (nullable = true)
 |-- partition: integer (nullable = true)
 |-- offset: long (nullable = true)
 |-- timestamp: timestamp (nullable = true)
 |-- timestampType: integer (nullable = true)

key and value are binary — parse them yourself:

parsed = kafka_stream.select(
    F.col("key").cast("string").alias("key"),
    F.from_json(F.col("value").cast("string"), schema).alias("data"),
    "timestamp",
).select("key", "data.*", "timestamp")

(parsed.writeStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "localhost:9092")
       .option("topic", "events-enriched")
       .option("checkpointLocation", "data/kafka_checkpoint")
       .start())

Spark commits Kafka offsets to its own checkpoint, not to a Kafka consumer group, which is why kafka-consumer-groups.sh shows no lag for a Spark reader. Monitor the query’s own progress metrics instead.

Monitoring

import json
print(json.dumps(query.lastProgress, indent=2)[:600])

{
  "id": "8f2a1c94-3e7b-4d21-9a5c-6b1e2f9d3a70",
  "batchId": 4,
  "numInputRows": 1250,
  "inputRowsPerSecond": 125.0,
  "processedRowsPerSecond": 410.2,
  "durationMs": {
    "addBatch": 2104,
    "triggerExecution": 3047
  },
  "stateOperators": [
    {
      "numRowsTotal": 842,
      "numRowsUpdated": 120,
      "numRowsDroppedByWatermark": 0,
      "memoryUsedBytes": 1048576
    }
  ]
}

The three numbers that matter: processedRowsPerSecond must stay above inputRowsPerSecond or you fall behind permanently; numRowsDroppedByWatermark must stay at zero; stateOperators[].memoryUsedBytes must stay flat rather than growing, which it will if your watermark is missing or too long.

Practice

1. Run a streaming aggregation in append mode with no watermark.

AnalysisException: Append output mode not supported when there are streaming
aggregations on streaming DataFrames/DataSets without watermark

Without a watermark Spark can never decide a group is final, so append has nothing it is allowed to emit. It fails at query-start rather than producing wrong results — the right behaviour.

2. Set a 10-second watermark and send a record 30 seconds late. Then set 5 minutes.

watermark 10s: numRowsDroppedByWatermark = 1
watermark 5m:  numRowsDroppedByWatermark = 0, memoryUsedBytes 4.2 MB -> 61 MB

The direct trade. Set the watermark from measured arrival delay in your actual pipeline, not from a guess — and alert on the dropped count.

3. Stop a query with a checkpoint, add three files, and restart it.

Batch: 5
numInputRows: 3

It processes exactly the three new files. The checkpoint’s offsets directory recorded which files were already consumed, so nothing is reprocessed and nothing is skipped.

4. Use trigger(once=True) instead of a processing time.

It processes all available data in one batch, then stops. This makes a streaming query behave like an incremental batch job — you get checkpoint-based exactly-once semantics and no cluster running between runs. Scheduling it hourly is often cheaper and simpler than a continuously running stream, and availableNow=True is the modern equivalent that processes the backlog in several batches instead of one.

That completes the PySpark track: from a first DataFrame through joins, windows, the execution model, tuning, and streaming.