Skip to content

[CORE] Decrease offheap memory size in resource profile for whole stage fallback case #8911

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jackylee-ch merged 2 commits into from
Apr 9, 2025
Merged

[CORE] Decrease offheap memory size in resource profile for whole stage fallback case #8911

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
584dc71
Initial
philo-he Mar 5, 2025
3dde772
Update doc
philo-he Apr 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions .../src/main/scala/org/apache/spark/sql/execution/GlutenAutoAdjustStageResourceProfile.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,10 @@ import scala.collection.mutable.ArrayBuffer

/**
* This rule is used to dynamic adjust stage resource profile for following purposes:
* 1. swap offheap and onheap memory size when whole stage fallback happened 2. increase executor
* heap memory if stage contains gluten operator and spark operator at the same time. Note: we
* don't support set resource profile for final stage now. Todo: support set resource profile
* for final stage.
* 1. Decrease offheap and increase onheap memory size when whole stage fallback happened; 2.
* Increase executor heap memory if stage contains gluten operator and spark operator at the
* same time. Note: we don't support set resource profile for final stage now. Todo: will
* support it.
*/
@Experimental
case class GlutenAutoAdjustStageResourceProfile(glutenConf: GlutenConfig, spark: SparkSession)
Expand Down Expand Up @@ -88,6 +88,11 @@ case class GlutenAutoAdjustStageResourceProfile(glutenConf: GlutenConfig, spark:
val newExecutorMemory =
new ExecutorResourceRequest(ResourceProfile.MEMORY, newMemoryAmount.toLong)
executorResource.put(ResourceProfile.MEMORY, newExecutorMemory)

val newExecutorOffheap =
new ExecutorResourceRequest(ResourceProfile.OFFHEAP_MEM, offheapRequest.get.amount / 10)
Copy link
Copy Markdown
Member Author

@philo-he philo-he Apr 2, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@zjuwangg, could you take a look? This is just an empirical setting. And I am not sure which setting is better.

Copy link
Copy Markdown
Contributor

@zjuwangg zjuwangg Apr 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM in whole stage fallback case!

Copy link
Copy Markdown
Contributor

@jackylee-ch jackylee-ch Apr 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This would work only for Spark 3.5.4 or higher as apache/spark#48963

Copy link
Copy Markdown
Contributor

@JkSelf JkSelf Apr 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@philo-he Should we consider making this setting configurable?

Copy link
Copy Markdown
Member Author

@philo-he philo-he Apr 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@JkSelf, let's first use this empirical fixed setting and wait for feedback from user side. In theory, user don't need to configure too much offheap memory in stage fallback case. Thanks!

Copy link
Copy Markdown
Member

@zhouyuan zhouyuan Apr 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@philo-he
Thanks for improving on this!
just to confirm, I can see the code logic is to reduce the offheap memory size to 1/10, but on-heap memory size is not increased, is this intended?

Thanks, -yuan

Copy link
Copy Markdown
Member Author

@philo-he philo-he Apr 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@zhouyuan, the on-heap memory is increased by specifying a larger value for ResourceProfile.MEMORY prior to the off-heap adjustment. If I have missed something, please let me know. Thanks!

executorResource.put(ResourceProfile.OFFHEAP_MEM, newExecutorOffheap)

val newRP = new ResourceProfile(executorResource.toMap, taskResource.toMap)
return applyNewResourceProfileIfPossible(plan, newRP, rpManager)
}
Expand Down
Loading