A restarted streaming job that writes duplicate rows and a grouped pandas job that runs a worker out of memory can appear in the same incident channel within an hour. They look related, but they come from two different operational contracts. The first is about what happens to an external write when a micro-batch is attempted again. The second is about how Spark gathers rows for each group and hands them to Python. Treating them separately is the fastest way to find the fix.
Contract one: what foreachBatch guarantees, and what it does not
foreachBatch lets you run your own logic on the output of every micro-batch. The callback receives a DataFrame and a unique batch ID. The Spark 3.5.8 Structured Streaming programming guide documents the default write guarantee for this pattern as at-least-once. In plain terms, a batch can be attempted more than once, so the side effects you perform inside the callback have to survive a second attempt.
As an Amazon Associate I earn from qualifying purchases.
The callback is your code, not Spark’s. Spark does not wrap an arbitrary database insert, API call, or file write in a transaction for you. The guide says the batch ID can be used to deduplicate output, which is how an external system ends up with exactly-once effects. That only holds when the destination enforces the check. Adding a batch ID parameter to a function does not make an arbitrary write transactional.
The Tool Desk
Outbyte Driver Updater FREEFix the driver behind crashes, sound loss and screen glitchesFind Drivers →Outbyte PC Repair FREEClear out junk files and repair common Windows errorsFree Scan →Checkpointing protects the streaming query’s progress. It does not, by itself, protect the rows your callback already pushed to a destination. Keep those two ideas apart when you read a restart log.
Making the callback safe to repeat
There are two patterns the guide points toward. Pick one per sink and verify it against the sink’s real behavior.
- Idempotent writes. Write using a natural key, such as an upsert or merge on
order_id. Applying the same batch twice then leaves the destination in the same final state. - Deduplication keyed on the batch ID. Record the batch ID in the destination in the same transaction as the data, and skip the batch when that ID is already present. This needs a destination that supports transactional writes. A plain append to object storage or a table without transaction support cannot provide this guarantee on its own.
A minimal shape for the callback looks like this. It is illustrative, not a complete implementation, and the merge step depends on your sink:
from pyspark.sql import DataFrame
def process_batch(batch_df: DataFrame, batch_id: int) -> None:
# batch_id is unique per micro-batch; use it for deduplication
# where the destination supports keyed or transactional writes.
deduped = batch_df.dropDuplicates(["order_id"])
merge_into_destination(deduped, key="order_id", batch_id=batch_id) # your sink-specific merge
query = stream_df.writeStream.foreachBatch(process_batch).option("checkpointLocation", ckpt).start()The guide also notes that foreachBatch relies on micro-batch execution. It is not available in continuous processing mode, so a continuous query cannot use this contract.
Restarts after an upgrade: the Spark 4.2 metadata case
The Spark 4.2 Structured Streaming migration guide describes a specific restart change. When the checkpoint metadata file is missing but the offset or commit logs contain data, the query now fails with a checkpoint metadata error. Earlier behavior could silently generate a new query ID. The guide identifies the risk of that silent path as duplicate data in exactly-once sinks, which is why the failure was made explicit.
The guide recommends one of two responses: restore the metadata file, or start with a new checkpoint location. A new checkpoint location begins the query’s progress from a fresh state, so decide first which source data must be reprocessed and whether the sink can absorb it. This is a documented scenario, not a complete account of every restart failure, so match it against your own error message before applying it.
Contract two: applyInPandas moves whole groups into Python
The PySpark 4.2.0 GroupedData.applyInPandas reference states that the grouped operation performs a full shuffle. In the regular pandas DataFrame form, all rows for one group are loaded into memory as a single pandas DataFrame before your function runs. The page explicitly names out-of-memory risk when a skewed group does not fit in memory.
Rank #3
Skew is what turns this into an incident. Total data volume can look comfortable while one key holds a disproportionate share of rows. For example, a job that processes 50 million rows across 10,000 keys looks modest on average, yet a single customer with 20 million rows forces one Python worker to hold all 20 million of them at once. The failure appears on one or a few workers, not across the cluster.
Measure group sizes before changing code
Find the largest groups first. This query shows the top keys by row count:
from pyspark.sql import functions as F
(df.groupBy("customer_id")
.count()
.orderBy(F.desc("count"))
.show(20, truncate=False))If the top few keys dwarf the rest, the grouped function is the risk. If group sizes are even, look elsewhere, such as an oversized input file or an unrelated executor setting.
The iterator form and its limits
The documented iterator form accepts an iterator of pandas DataFrames and yields DataFrames. It can reduce the need to hold a whole group in memory at once because your function can process the group in chunks. The Spark 4.2.0 reference page says iterator support was added in Spark 4.1.0, so confirm your deployed runtime and the exact function signature before proposing it as a replacement.
The iterator form does not remove the full shuffle. Memory needs still depend on the chunk size and on your algorithm. A computation that needs every row of a group before it can emit any output, such as a global sort or a whole-group statistic, gains little from chunking. Check that the logic can be expressed incrementally before relying on it.
Outdated Drivers Are Slowing You Down
One free scan finds every outdated or missing driver and matches the right update for your exact hardware.Free scan · exact hardware matchPC Slower Than It Used to Be?
A free scan shows the junk files, broken settings and background clutter dragging Windows down - then fixes them in one click.Free scan · Windows 10 & 11Choosing a mitigation without breaking results
Mitigations that change how keys are split or spread can alter the output of a computation. A transformation that is not decomposable across sub-groups can return different answers if a hot key is split. Verify the correctness requirement of the function before changing how its input is partitioned. Filtering unneeded rows and columns before the grouped step is generally the lowest-risk reduction, because it leaves the group logic untouched.
Independent reader supportYour contribution helps us test, update, and keep practical guides available for everyone.Triage on the day: which contract is failing
- Duplicate external records after a restart or retry. This is contract one. Check whether the callback’s writes are idempotent or deduplicated by batch ID for that specific sink.
- Python worker out-of-memory errors inside a grouped pandas function. This is contract two. Run the group-size query above, then decide between reducing input, restructuring the function, or using the iterator form on a supported runtime.
- Query refuses to restart after an upgrade, citing checkpoint metadata. This is the Spark 4.2 migration case. Restore the metadata file or use a new checkpoint location, and confirm which data will be reprocessed.
| Symptom | Contract to inspect | Documented direction |
|---|---|---|
| Duplicate external rows after restart or retry | Default at-least-once write guarantee of foreachBatch (Spark 3.5.8 guide) |
Make the write idempotent, or deduplicate on batch ID where the destination supports it; test the sink’s real behavior |
| Worker out-of-memory in grouped pandas code | Full shuffle and full-group-in-memory behavior of the DataFrame form of applyInPandas |
Inspect skewed keys; consider the iterator form on Spark 4.1.0 or later, and validate memory use of the function |
| Restart fails with checkpoint metadata error | Missing metadata file alongside offset or commit logs (Spark 4.2 migration guide) | Restore the metadata file, or start with a new checkpoint location after deciding what to reprocess |
Version and scope notes
The foreachBatch delivery contract cited here comes from the Spark 3.5.8 programming guide. If your cluster runs a different Spark release, confirm the behavior against that version’s documentation and against the actual sink. The applyInPandas iterator form is documented from Spark 4.1.0 onward. The checkpoint behavior described above is documented in the Spark 4.2 migration guide and applies to that migration case, not to every restart path.
No published benchmark or failure-rate statistic supports a claim about how often either problem occurs, so the guidance here is about mechanism and diagnosis rather than prevalence or relative performance.
Quick Recap
Product prices and availability are accurate as of the date/time indicated and are subject to change. Any price and availability information displayed on Amazon at the time of purchase will apply.
Quick wins for a faster PC:
Repair Windows errors before they cause bigger problemsFix Now →Scan for outdated or missing drivers - takes under a minuteDriver Scan →Clear out junk files and repair common Windows errorsFree Scan →
