[
https://issues.apache.org/jira/browse/SPARK-59731?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Descotte updated SPARK-59731:
-----------------------------
Description:
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 on Databricks 18 LTS
* 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
was:
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
> 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 on Databricks 18 LTS
> * 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]
