[
https://issues.apache.org/jira/browse/SPARK-59731?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18123854#comment-18123854
]
yoojung commented on SPARK-59731:
---------------------------------
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]
