[
https://issues.apache.org/jira/browse/SPARK-59731?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18123854#comment-18123854
]
yoojung edited comment on SPARK-59731 at 10/6/26 3:33 PM:
----------------------------------------------------------
Hi [~loicd38], thanks again for the detailed report. I tried to reproduce this
on current {{master}} and want to share what I found before going further.
*Setup*
* Build: {{master}} at 43ccc6baa6e, Java 17, Python 3.12, py4j 0.10.9.9,
{{{}local[2]{}}}.
* Query: {{rate-micro-batch}} source, {{foreachBatch}} with a trivial callback
({{{}df.count(){}}}), {{{}trigger(processingTime="50 milliseconds"){}}}.
* Measurement: every 20s, {{{}jcmd <pid> GC.class_histogram{}}}, which forces
a full GC first, so the counts are the retained set.
*Results (about 1,000 micro-batches per run)*
||Run||{{classic.SparkSession}}||{{classic.Dataset}} / {{QueryExecution}}||
|master, default Python GC|2 (constant)|13-35, sawtooth, no upward trend|
|master, {{gc.disable()}} in the driver|2 (constant)|72 -> 573, +1 per batch|
# *No per-batch {{SparkSession}} retention.* The session count stays at 2 in
both runs, so I could not reproduce "a distinct SparkSession per micro-batch"
on Apache Spark. As you suspected, this may depend on something outside the OSS
code path.
# *The Py4J retention itself is real.* With Python's cyclic GC disabled, each
batch's {{{}Dataset{}}}, {{QueryExecution}} and accumulator metadata
({{{}AccumulatorMetadata{}}}) stays alive until the Python proxy is collected.
With the cyclic GC enabled it is bounded, because the GC reclaims the proxies
periodically. So the mechanism you described exists, but in my runs the default
GC keeps it from growing without bound.
*Possible hardening*
In {{{}ForeachBatchFunction.call{}}}, release the two Py4J references
({{{}jdf{}}} and {{{}jdf.sparkSession(){}}}) in a {{finally}} once the user
function returns, instead of waiting for the proxies to be collected. PySpark
already does this in {{mllib/common.py}} and {{ml/wrapper.py}} via
{{{}gateway.detach(...){}}}. In my runs this keeps the retained {{Dataset}}
count flat (1-13) even with {{{}gc.disable(){}}}.
One behavior change to be aware of: a batch DataFrame that a user stores
outside the function and uses after the batch has finished would raise
{{{}Py4JError{}}}.
*Questions*
* @Descotte: do you see the growth on plain Apache Spark 4.1.0, or only on a
vendor distribution? Does your driver disable or tune Python's GC (for example
{{{}gc.disable(){}}}, {{{}gc.freeze(){}}}, or large thresholds), or run many
other threads or libraries that could delay it?
* Committers: is this hardening worth a PR given that I could not reproduce
the unbounded growth? I am happy to open one if you think it is, and will
otherwise leave this ticket as is.
I'll hold off on opening a PR until I hear back.
was (Author: JIRAUSER313916):
Hi [~loicd38] , thanks for the detailed report.
I'd like to work on this issue. Nobody seems to be assigned, and I couldn't
find an existing PR for it.
My plan:
# Try to reproduce on a current {{master}} build, using a PySpark
{{foreachBatch}} query with a short processing-time trigger. I'll sample {{jcmd
GC.class_histogram}} periodically and watch the counts of {{{}Dataset{}}},
{{classic.SparkSession}} and {{AccumulatorMetadata}} instances.
# If the growth reproduces, I'll look at {{ForeachBatchFunction.call}} in
{{{}python/pyspark/sql/utils.py{}}}. The idea is to release the Py4J references
to the batch {{jdf}} and {{jdf.sparkSession()}} once the callback returns,
instead of relying on GC of the Python-side proxies.
# If it does not reproduce on OSS Spark, I'll post my measurements here rather
than open a PR.
> 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
> Priority: Major
>
> 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]
