Structured Streaming Fundamentals

Structured Streaming is the part of Spark that works on data that never stops arriving. Orders, clicks and sensor readings are examples.

You write the same DataFrame code you already know. Spark runs it again and again on the new rows.

You met the streaming ideas in four data-pipeline chapters. They are "Batch vs Streaming vs Micro-Batch", "Streaming Architecture Fundamentals", "Stream Processing" and "Delivery Guarantees & Ordering". Here you use them in Spark.

This chapter uses the shop stream. Orders 110 to 114 and 150 and 151 arrive on 2026-01-22. Spark reads them from a Kafka topic called orders.

The chapter has nine parts.

1. The model: a stream is a table that keeps growing.
2. Sources: where the rows come from.
3. Sinks: where the results go.
4. Triggers: when Spark runs.
5. Output modes: which rows Spark writes each time.
6. The checkpoint: how a query survives a crash.
7. foreachBatch: writing with your own code.
8. Exactly-once from end to end.
9. Monitoring a running query.

Windows and watermarks are in the next chapter, "Event-Time, Windows, Watermarks & Stateful Streaming".

Think of the stream as one table. New rows are added at the bottom and the table never stops growing. This is the unbounded table.

You write one query on that table. Each time Spark runs, it reads only the new rows and updates the result. One run is called a micro-batch.

Here is a query on the shop stream. It keeps orders of 10.00 or more. The stream is called orders, and the next section shows how to build it.

big = (orders
    .filter("amount >= 10")
    .select("order_id", "customer_id", "amount"))

query = (big.writeStream
    .format("console")
    .outputMode("append")
    .option("checkpointLocation", "/chk/big-orders")
    .start())

Look at the code again. Only writeStream is new here. The filter and select in the middle are the same as in batch.

The query keeps running until you stop it. The call start() launches it and returns at once.

A source is where a stream starts. You use spark.readStream instead of spark.read. The rest of the code looks the same.

Kafka is the most common source. Each Kafka record arrives as a row with fixed columns. The column value holds the message as raw bytes (binary).

A Kafka topic is split into partitions, and each partition is one ordered log, as in "Streaming Architecture Fundamentals". To keep it simple, the orders topic here has one partition. An offset is the position of a record in that partition.

Here are the first four orders in the topic, at offsets 0 to 3.

from pyspark.sql.functions import from_json, col

raw = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("subscribe", "orders")
    .option("startingOffsets", "earliest")
    .load())

schema = "order_id INT, customer_id INT, amount DECIMAL(10,2), order_ts TIMESTAMP"

orders = (raw
    .select(from_json(col("value").cast("string"), schema).alias("o"))
    .select("o.*"))

The option startingOffsets says where a new query starts: "earliest" reads the whole topic, "latest" only new records. For a stream, the default is "latest".

It matters only the first time. After a restart, Spark continues from the position saved in the checkpoint. You meet the checkpoint soon.

Kafka is not the only source.

A sink is where the results go. You use writeStream instead of write. You call .format(...) to pick the sink.

Some sinks are for real pipelines. Others are only for tests.

Safe to replay means the same batch can run twice and leave the same result. The later parts of this chapter explain why that matters.

The Kafka sink needs a column named value, as text or bytes. To send an order as JSON, build it with to_json(struct(...)).

A trigger decides when the next micro-batch starts. You set it with .trigger(...) on writeStream.

The trigger availableNow replaces the older once trigger. Look at a job that runs three times, with the same checkpoint.

query = (orders.writeStream
    .format("parquet")
    .option("path", "/lake/orders")
    .option("checkpointLocation", "/chk/orders")
    .trigger(availableNow=True)
    .start())
query.awaitTermination()      # returns when the data is read

The job does not need to run all day, but it still has all the streaming features. It saves its position and its state in the checkpoint.

An output mode decides which result rows Spark writes after each micro-batch. There are three.

• Append: only rows that are new. Spark never changes a row it has written.
• Update: only rows that are new or changed in this batch.
• Complete: the whole result table, every batch.

See the difference on a query that adds up the amount for each customer. A running total changes when a new order arrives.

totals = (orders
    .groupBy("customer_id")
    .sum("amount")
    .withColumnRenamed("sum(amount)", "total"))

# run once with .outputMode("update"),
# and once with .outputMode("complete")

Append mode is not on that table. A total can still change, so Spark cannot write it as final. Spark refuses to start the query and shows an error.

Each mode works only for some queries. A watermark tells Spark how long to wait for late rows. The next chapter explains it.

A checkpoint is a folder where the query saves its progress. You set it with the option checkpointLocation. Use a path on shared, reliable storage.

A small metadata file keeps the query id. Spark also saves three main things there.

• Offsets: the range of source data each batch will read. Spark writes this before it processes the batch.
• Commits: a mark that says a batch is finished.
• State: the running values of an aggregation, such as the totals per customer.

After a crash, Spark reads the checkpoint and finishes what was not finished. Look at a crash in the middle of batch 2.

# the query crashes while it writes batch 2
# you start the same code again, with the same checkpointLocation

Follow four rules for the checkpoint.

1. Use one checkpoint folder for each query. Never share it between two queries.
2. Never delete it. Without it, the query starts as a new one, and the old state is gone.
3. Some changes to a query are safe and some break the restart. A different grouping column, for example, usually fails.
4. When you are not sure, use a new checkpoint folder and plan how to handle the old output.

Some writes have no streaming sink, such as a write to a database table or a MERGE into a Delta table. A MERGE adds or changes rows by a key.

For those, use foreachBatch. Spark calls your function once for each micro-batch.

The function gets two things. One is a normal DataFrame with the rows of that batch. The other is the batch_id, a number that counts the batches.

There is a catch. After a crash, Spark runs the same batch again with the same batch_id. Your write must give the same result when it runs twice.

Orders 150 and 151 are in batch 2. The table already holds orders 110 to 114, so it has 5 rows. In both versions, the write finishes and then the query crashes before the commit.

Here are the two versions of the function.

# Version A: plain append
def save(batch_df, batch_id):
    batch_df.write.mode("append").saveAsTable("orders_copy")

# Version B: MERGE (Delta table), skips orders that exist
def save(batch_df, batch_id):
    batch_df.createOrReplaceTempView("batch")
    batch_df.sparkSession.sql("""
        MERGE INTO orders_copy t USING batch s
        ON t.order_id = s.order_id
        WHEN NOT MATCHED THEN INSERT *""")

orders.writeStream.foreachBatch(save) \
    .option("checkpointLocation", "/chk/copy").start()

Another way to be safe is to save each batch_id in a log table. Skip a batch if its id is already in the log. The log write and the data write must succeed or fail together.

Recap from "Delivery Guarantees & Ordering": at-least-once means no row is lost but a row can arrive twice. Exactly-once means the result looks as if each row was processed one time.

Structured Streaming can reach exactly-once. It needs three parts, and each part has one job.

Take one part away and the promise breaks. The Kafka sink and a plain foreachBatch append can write a batch twice. That gives at-least-once.

Exactly-once does not mean your code runs once. A batch can run twice. The sink makes sure the second run changes nothing.

The object returned by start() is your handle on the query. The property query.lastProgress shows the report of the last finished batch.

Here is a shortened report for batch 2 of the shop stream. The two rate numbers are made up.

query.lastProgress
# {
#   "batchId": 2,
#   "numInputRows": 2,
#   "inputRowsPerSecond": 0.4,
#   "processedRowsPerSecond": 8.0,
#   "sources": [{
#     "startOffset": {"orders": {"0": 5}},
#     "endOffset":   {"orders": {"0": 7}},
#     "latestOffset": {"orders": {"0": 7}}
#   }]
# }

Compare the two rate numbers. Say 500 rows arrive each second but Spark finishes only 300. The query falls 200 rows behind every second.

After 10 minutes, which is 600 seconds, it is 200 × 600 = 120,000 rows behind. You need more parallel work or faster code. More Kafka partitions let Spark read with more tasks.

Five points to keep.

• A stream is a table that keeps growing. Your query runs on the new rows of each micro-batch.
• readStream picks the source, writeStream picks the sink, and the middle is normal DataFrame code.
• The trigger sets when Spark runs. The output mode sets which rows it writes.
• The checkpoint stores offsets, commits and state. Use one per query and never delete it.
• Exactly-once needs a replayable source, a checkpoint and a sink that is safe to replay.

Answer these without looking back.

1. A Kafka value is binary. What two steps turn it into columns?
2. Which output mode needs an aggregation, and which needs a watermark when there is one?
3. In update mode, what does batch 2 print for the totals per customer?
4. What does the availableNow trigger do, and why is it useful?
5. Why is a plain append inside foreachBatch not safe after a crash?