Databricks Spark Developer Associate: Structured Streaming

The pipeline

spark.readStream → DataFrame (an unbounded table) → transformations → writeStream. Common sources are Kafka, Delta Lake, and file/JSON/CSV directories; sinks are Delta/file, Kafka, foreachBatch, and memory/console for debugging. File sources need an explicit schema - readStream cannot infer from an empty directory reliably.

Modes and triggers

• append - only new rows are written; for aggregations it requires a watermark so Spark knows when a window is final.
• update - only rows that changed are written.
• complete - the entire result table is rewritten each batch; not all sinks support it.

Triggers: default (as fast as possible), processingTime="1 minute", availableNow=True (process all available data then stop - the Databricks default for many workloads), and once=True.

State, watermark, checkpoint

A checkpoint location stores offsets, commit metadata and state. It is mandatory for stateful operations and gives exactly-once semantics with replayable sources and idempotent sinks. withWatermark("event_time", "10 minutes") bounds state and defines which records count as late; late rows outside the watermark are dropped.

Traps

• Reusing one checkpoint directory for two different queries fails the second one.
• In append mode, an aggregation without a watermark emits nothing until state is bounded.
• foreachBatch(df, batchId) gives you a static DataFrame per micro-batch - the way to do MERGE INTO or write to a non-streaming sink.
• Kafka offsets are tracked by the checkpoint, not by you.

Worked example

(spark.readStream.schema(s).load(path)
  .withWatermark("ts", "10 minutes")
  .groupBy(F.window("ts", "5 minutes").alias("w"))
  .count()
  .writeStream.outputMode("append")
  .option("checkpointLocation", ck).toTable("counts"))