[SPARK-31354] SparkContext only register one SparkSession ApplicationEnd listener #28128

vinooganesh · 2020-04-05T14:41:29Z

What changes were proposed in this pull request?

This change was made as a result of the conversation on https://issues.apache.org/jira/browse/SPARK-31354 and is intended to continue work from that ticket here.

This change fixes a memory leak where SparkSession listeners are never cleared off of the SparkContext listener bus.

Before running this PR, the following code:

SparkSession.builder().master("local").getOrCreate()
SparkSession.clearActiveSession()
SparkSession.clearDefaultSession()

SparkSession.builder().master("local").getOrCreate()
SparkSession.clearActiveSession()
SparkSession.clearDefaultSession()

would result in a SparkContext with the following listeners on the listener bus:

[org.apache.spark.status.AppStatusListener@5f610071,
org.apache.spark.HeartbeatReceiver@d400c17,
org.apache.spark.sql.SparkSession$$anon$1@25849aeb, <-First instance
org.apache.spark.sql.SparkSession$$anon$1@fadb9a0] <- Second instance

After this PR, the execution of the same code above results in SparkContext with the following listeners on the listener bus:

[org.apache.spark.status.AppStatusListener@5f610071,
org.apache.spark.HeartbeatReceiver@d400c17,
org.apache.spark.sql.SparkSession$$anon$1@25849aeb] <-One instance

How was this patch tested?

Unit test included as a part of the PR

…RK-27958

…7958

vinooganesh · 2020-04-08T02:37:41Z

Hey @srowen - could I get an "ok to test" on here? also thoughts on this approach?

srowen · 2020-04-08T04:04:25Z

Jenkins test this please

SparkQA · 2020-04-08T05:45:16Z

Test build #120948 has finished for PR 28128 at commit 99c5f64.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

vinooganesh · 2020-04-30T01:38:46Z

hey @srowen - apologies for the repeated ping but I think you had some thoughts about this one. is this approach preferred compared to my previous one?

tagging some interested parties from the other ticket as well @cloud-fan @jiangxb1987 @rdblue @HyukjinKwon @britishbadger

cloud-fan · 2020-05-01T07:41:30Z

I'm not against adding a clear method to free a session, but I'd like to understand more about the memory leak.

According to #24807 (comment) , if the memory leak is caused by the listener, can we simply register that listener only once? That listener just null out the defaultSession at the end of application, and I don't see why we need to add one listener per spark session.

vinooganesh · 2020-05-01T16:30:14Z

Hi @cloud-fan - Good question! A few thoughts here:

I've seen people use the per session listener for various monitoring functions on a per session basis (ie. custom monitoring tools). So I think there is an expectation that there is a listener per session - which intuitively also makes sense since these are intended as short-lived, monitorable sessions.
The default session is a JVM wide singleton. Spinning up multiple spark sessions (multi-tenant inside a JVM) is also an SOP that firms have developed (see the comment here [SPARK-27958][SQL] Stopping a SparkSession should not always stop Spark Context #24807 (comment)). So nulling out the JVM wide default session once one particular active session is terminated removes the default session for every single other currently-existing session.

Does that make sense?

cloud-fan · 2020-05-05T09:03:29Z

per-session listeners is a common use case. Does the new method remove these user-registered per-session listeners? (registered by session.listenerManager)

AFAIK the reported issue is memory leak. If it's caused by listeners, I think there are 2 ways to fix it:

add an explicit API to remove listeners
make the listeners not reference the session instance.

Can you explain more about how we leak the spark session instance? Then we can discuss how to fix it.

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

vinooganesh · 2020-05-06T19:44:15Z

Hey @cloud-fan - Sure, right now the listener issue is coupled with the operating model for SparkSessions (which is where I think the confusion is coming from).

Currently, every time a SparkSession is created, a listener is attached to the SparkContext - https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala#L947.

Once the listener is created, because of the way it is created, the reference is lost after it is attached to the SparkContext. This means that there is currently no way to remove the listener from the SparkContext (even after the lifetime of the session is "done"). Even if I call clearActiveSession() or clearDefaultSession() the listener continues to live on the SparkContext, even after the sessions are GCed away. This create an issue for JVMs with multi-tenancy. Many SparkSessions may be spun up , but without a clean way to remove the listener, we leak it. This is the listener memory leak.

I think my PR description may have been unclear- there isn't a leak of the SparkSessions instances themselves (at least that I'm aware of).

The reason this PR does more than just removing the listeners, is because there isn't a lifecycle method that actually allows to mark the end of a SparkSession that doesn't kill the underlying SparkContext (stop() - which then kills all other sessions that rely on the SparkContext staying alive). The strangeness here is the interaction between SparkSessions, which should be lightweight and easy to clean upwith the longer-livedSparkContextwhich should exist for the duration of all aliveSparkSession`.

So, in order to ever clean up the listener leak, we need a way to mark the session as over, and that currently doesn't exist.

Does that make sense?

cloud-fan · 2020-05-07T04:52:18Z

If the only problem is https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala#L947

Can we just register it only once? The listener watches the SparkListenerApplicationEnd, which can happen only once for a Spark application. There is really no point in having many listeners do the same thing when the application ends.

vinooganesh · 2020-05-12T21:15:35Z

@cloud-fan - Okay, I've pushed a fix to immediately resolve this leak. The context keep track of whether the SparkSession SparkListenerApplicationEnd` listener has been registered. I've also renamed the PR to make sure it's clear that this PR doesn't correspond to the lifecycle spark session methods.

Does this approach make sense to you?

cloud-fan · 2020-05-13T14:31:13Z

core/src/main/scala/org/apache/spark/SparkContext.scala

  val startTime = System.currentTimeMillis()

  private[spark] val stopped: AtomicBoolean = new AtomicBoolean(false)
+  private[spark] val sessionListenerRegistered: AtomicBoolean = new AtomicBoolean(false)


can we move this to object SparkSession?

Not easily - that was my original plan, however a SparkSession object can be constructed with an instance of SparkContext. Meaning that it's really on the SparkContext to know whether a session has already attached a listener. The SparkSession instance can then decide whether or not to attach the listener, but the context will need to report whether the listener is attached first.

Sorry if I missed anything. object SparkSession is a singleton and it can make sure we do something only once. Are you saying that we can register a session cleanup listener to SparkContext outside of SparkSession?

Yeah so the challenge is that even though SparkSession is a singleton, I can create various spark session instances that are all "backed" by a different spark context instance.

val conf = new SparkConf() .setMaster("local") .setAppName("test-app-SPARK-31354-1") val context = new SparkContext(conf) val session1 = SparkSession.builder() .master("local") .sparkContext(context) // <- here .getOrCreate()

So, the state actually needs to moreso be on the context, since the context is the only one that knows whether it has already had a session listener attached. I agree that this feels really weird, and that's kind of why I was initially pursuing adding lifecycle methods to encapsulate the state in one place. Definitely open to other ways to resolve this though.

I think we can assume there is only one SparkContext instance. See #23311

One problem is we create a SparkContext, stop it and create it again. Then it's not safe to rely on the flag in singleton object SparkSession

If we really want to guide against the corner case, we can track the SparkContext instance id in object SparkSession. If SparkContext is re-created, we can reset the boolean flag.

It's a bit hacky to leak the session concept to the core module.

@cloud-fan - updated, how does this look?

cloud-fan · 2020-05-20T09:41:35Z

ok to test

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

SparkQA · 2020-05-20T15:17:34Z

Test build #122886 has finished for PR 28128 at commit e5563a7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vinooganesh · 2020-05-20T18:40:40Z

Thanks, @cloud-fan and @stczwd. Did some cleanup and added a test.

If anyone was curious, before this PR, running the following

SparkSession.builder().master("local").getOrCreate()
SparkSession.clearActiveSession()
SparkSession.clearDefaultSession()

SparkSession.builder().master("local").getOrCreate()
SparkSession.clearActiveSession()
SparkSession.clearDefaultSession()

would result in a SparkContext with the following listeners on the listener bus:

[org.apache.spark.status.AppStatusListener@5f610071,
org.apache.spark.HeartbeatReceiver@d400c17,
org.apache.spark.sql.SparkSession$$anon$1@25849aeb, <-First instance
org.apache.spark.sql.SparkSession$$anon$1@fadb9a0] <- Second instance

After this PR, the execution of the same code above results in SparkContext with the following listeners on the listener bus:

[org.apache.spark.status.AppStatusListener@5f610071,
org.apache.spark.HeartbeatReceiver@d400c17,
org.apache.spark.sql.SparkSession$$anon$1@25849aeb] <-One instance

SparkQA · 2020-05-20T23:09:33Z

Test build #122905 has finished for PR 28128 at commit e5ef33a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-05-21T03:38:37Z

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

+
+  /** Register the AppEnd listener onto the Context  */
+  private def registerContextListener(sparkContext: SparkContext): Unit = {
+    if (!SparkSession.listenerRegistered.get()) {


nit: we are in the same class so we can remove SparkSession.

cloud-fan · 2020-05-21T03:38:42Z

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

+          defaultSession.set(null)
+        }
+      })
+      SparkSession.listenerRegistered.set(true)


SparkQA · 2020-05-21T16:03:57Z

Test build #122929 has finished for PR 28128 at commit cfa1462.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-05-21T16:06:21Z

thanks, merging to master/3.0!

…End listener ## What changes were proposed in this pull request? This change was made as a result of the conversation on https://issues.apache.org/jira/browse/SPARK-31354 and is intended to continue work from that ticket here. This change fixes a memory leak where SparkSession listeners are never cleared off of the SparkContext listener bus. Before running this PR, the following code: ``` SparkSession.builder().master("local").getOrCreate() SparkSession.clearActiveSession() SparkSession.clearDefaultSession() SparkSession.builder().master("local").getOrCreate() SparkSession.clearActiveSession() SparkSession.clearDefaultSession() ``` would result in a SparkContext with the following listeners on the listener bus: ``` [org.apache.spark.status.AppStatusListener5f610071, org.apache.spark.HeartbeatReceiverd400c17, org.apache.spark.sql.SparkSession$$anon$125849aeb, <-First instance org.apache.spark.sql.SparkSession$$anon$1fadb9a0] <- Second instance ``` After this PR, the execution of the same code above results in SparkContext with the following listeners on the listener bus: ``` [org.apache.spark.status.AppStatusListener5f610071, org.apache.spark.HeartbeatReceiverd400c17, org.apache.spark.sql.SparkSession$$anon$125849aeb] <-One instance ``` ## How was this patch tested? * Unit test included as a part of the PR Closes #28128 from vinooganesh/vinooganesh/SPARK-27958. Lead-authored-by: Vinoo Ganesh <[email protected]> Co-authored-by: Vinoo Ganesh <[email protected]> Co-authored-by: Vinoo Ganesh <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit dae7988) Signed-off-by: Wenchen Fan <[email protected]>

advancedxy · 2020-07-02T16:20:32Z

Hi, @vinooganesh and @cloud-fan
I'd like to point out that this pr doesn't fix the memory leaky completely. Once SessionState is touched, it will add two more listeners into the SparkContext, namely SQLAppStatusListener and ExecutionListenerBus

It can be reproduced easily as

  test("SPARK-31354: SparkContext only register one SparkSession ApplicationEnd listener") {
    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("test-app-SPARK-31354-1")
    val context = new SparkContext(conf)
    SparkSession
      .builder()
      .sparkContext(context)
      .master("local")
      .getOrCreate()
      .sessionState // this touches the sessionState
    val postFirstCreation = context.listenerBus.listeners.size()
    SparkSession.clearActiveSession()
    SparkSession.clearDefaultSession()

    SparkSession
      .builder()
      .sparkContext(context)
      .master("local")
      .getOrCreate()
      .sessionState // this touches the sessionState
    val postSecondCreation = context.listenerBus.listeners.size()
    SparkSession.clearActiveSession()
    SparkSession.clearDefaultSession()
    assert(postFirstCreation == postSecondCreation)
  }

vinooganesh · 2020-07-02T21:08:01Z

@advancedxy - can you file a new jira ticket? I can look into fixing this.

advancedxy · 2020-07-03T03:32:42Z

@advancedxy - can you file a new jira ticket? I can look into fixing this.

@vinooganesh filed https://issues.apache.org/jira/browse/SPARK-32165

…End listener This change was made as a result of the conversation on https://issues.apache.org/jira/browse/SPARK-31354 and is intended to continue work from that ticket here. This change fixes a memory leak where SparkSession listeners are never cleared off of the SparkContext listener bus. Before running this PR, the following code: ``` SparkSession.builder().master("local").getOrCreate() SparkSession.clearActiveSession() SparkSession.clearDefaultSession() SparkSession.builder().master("local").getOrCreate() SparkSession.clearActiveSession() SparkSession.clearDefaultSession() ``` would result in a SparkContext with the following listeners on the listener bus: ``` [org.apache.spark.status.AppStatusListener5f610071, org.apache.spark.HeartbeatReceiverd400c17, org.apache.spark.sql.SparkSession$$anon$125849aeb, <-First instance org.apache.spark.sql.SparkSession$$anon$1fadb9a0] <- Second instance ``` After this PR, the execution of the same code above results in SparkContext with the following listeners on the listener bus: ``` [org.apache.spark.status.AppStatusListener5f610071, org.apache.spark.HeartbeatReceiverd400c17, org.apache.spark.sql.SparkSession$$anon$125849aeb] <-One instance ``` * Unit test included as a part of the PR Closes apache#28128 from vinooganesh/vinooganesh/SPARK-27958. Lead-authored-by: Vinoo Ganesh <[email protected]> Co-authored-by: Vinoo Ganesh <[email protected]> Co-authored-by: Vinoo Ganesh <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

Vinoo Ganesh and others added 13 commits June 5, 2019 11:58

add counter to monitor number of active SparkSessions

bcdd070

clear session always in stop()

f333a30

added tests

8fc95e9

Merge branch 'master' of github.com:apache/spark into vinooganesh/SPA…

ff60b84

…RK-27958

responding to PR comments

f69cabb

cleanup

3fafd2a

adding ticket number to test

0c9c426

style

843491f

addressing sean's PR and updating test

92c7b22

Merge branch 'master' into vinooganesh/SPARK-27958

d4e4e27

first iteration of new proposal

61c6fed

Merge remote-tracking branch 'origin/master' into vinooganesh/SPARK-2…

b586bf9

…7958

testing and style

387acb1

vinooganesh mentioned this pull request Apr 5, 2020

[SPARK-27958][SQL] Stopping a SparkSession should not always stop Spark Context #24807

Closed

style fix

99c5f64

Merge remote-tracking branch 'origin' into vinooganesh/SPARK-27958

7613046

probot-autolabeler bot added the SQL label Apr 30, 2020

kiszk reviewed May 5, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala Outdated Show resolved Hide resolved

Merge branch 'master' into vinooganesh/SPARK-27958

7a70369

vinooganesh added 2 commits May 6, 2020 15:44

remove unnecessary s

5a1e0ab

Merge branch 'master' into vinooganesh/SPARK-27958

38dea00

vinooganesh added 2 commits May 12, 2020 17:12

remove unnecessary import

0284c79

Merge branch 'master' into vinooganesh/SPARK-27958

19f45da

add ticket number to test

9573805

cloud-fan reviewed May 13, 2020

View reviewed changes

vinooganesh added 2 commits May 18, 2020 18:43

Merge branch 'master' into vinooganesh/SPARK-27958

e877ee1

move logic to listener

e5563a7

cloud-fan reviewed May 20, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala Outdated Show resolved Hide resolved

cloud-fan reviewed May 20, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala Outdated Show resolved Hide resolved

cloud-fan approved these changes May 20, 2020

View reviewed changes

adding test and cleanup

e5ef33a

cloud-fan reviewed May 21, 2020

View reviewed changes

cloud-fan approved these changes May 21, 2020

View reviewed changes

same class, don't need SparkSession

cfa1462

cloud-fan closed this in dae7988 May 21, 2020

Alexis-D mentioned this pull request Dec 9, 2020

[SPARK-31354] SparkContext only register one SparkSession ApplicationEnd listener palantir/spark#721

Closed

Ngone51 mentioned this pull request Jan 7, 2021

[SPARK-32165][SQL] Ensure Spark only initiates SharedState once across SparkSessions #31053

Closed

vinooganesh mentioned this pull request Jan 17, 2022

[SPARK-32165][SQL] Ensure Spark only initiates SharedState once across SparkSessions #35224

Closed

[SPARK-31354] SparkContext only register one SparkSession ApplicationEnd listener #28128

[SPARK-31354] SparkContext only register one SparkSession ApplicationEnd listener #28128

Uh oh!

Conversation

vinooganesh commented Apr 5, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

vinooganesh commented Apr 8, 2020

Uh oh!

srowen commented Apr 8, 2020

Uh oh!

SparkQA commented Apr 8, 2020

Uh oh!

vinooganesh commented Apr 30, 2020

Uh oh!

cloud-fan commented May 1, 2020

Uh oh!

vinooganesh commented May 1, 2020

Uh oh!

cloud-fan commented May 5, 2020

Uh oh!

Uh oh!

vinooganesh commented May 6, 2020

Uh oh!

cloud-fan commented May 7, 2020

Uh oh!

vinooganesh commented May 12, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan May 13, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan May 13, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented May 20, 2020

Uh oh!

Uh oh!

Uh oh!

SparkQA commented May 20, 2020

Uh oh!

vinooganesh commented May 20, 2020

Uh oh!

SparkQA commented May 20, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented May 21, 2020

Uh oh!

cloud-fan commented May 21, 2020

Uh oh!

advancedxy commented Jul 2, 2020

Uh oh!

vinooganesh commented Jul 2, 2020

Uh oh!

advancedxy commented Jul 3, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

vinooganesh commented Apr 5, 2020 •

edited

Loading

cloud-fan May 13, 2020 •

edited

Loading

cloud-fan May 13, 2020 •

edited

Loading