[SPARK-49249][SPARK-49122] Artifact isolation in Spark Classic #48120
There are no files selected for viewing
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| import org.apache.spark.sql.api.java.UDF2 | ||
|
|
||
| class IntSumUdf extends UDF2[Long, Long, Long] { | ||
| override def call(t1: Long, t2: Long): Long = t1 + t2 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -115,93 +115,95 @@ object SQLExecution extends Logging { | |
| } | ||
| val redactedConfigs = sparkSession.sessionState.conf.redactOptions(modifiedConfigs) | ||
|
|
||
| JobArtifactSet.withActiveJobArtifactState(sparkSession.artifactManager.state) { | ||
|
There was a problem hiding this comment. Can you check how this interacts with all the stuff we do in Connect to make this work? I feel that we are duplicating code now. cc @vicennial There was a problem hiding this comment. An FYI to other reviewers: look at this file with hidden whitespace. There was a problem hiding this comment. Hmm, with this in the execution code path, we may not need SessionHolder#withSession in a few places and can be cleaned up. There was a problem hiding this comment. @vicennial Is there a end-to-end test for this? I did some modifications and want to know if it won't break anything. There was a problem hiding this comment. @xupefei The ReplE2ESuite has some tests for the overall client->artifact->execution with artifact flow. |
||
| withSQLConfPropagated(sparkSession) { | ||
| var ex: Option[Throwable] = None | ||
| var isExecutedPlanAvailable = false | ||
| val startTime = System.nanoTime() | ||
| val startEvent = SparkListenerSQLExecutionStart( | ||
| executionId = executionId, | ||
| rootExecutionId = Some(rootExecutionId), | ||
| description = desc, | ||
| details = callSite.longForm, | ||
| physicalPlanDescription = "", | ||
| sparkPlanInfo = SparkPlanInfo.EMPTY, | ||
| time = System.currentTimeMillis(), | ||
| modifiedConfigs = redactedConfigs, | ||
| jobTags = sc.getJobTags(), | ||
| jobGroupId = Option(sc.getLocalProperty(SparkContext.SPARK_JOB_GROUP_ID)) | ||
| ) | ||
| try { | ||
| body match { | ||
| case Left(e) => | ||
| sc.listenerBus.post(startEvent) | ||
| throw e | ||
| case Right(f) => | ||
| val planDescriptionMode = | ||
| ExplainMode.fromString(sparkSession.sessionState.conf.uiExplainMode) | ||
| val planDesc = queryExecution.explainString(planDescriptionMode) | ||
| val planInfo = try { | ||
| SparkPlanInfo.fromSparkPlan(queryExecution.executedPlan) | ||
| } catch { | ||
| case NonFatal(e) => | ||
| logDebug("Failed to generate SparkPlanInfo", e) | ||
| // If the queryExecution already failed before this, we are not able to generate | ||
| // the the plan info, so we use and empty graphviz node to make the UI happy | ||
| SparkPlanInfo.EMPTY | ||
| } | ||
| sc.listenerBus.post( | ||
| startEvent.copy(physicalPlanDescription = planDesc, sparkPlanInfo = planInfo)) | ||
| isExecutedPlanAvailable = true | ||
| f() | ||
| } | ||
| } catch { | ||
| case e: Throwable => | ||
| ex = Some(e) | ||
| throw e | ||
| } finally { | ||
| val endTime = System.nanoTime() | ||
| val errorMessage = ex.map { | ||
| case e: SparkThrowable => | ||
| SparkThrowableHelper.getMessage(e, ErrorMessageFormat.PRETTY) | ||
| case e => | ||
| Utils.exceptionString(e) | ||
| } | ||
| if (queryExecution.shuffleCleanupMode != DoNotCleanup | ||
| && isExecutedPlanAvailable) { | ||
| val shuffleIds = queryExecution.executedPlan match { | ||
| case ae: AdaptiveSparkPlanExec => | ||
| ae.context.shuffleIds.asScala.keys | ||
| case _ => | ||
| Iterable.empty | ||
| } | ||
| shuffleIds.foreach { shuffleId => | ||
| queryExecution.shuffleCleanupMode match { | ||
| case RemoveShuffleFiles => | ||
| // Same as what we do in ContextCleaner.doCleanupShuffle, but do not unregister | ||
| // the shuffle on MapOutputTracker, so that stage retries would be triggered. | ||
| // Set blocking to Utils.isTesting to deflake unit tests. | ||
| sc.shuffleDriverComponents.removeShuffle(shuffleId, Utils.isTesting) | ||
| case SkipMigration => | ||
| SparkEnv.get.blockManager.migratableResolver.addShuffleToSkip(shuffleId) | ||
| case _ => // this should not happen | ||
| } | ||
| } | ||
| } | ||
| val event = SparkListenerSQLExecutionEnd( | ||
| executionId, | ||
| System.currentTimeMillis(), | ||
| // Use empty string to indicate no error, as None may mean events generated by old | ||
| // versions of Spark. | ||
| errorMessage.orElse(Some(""))) | ||
| // Currently only `Dataset.withAction` and `DataFrameWriter.runCommand` specify the | ||
| // `name` parameter. The `ExecutionListenerManager` only watches SQL executions with | ||
| // name. We can specify the execution name in more places in the future, so that | ||
| // `QueryExecutionListener` can track more cases. | ||
| event.executionName = name | ||
| event.duration = endTime - startTime | ||
| event.qe = queryExecution | ||
| event.executionFailure = ex | ||
| sc.listenerBus.post(event) | ||
| } | ||
| } | ||
| } | ||
| } finally { | ||
|
|
@@ -281,7 +283,10 @@ object SQLExecution extends Logging { | |
| val activeSession = sparkSession | ||
| val sc = sparkSession.sparkContext | ||
| val localProps = Utils.cloneProperties(sc.getLocalProperties) | ||
| // `getCurrentJobArtifactState` will return a stat only in Spark Connect mode. In non-Connect | ||
|
There was a problem hiding this comment. I think it should be safe to use the SparkSession's jobArtifactState. They should be the same. cc @vicennial. |
||
| // mode, we default back to the resources of the current Spark session. | ||
| val artifactState = JobArtifactSet.getCurrentJobArtifactState.getOrElse( | ||
| activeSession.artifactManager.state) | ||
| exec.submit(() => JobArtifactSet.withActiveJobArtifactState(artifactState) { | ||
| val originalSession = SparkSession.getActiveSession | ||
| val originalLocalProps = sc.getLocalProperties | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you use a UDF defined in the REPL? If so how does this work with a JobArtifactSet? Do we layer the globally defined classpath over the session specific classpath? (I'd be nice to document this somewhere).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I added one more test, which defines a UDF that initialises an external class added as an artifact.
Can you elaborate? Afaik
JobArtifactSetis not involved here since it's the artifact path that is applied when an active SparkSession is applied.Classpath - It's the other way around: the session classpath is laid over the global one.