Structured streaming + Kafka architecture — offsets, checkpoint, exactly-once

The execution model

A structured streaming query on Kafka runs as a sequence of micro-batches. For each batch the driver asks Kafka for the latest offsets, decides an offset range per partition -- from where the last batch ended to some new end -- writes that plan durably, and only then launches tasks that read exactly those ranges. Each batch is therefore a bounded, deterministic, replayable unit of work: given the same offset range, re-running it produces the same rows.

That determinism is the foundation of everything else. Fault tolerance is 'replay the batch whose plan we wrote but never committed'. Exactly-once is 'make the sink ignore a replayed batch it already applied'. Rate limiting is 'choose a smaller end offset'. None of it would work if the batch boundary were decided by the consumer as it read.

The consequence for latency is that end-to-end delay has a floor of roughly one batch duration plus planning overhead -- typically hundreds of milliseconds to seconds, not microseconds. Spark's continuous processing mode exists to attack that floor and has remained experimental with a restricted operator set; for the overwhelming majority of pipelines the micro-batch engine is the right answer and sub-second latency is not the requirement people assume it is.

Reading from Kafka

The source is configured with options, and a handful of them matter more than the rest.

df = (spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
      .option("subscribe", "orders,payments")
      .option("startingOffsets", "latest")
      .option("maxOffsetsPerTrigger", 500000)
      .option("failOnDataLoss", "true")
      .load())

Topic selection comes in three forms: subscribe with an explicit list, subscribePattern with a regular expression that picks up newly created topics, and assign with specific partitions as JSON when you want manual control. Anything prefixed kafka. is passed through to the underlying consumer, which is how security settings -- SASL mechanism, JAAS config, truststore -- are supplied.

startingOffsets accepts earliest, latest, or a JSON map of topic to partition to offset, and it applies only when the query starts with no checkpoint. On every subsequent restart the checkpoint wins and the option is ignored, which is the source of a recurring confusion: changing startingOffsets on a running pipeline does nothing at all.

The returned DataFrame has a fixed schema -- key, value, topic, partition, offset, timestamp, timestampType -- with key and value as binary. Deserialisation is your job: cast to string for text, from_json with an explicit schema for JSON, or the Avro functions with a schema registry. Keep the metadata columns rather than projecting them away immediately; partition and offset are the only things that make a bad record traceable back to its origin.

Where offsets actually live

Spark does not commit offsets to Kafka. It generates a consumer group identifier for internal use, disables automatic commits, and records progress in the checkpoint directory. Kafka is treated purely as a replayable log addressed by offset.

The operational consequence catches every team once: the standard consumer lag tooling shows nothing, because no group is committing. Monitoring has to come from Spark's own progress metrics, or from comparing the offsets Spark reports against the topic's latest offsets, or -- in recent versions -- by supplying a group identifier and enabling optional offset commit-back purely so external tools can observe it. That last option is for observability only; Spark still does not read those committed offsets on restart.

The checkpoint directory has a specific structure, and knowing it turns several mysteries into simple checks. offsets/ holds one file per batch recording the planned offset range -- written before the batch executes, which is what makes replay possible. commits/ holds a marker per successfully completed batch. sources/ and state/ hold source metadata and any stateful operator's data. On restart Spark reads the last entry in offsets/; if there is no matching entry in commits/, that batch is re-executed with the identical offset range.

So 'where is my job up to' is answerable by reading the newest file in offsets/, and 'why did it reprocess' is usually answerable by seeing an offsets entry with no commits entry.

What a checkpoint will not let you change

A checkpoint ties a query's progress to that query's structure. Some changes are safe across a restart and some are not, and the unsafe ones either fail loudly at startup or -- worse -- produce wrong results.

Generally safe: changing the code inside a map or a foreachBatch, adjusting maxOffsetsPerTrigger, changing the trigger interval, altering non-structural Kafka consumer settings, adding a topic to a subscription in some versions.

Generally unsafe: changing the output mode, changing the keys or aggregation structure of a stateful operator, adding or removing a stateful operator, changing the schema of state, or switching the source type. These change the meaning of the state and the checkpoint has no way to migrate it.

The practical protocol is to treat a structural change as a new query: new checkpoint directory, deliberate startingOffsets, and a decision about the gap. Either start from the offsets the old query reached -- read them out of the old checkpoint and pass them as JSON, which is the clean cutover -- or start from a timestamp, or accept reprocessing from earliest if the sink is idempotent. Running old and new side by side into different sinks, then switching readers, is the blue-green version and is worth the effort for anything critical.

The corollary is that checkpoint directories are production state, not scratch. Losing one means either replaying a topic from the beginning or accepting a gap. Put them somewhere durable, back them up if the topic's retention is shorter than your recovery expectations, and never point two queries at the same one.

Delivery semantics, end to end

Exactly-once is a property of three things together. The source must be replayable by offset -- Kafka is. The engine must record what it planned before it acts -- the checkpoint does. And the sink must be idempotent or transactional with respect to a replayed batch. Two out of three gives at-least-once, which means duplicates after any failure.

Sinks divide accordingly. The file sink maintains a manifest of committed files and ignores output from a replayed batch, so it is exactly-once. Table formats such as Delta, Iceberg and Hudi record the batch identifier in their transaction log and skip a batch they have already applied -- also exactly-once, and the reason they are the default sink for lakehouse pipelines. The Kafka sink is at-least-once: Spark's writer does not use Kafka transactions, so a replayed batch republishes its records and downstream consumers must deduplicate, typically on a business key or on a producer-supplied identifier.

For anything else, foreachBatch is the escape hatch and the place where you own the guarantee. It hands you a normal DataFrame and the batch identifier, and the identifier is what makes idempotence implementable:

def upsert(batch_df, batch_id):
    (batch_df.write
       .mode("overwrite")
       .option("txnAppId", "orders_stream")
       .option("txnVersion", batch_id)      # sink skips a replayed batch
       .save(target))

query = df.writeStream.foreachBatch(upsert).option(
    "checkpointLocation", "s3://bucket/checkpoints/orders_v3").start()

Where the sink offers no such mechanism -- an arbitrary relational database, an HTTP endpoint -- the two workable patterns are a merge keyed on a unique business identifier, or a small table recording the last applied batch identifier written in the same transaction as the data. Anything else is at-least-once, and the honest thing is to say so in the pipeline's documentation rather than to assume the engine covers it.

Rate limiting and triggers

The first batch of a new query with startingOffsets=earliest will try to read the entire topic in one batch. On a topic with weeks of retention that is a job that runs for hours, spills, and often dies -- and the failure looks like a Spark problem rather than a configuration one.

maxOffsetsPerTrigger caps the number of records per batch, distributed proportionally across partitions. Setting it is close to mandatory: it bounds batch size, keeps batch duration stable, and turns a catch-up from a cliff into a ramp. Recent versions add a lower bound with a maximum delay, so that a quiet stream waits briefly to accumulate a worthwhile batch instead of producing many tiny ones -- which matters when the sink is a file or table format, because tiny batches make small files.

Triggers control cadence. The default fires a new batch as soon as the previous one finishes, which minimises latency and can produce very small batches. A fixed processing time gives predictable batch sizes and predictable output file sizes. AvailableNow processes everything currently in the topic and then stops -- crucially, in multiple batches that respect the rate limit, which is what makes it the right choice for scheduled incremental jobs and the replacement for the older once-only trigger that ignored the limit and tried to do everything in one go.

The AvailableNow pattern deserves emphasis because it is underused: a streaming query run on a schedule, keeping a checkpoint between runs, gives you incremental processing with exactly-once semantics and no always-on cluster. For pipelines that do not need continuous latency, it is cheaper and simpler than either a real stream or a hand-rolled batch job tracking its own watermark.

Parallelism and partitions

By default Spark creates one task per Kafka partition per batch. That is a clean mapping and it means the topic's partition count is a hard ceiling on read parallelism: a four-partition topic uses four cores no matter how large the cluster, which is the answer to the recurring 'why is my streaming job not using the executors I gave it'.

The minPartitions option relaxes it by splitting a partition's offset range across several tasks, so a four-partition topic can be read by sixteen. It costs more consumer connections and it does not help if the bottleneck is downstream, but for a catch-up backlog on a narrow topic it is the difference between minutes and hours. The durable fix is to partition the topic adequately in the first place, sized for the consumers rather than the producers.

Executors cache Kafka consumers between batches, since creating one per batch would dominate the cost of a short batch. The cache is sized by configuration and the relevant symptom of it being too small is a rise in consumer creation and connection churn on the brokers.

Downstream parallelism is a separate question. A shuffle after the read repartitions by whatever the query needs, and the standard advice applies: avoid a shuffle partition count wildly larger than the data justifies, because in a streaming job that overhead is paid every batch, several times a minute, forever. Tuning shuffle partitions matters far more in streaming than in batch for exactly this reason.

Schema handling and bad records

Kafka carries bytes, so the schema contract lives in your code or in a registry, and streaming makes schema drift a runtime problem rather than a deploy-time one.

With JSON, from_json takes an explicit schema and returns null for fields it cannot parse -- silently. A malformed record yields a row of nulls that flows into the sink unless something checks. Keeping the raw value column alongside the parsed struct, and routing rows whose parse failed to a separate dead-letter sink, is the pattern that turns silent corruption into a visible, replayable queue.

With Avro and a schema registry, the writer schema travels with the message and evolution is governed by the registry's compatibility rules, which is a meaningfully stronger position. The Spark integration reads the schema identifier from the payload and resolves it; the operational requirement is that the streaming job can reach the registry, which becomes a runtime dependency of the pipeline and needs to be in the availability model.

Either way, decide explicitly what happens when a producer adds a field, removes one, or changes a type. The answers -- ignore unknown fields, fail the batch, dead-letter the record -- are all defensible, and the failure mode is having no answer and discovering the behaviour during an incident.

Failure modes worth knowing before they happen

Offsets aged out of retention. If the job is down longer than the topic's retention, the offsets in the checkpoint no longer exist. Spark fails the query because failOnDataLoss defaults to true, and the temptation is to set it false so the job restarts. Understand what that does: it silently skips to the earliest available offset and the missing records are gone with no record of how many. Treat the failure as the correct behaviour, decide deliberately, and if you do disable the check, log the gap.

New partitions. Partitions added to a topic are picked up on a later batch and read from their beginning, which for a partition created during the gap is the right default. Reducing partitions is not supported by Kafka and topic recreation with the same name is a genuine hazard -- offsets reset and Spark sees them as going backwards.

Checkpoints on object storage. A checkpoint is a directory of small files written and renamed every batch. On object storage, renames are copies and listings are eventually complete, which makes this a well-known source of slow batches and, historically, of corruption. Use a filesystem with real rename semantics where possible, or the storage-specific committer and log store the platform provides. Never assume a plain bucket path is fine because it worked in testing at low volume.

Two queries, one checkpoint. Guaranteed corruption, and easily done by copying a job definition. Include the query name in the checkpoint path by convention.

Silent stalls. A query can be running with zero input rows because of a broker authorisation change, a topic deletion, or a paused producer. Nothing fails; throughput is simply zero. Only monitoring catches this.

Monitoring what matters

Every query emits a progress report per batch, and it contains the numbers worth alerting on: input rows per second and processed rows per second, batch duration, the duration breakdown by phase, and the source's start and end offsets.

The single most important derived metric is lag: the latest offset available in Kafka minus the offset Spark has processed, per partition. Because the usual consumer-group tooling cannot see this job, compute it yourself -- a listener that records the reported end offsets and a periodic query of the topic's latest offsets is enough -- and alert on it growing rather than on any absolute value.

The second is the ratio of processed to input rate. Sustained processing below input means the job is falling behind and lag will grow without bound; they should be approximately equal in steady state, with processing higher during catch-up.

Attach a StreamingQueryListener to export these into whatever metrics system you run, and alert on three things: lag increasing over a sustained window, batch duration approaching the trigger interval, and zero input rows for longer than the quietest expected period. Those three catch nearly every real incident, and none of them is visible in a job-succeeded check -- a streaming job that is hopelessly behind is still a running job.

Operational recipes

Reset to a point in time. Stop the query, point it at a new checkpoint directory, and start with offsets by timestamp so the replay begins where you intend. Editing the old checkpoint by hand is not a supported operation.

Upgrade a stateful query. Treat it as a new query. Start the new version on a new checkpoint from the offsets the old one reached, let both run into separate targets briefly if you need to compare, then cut readers over.

Backfill and live in one pipeline. Run an AvailableNow query over the historical range into the same idempotent sink, then start the continuous query from the point the backfill ended. Idempotent sinks make the overlap harmless, which is another argument for a table format.

Right-size the batch. Aim for a batch duration comfortably below the trigger interval -- roughly half is a reasonable target -- so that a slow batch does not cascade. Tune with the rate limit rather than by adding executors first; a smaller, predictable batch beats a large one that occasionally overruns.

Keep the checkpoint and the code versioned together. Name checkpoint directories with a version suffix that you bump whenever the query structure changes. It makes the unsafe-change rule mechanical instead of a judgement call under pressure.