[
https://issues.apache.org/jira/browse/SPARK-59731?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18125708#comment-18125708
]
Craig Ingram commented on SPARK-59731:
--------------------------------------
We see the same per-micro-batch SparkSession retention on DBR 18 LTS (Spark
4.1.0) with a *Scala* foreachBatch job, so Py4J is not required to trigger it.
*Setup:* Structured Streaming, processingTime(5 minutes), foreachBatch running
Delta MERGEs into small liquid-clustered tables; 3 queries in one JVM; driver
m6g.xlarge (7.85 GB heap).
*Traffic is very low:* each MERGE has about 4 source rows and the tables are
about 330 KiB in a handful of files. The leak follows the number of
micro-batches that write, not the data volume: with a 5-minute trigger there
are about 29 MERGE commits per hour into the main table (plus about 10 per hour
into a second table from the hourly query), and about one leaked execution per
auto {{OPTIMIZE}}, which fires after roughly 8 small files accumulate (every 8
to 9 commits on the main table). We have not varied the input rate, so we
cannot say how it behaves at much lower or higher volume.
*Measurements (live objects only, 3 heap dumps of one driver, at 0.75 h / 6 h /
16 h of uptime):*
* retained {{SparkSession}}: 30 / 94 / 199
* retained {{ActiveJob}}: 5 / 42 / 106
* entries in {{DAGScheduler.activeQueryToJobs}}: 3 / 22 / 54
* live heap grows about 35 MB per hour; a smaller driver fails sooner.
*Retention path (path to GC roots, weak/soft references excluded):*
{{DAGScheduler.activeQueryToJobs}} -> {{ActiveJob}} -> {{JobWaiter}} -> collect
plan -> {{SparkSession}}.
In the 0.75 h and 6 h dumps {{activeJobs}}, {{jobIdToActiveJob}} and
{{runningStages}} are all empty, and in the 6 h dump every retained job has
numFinished == numPartitions, so these are finished jobs that were never
removed from the map.
*Which executions:* 52 of the 54 keys in the 16 h dump hold exactly two
finished jobs (a map-stage job and a result job, one adaptive collect); the
other two hold one. Over a 6 hour window, 19 executions had jobs whose
submissionTime was 1.2 to 4.4 s _after_ the execution's own end time (from the
REST {{/sql}} and {{/jobs}} endpoints), and all 19 ids are keys of
{{activeQueryToJobs}} in a later dump, with no other key in that id range. In
the table's Delta log, each of those executions is followed 4 to 8 s later by
an {{OPTIMIZE}} commit with {{operationParameters.auto = true}}; only about 28%
of all MERGE commits are followed by one within 30 s. The leaked plans look
like Delta file listings (their columns match Delta's AddFile record: path,
partitionValues, size, modificationTime).
*Controls:* the single-query variant of this job (same code path, older build)
ran 206 hours on DBR 17.3 (Spark 4.0) in another environment with 10 retained
sessions and no retained ActiveJob objects. Other streaming jobs on DBR 18
(partitioned tables) showed no ActiveJob growth in a single heap histogram at
about 15 h.
*Where this sits in Apache Spark:* {{activeQueryToJobs}} was added by
SPARK-55064 (fix version 4.2.0); it is not in the v4.1.0 or v4.1.1 sources, so
this runtime includes it by some other route (the dump shows a Scala
{{HashMap}} where master uses a {{ConcurrentHashMap}}, so it may be a separate
implementation). In master, entries are added unconditionally in
{{handleJobSubmitted}} and {{handleMapStageSubmitted}} under the job's
{{spark.sql.execution.id}}, and the only removal is the {{CleanupQueryJobs}}
event posted when {{SQLExecution.withNewExecutionId}} finishes. There is no
check at the add site, so a job submitted with that execution id after the
cleanup event is added and never removed, which pins its {{ActiveJob}}, plan
and session. On our runtime we observe exactly that sequence (above). We have
not identified which component submits those jobs, and we have not tried to
reproduce on a stock Apache Spark build.
*A check for the reporter:* if your table gets {{OPTIMIZE}} commits with {{auto
= true}} (DESCRIBE HISTORY) right after the batches that leak, it may be the
same cause. We have not tested whether the native table sink (your workaround)
avoids it for that reason.
> 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]
