TopolearnStreaming ETL and near-real-time metrics on Kafka or files use the same DataFrame API you already know.
Structured Streaming treats a stream as an unbounded table that grows over time. You write a normal DataFrame query, and Spark runs it incrementally as new data arrives.
- Sources:
spark.readStream.format("kafka" | "parquet" | "json" | "rate" ...). File sources need a schema. - Sinks:
df.writeStream.format("parquet" | "kafka" | "delta" | "console" | "memory"), orforeachBatch(func)to run any batch code per micro-batch, such as a MERGE into a table. - Output mode:
append(only new final rows),update(changed rows),complete(the whole result, for small aggregations). - Trigger: default micro-batches as fast as possible,
processingTime="1 minute", oravailableNow=Trueto process everything available and stop, which is great for scheduled incremental jobs. - Checkpoint location (
option("checkpointLocation", path)): stores offsets and state so a restarted query continues exactly where it left off. Each query needs its own.
Aggregations over event time keep state. A watermark, withWatermark("ts", "10 minutes"), tells Spark how late data may arrive, so it can finalize old windows, emit them in append mode and drop their state. Without one, state grows forever.
query.status, query.lastProgress and the Structured Streaming tab in the UI monitor a running query; query.stop() ends it.
Example
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.master("local[2]").getOrCreate()
import json, os, shutil
shutil.rmtree("in", ignore_errors=True); os.makedirs("in")
for i, ts in enumerate(["2026-01-01 10:01:00", "2026-01-01 10:04:00", "2026-01-01 10:12:00"]):
with open(f"in/e{i}.json", "w") as f:
json.dump({"ts": ts, "user": f"u{i % 2}", "amount": 10 * (i + 1)}, f)
events = (spark.readStream.schema("ts TIMESTAMP, user STRING, amount INT").json("in")
.withWatermark("ts", "5 minutes"))
per_window = events.groupBy(F.window("ts", "10 minutes")).agg(F.sum("amount").alias("total"))
query = (per_window.writeStream.outputMode("update").format("memory")
.queryName("totals").option("checkpointLocation", "ckpt_stream")
.trigger(availableNow=True).start())
query.awaitTermination()
rows = spark.sql("SELECT window.start AS start, total FROM totals ORDER BY start").collect()
print([(r.start.strftime("%H:%M"), r.total) for r in rows])
# [('10:00', 30), ('10:10', 30)]Remember
- Streams are unbounded tables processed incrementally
- Checkpoints make restarts exactly-once; one location per query
- Watermarks bound state and let windows be finalized
Watch out for
- Running a stateful streaming aggregation without a watermark, so state grows until the job fails.