Descotte created SPARK-59731:
--------------------------------
Summary: foreachBatch in PySpark Structured Streaming leaks a full
SparkSession per micro-batch, causing driver OOM on long-running continuous
streams
Key: SPARK-59731
URL: https://issues.apache.org/jira/browse/SPARK-59731
Project: Spark
Issue Type: Bug
Components: PySpark, Structured Streaming
Affects Versions: 4.1.0
Reporter: Descotte
h3. Problem
On a long-running - about 10 hours - PySpark Structured Streaming query using
{{foreachBatch}} with
{{trigger(processingTime=...)}} (a standard periodic trigger, not
{{{}availableNow{}}}), the driver's live heap (post-GC retained set) grows
monotonically until it hits *java.lang.OutOfMemoryError: GC overhead limit*
{*}exceeded{*}. This happens even though the {{foreachBatch}} callback itself is
completely stateless (e.g. a single {{df.write.saveAsTable(..., mode="append")}}
call) and holds no external references.
Heap dump analysis on a crashed driver shows thousands of live
{{org.apache.spark.sql.classic.SparkSession}} instances (~3,700+ at crash time),
each fully retained along with its {{{}SessionState{}}}, session catalog, SQL
metrics/accumulators (~200 per session) and associated
{{ConcurrentHashMap$Node}} entries (~20,000+ per session). The number of
retained sessions tracks the number of micro-batches processed, not wall-clock
time: heavier-traffic streams (more micro-batches/hour) hit the OOM sooner than
lighter ones running on an identical driver size, which points to a per-batch
leak rather than a time-based or load-based effect.
We traced this to the {{ForeachBatchFunction}} boundary
({{{}pyspark/sql/utils.py{}}}, in the code path invoked by
{{{}ForeachBatchFunction.call{}}}): on every JVM→Python callback, two JVM
objects
are passed to Python — the batch {{Dataset}} and its
{{{}Dataset.sparkSession(){}}}. Under Py4J, objects passed across the JVM/Python
boundary are registered in {{{}Gateway.bindings{}}}, a {{ConcurrentHashMap}} of
*strong* references, released only when the corresponding Python-side proxy is
garbage collected — which does not reliably happen because {{JavaObject}} and
{{JavaMember}} form a reference cycle on the Python side (documented Py4J
behavior). As a result, the retained {{Dataset}} keeps its
{{{}QueryExecution{}}}/physical plan (hence its {{SQLMetric}} accumulators)
alive,
and the retained session drags in its full session state.
We were unable to determine, from the current Spark source alone, what
specifically causes a *distinct* {{SparkSession}} to be retained per
micro-batch rather than a single shared session being reused — we could not
find anywhere in {{MicroBatchExecution}} that calls {{cloneSession()}} per
batch (unlike the per-query clone in {{{}StreamExecution{}}}, which is fine).
{{SPARK-34087}} (a previously known session-clone leak) appears fixed since
3.2.0 and does not seem to match this pattern, but we cannot rule out
Databricks Runtime-specific glue for {{ForeachBatchSink}} being involved, since
we cannot inspect that code.
h3. Impact
On streams with a periodic trigger and enough throughput, the driver's live
heap grows without bound and the streaming query eventually crashes with an
OOM, restarts, and repeats the cycle indefinitely. Lower-traffic streams show
the same retention pattern but take proportionally longer (days instead of
hours) to hit the same ceiling — increasing driver memory only delays the
crash, it does not fix it.
h3. Repro sketch
{code:python}
def write_batch(batch_df, batch_id):
batch_df.write.saveAsTable("some_table", mode="append")
(spark.readStream
.format("...")
.load()
.writeStream
.foreachBatch(write_batch)
.trigger(processingTime="a few seconds")
.start())
{code}
Let this run for several hours while taking periodic driver heap
histograms/dumps. The count of live
{{org.apache.spark.sql.classic.SparkSession}} instances (and of
{{AccumulatorMetadata}} entries) grows roughly linearly with the number of
micro-batches processed, and never goes back down.
h3. Workaround we applied
For streams whose {{foreachBatch}} body is a plain append (no per-batch
business logic requiring Python), replacing:
{code:python}
.foreachBatch(write_batch).start()
{code}
with the native sink:
{code:python}
.toTable("some_table")
{code}
removes the leak entirely, because the native Delta/table sink runs fully
inside the JVM and never crosses the Py4J boundary per micro-batch. This is
only a workaround for the subset of pipelines that do not actually need
Python-side per-batch logic; queries that genuinely need {{foreachBatch}}
still have the underlying issue.
As a more generic (unverified/unmeasured at scale) mitigation for cases that
do need {{{}foreachBatch{}}}, explicitly detaching the Py4J binding at the end
of
the callback (e.g. {{{}df._sc._gateway.detach(df._jdf){}}}, or calling
{{{}gc.collect(){}}}) appears to release the retained objects, but this relies
on
private PySpark/Py4J internals and its per-batch overhead has not been
benchmarked.
h3. Environment
* Spark version: 4.1.0
* Structured Streaming, PySpark, {{foreachBatch}} sink, periodic
({{{}processingTime{}}}) trigger, continuous/long-running query
* Reproduced with a driver heap of several GB; leak rate scales with
micro-batch count, not elapsed time
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
