Consider additional factors in spark.sql.shuffle.partitions recommendation in Autotuner #722

cindyyuanjiang · 2024-01-10T02:06:36Z

Fixes #575

Goal
We want to enhance the spark.sql.shuffle.partitions recommendation logic to include more factors to improve accuracy, e.g. GC time, data size, data skew, etc.

Changes
In this PR, we consider spilling and data skew in shuffle stages for spark.sql.shuffle.partitions recommendation logic. Specifically:

If spilling occurred in shuffle stages:

If there is no shuffle skew, we recommend increasing spark.sql.shuffle.partitions(Previously we only check spilling in any stage).
If there is shuffle skew (when task's Shuffle Read Size > 3 * Avg Stage-level size), we add a comment to highlight this and keep recommending default spark.sql.shuffle.partitions.

…ation Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

kuhushukla · 2024-01-11T15:12:12Z

I will look at this closely. In the meantime, can we run some manual and even write unit tests that can show the before and after recommendations. Did we try this with different stage level spill metrics and observe better partition numbers being suggested?

amahussein · 2024-01-19T14:38:41Z

Is there any progress on this?

cindyyuanjiang · 2024-01-19T23:15:35Z

Is there any progress on this?

I will comment my test results, then mark this ready for review.

cindyyuanjiang · 2024-01-19T23:16:20Z

I will look at this closely. In the meantime, can we run some manual and even write unit tests that can show the before and after recommendations. Did we try this with different stage level spill metrics and observe better partition numbers being suggested?

Thanks @kuhushukla! I will comment on my observations.

cindyyuanjiang · 2024-01-20T01:16:24Z

Here is a comparison of the recommendations before and after this PR.
Before:

After:

cc: @kuhushukla @amahussein

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala

amahussein

The changes in the PR have no effect on the AutoTuner.
A unit-test would have helped to verify that the AutoTuner will recommend the default settings without running any calculations.

amahussein · 2024-01-22T20:19:44Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala

+      val shuffleStageSpilledMetrics = appInfoProvider.getSpillMetricsForStage("shuffle")
+
+      val dataSkewStages = appInfoProvider.getDataSkewStages
+      val dataSkewShuffleStages = shuffleStageSpilledMetrics.collect({
+        case (id, _) if dataSkewStages.contains(id) => id
+      })
+      if (!dataSkewShuffleStages.isEmpty) {
+        appendOptionalComment(lookup, "There is data skew (when task's Shuffle Read Size > 3 * " +
+          "Avg Stage-level size) in shuffle stages.")
+      }
+
+      val totalShuffleSpilledMetrics = shuffleStageSpilledMetrics.unzip._2.sum
+      if (totalShuffleSpilledMetrics > 0) {


I believe this code represents a dead-block that will never be executed.
isCalculationEnabled(lookup) always returns false for spark.sql.shuffle.partitions

I have run some tests, and confirm this is not dead code. isCalculationEnabled(lookup) returns true for command below.
Command: spark_rapids_user_tools onprem profiling --worker_info <my-worker-info> --verbose --eventlogs <my-event-log> -t $TOOLS_REPO/spark-rapids-tools/core/target/rapids-4-spark-tools_2.12-23.12.4-SNAPSHOT.jar

amahussein · 2024-01-22T20:35:05Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala

@@ -52,6 +52,8 @@ trait AppInfoPropertyGetter {
 trait AppInfoSqlTaskAggMetricsVisitor {
  def getJvmGCFractions: Seq[Double]
  def getSpilledMetrics: Seq[Long]
+  def getSpillMetricsForStage(stage: String): Seq[(Long, Long)]


Since the map is not part of the function name, Please add a comment of what the map keys-values are expected to be

Why do we have two APIs getSpilledMetrics and getSpillMetricsForStage ?

The definition is confusing.
When I see getSpillMetricsForStage then I expect that the method takes a stageID as an argument. But the callers pass "shuffle" which is not stage. It looks more like taking a certain predicate to filter the nodes. In that case, this should not be in the upper level of the Interface.
// aggregates the sum of spill per stage and returns a map stageID to sum spill metrics def getSpillMetricsForShufflePerStage: Seq[(Long, Long)]

Update this function. Thanks!

amahussein · 2024-01-22T20:48:31Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala

+    app.skewInfo.collect({
+      case row if (row.taskShuffleReadMB > 3 * row.avgShuffleReadMB) => row.stageId


Dow we need this condition? I thought that the purpose of ShuffleSkewProfileResult (see analysis code) is to filter tasks which have skew. So, we guarantee that any entry in that sequence exhibits skew.

Thanks, I will update this.

amahussein · 2024-01-22T20:54:39Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala

@@ -81,6 +83,8 @@ class AppSummaryInfoBaseProvider extends AppInfoPropertyGetter
  override def getRapidsJars: Seq[String] = Seq()
  override def getDistinctLocationPct: Double = 0.0
  override def getRedundantReadSize: Long = 0
+  override def getSpillMetricsForStage(stage: String): Seq[(Long, Long)] = Seq()


see comment on the super method

amahussein · 2024-01-22T20:57:36Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala

+  override def getSpillMetricsForStage(stage: String): Seq[(Long, Long)] = {
+    // Get ids for input specified stage
+    val stageIds = app.sqlStageInfo.collect ({
+      case row if (row.nodeNames.exists(name => name.toLowerCase.contains(stage))) =>


why do we need to match nodeNames on "shuffle"? Should we match on the metric type? For example "AverageStageInfo.avgShuffleRead"

Thanks, I have updated the condition to identify shuffle stages to depend on shuffle read/write.

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

parthosa

Thanks @cindyyuanjiang. LGTM.

amahussein

I was hoping that in this iteration you investigate deeper my comment that the code was in a dead-block.
Anyway, I found that there was a bug in the way the arguments were processed in getRecommendations. This is a bug that I was hoping you address it in your PR after I mentioned the default arguments in
getRecommendations
I created a PR #812 to fix that bug.
The above PR #812 should get merged first because it is a bug that affects the path of code target of change here.

I don't have clues of how to verify the calculations. Most of the GPU eventlogs I tested recommends 200 which is the default spark.sql.shuffle.partitions hardcoded in the code.

cindyyuanjiang · 2024-02-27T20:06:56Z

I was hoping that in this iteration you investigate deeper my comment that the code was in a dead-block. Anyway, I found that there was a bug in the way the arguments were processed in getRecommendations.

Apologies, I misunderstood this in our previous discussion. Thank you for putting up a fix @amahussein!

I don't have clues of how to verify the calculations. Most of the GPU eventlogs I tested recommends 200 which is the default spark.sql.shuffle.partitions hardcoded in the code.

I will share my manually tested results here and follow up with them.

amahussein

PR #812 is merged. @cindyyuanjiang please merge with dev and address the new comments.
We need a way to verify the calculations.
In worst case, We could use the mock classes to return literal numbers to verify the functionality.

amahussein · 2024-02-27T19:54:30Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala

        shufflePartitions *= DEF_SHUFFLE_PARTITION_MULTIPLIER
        // Could be memory instead of partitions
        appendOptionalComment(lookup,
          s"'$lookup' should be increased since spilling occurred.")
      }
+
+      val shuffleSkewStages = appInfoProvider.getShuffleSkewStages
+      val shuffleSkewStagesWithSpill = appInfoProvider.getSpillMetricsForShuffleStages.filter {


It is redundant to call the same method twice appInfoProvider.getSpillMetricsForShuffleStages. One time to get the totalSpilledMetricsForShuffleStages and another one applying the filter.
It should be stored into a local variable.

amahussein · 2024-02-27T19:58:10Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala

        shufflePartitions *= DEF_SHUFFLE_PARTITION_MULTIPLIER
        // Could be memory instead of partitions
        appendOptionalComment(lookup,
          s"'$lookup' should be increased since spilling occurred.")
      }
+
+      val shuffleSkewStages = appInfoProvider.getShuffleSkewStages
+      val shuffleSkewStagesWithSpill = appInfoProvider.getSpillMetricsForShuffleStages.filter {


It does not look like we actually need the shuffleSkewStagesWithSpill. The following code only checks if it is empty or not.
A filter in that case is not suitable for performance reasons. The code should use a predicate like "exists" to return true once an item satisfies the condition.

amahussein · 2024-02-27T20:02:12Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala

+      }
+      if (shuffleSkewStagesWithSpill.nonEmpty) {
+        appendOptionalComment(lookup, "There is data skew (when task's Shuffle Read Size > 3 * " +
+          s"Avg Stage-level size) in shuffle stages. $lookup recommendation may be less effective.")


"bla..bla...$lookup recommendation may be less effective."

I do not find this comment easy to understand. "Less effective", what does it mean for the user? Is there a better way to explain this? If we want to say that the recommendation is not that good in the user's case, then why do we recommend it.

Updated the message and the logic. Thanks!

amahussein · 2024-02-27T20:05:38Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala

-      val totalSpilledMetrics = appInfoProvider.getSpilledMetrics.sum
-      if (totalSpilledMetrics > 0) {
+      val totalSpilledMetricsForShuffleStages =
+        appInfoProvider.getSpillMetricsForShuffleStages.unzip._2.sum


For performance reason, we don't need to get the sum. "sum" iterates on the entire list.
We only need to check if any of the values is larger than 0. A predicate like exists should work better.

amahussein · 2024-02-27T20:21:38Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala

+    app.jsMetAgg.collect { case row if (row.id.contains("stage") &&
+      row.srTotalBytesReadSum + row.swBytesWrittenSum > 0) =>
+        (row.id.split("_")(1).toLong, row.diskBytesSpilledSum + row.memoryBytesSpilledSum)
+    }


@nartal1 Can you help Cindy with making sure how to pick "shuffles"?
are row.srTotalBytesReadSum and swBytesWrittenSum exclusive to shuffle operations? IF not, then what is the right conditions to successfully visit all the shuffles?

srTotalBytesReadSum is the sum of local and remote bytes read in shuffle metrics across all tasks and swBytesWrittenSum is the sum of bytes written for the shuffle across all tasks. The above condition is correct.
So this is getting the spilled metrics for those shuffle stages(which will have either shuffle read or write).

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

cindyyuanjiang · 2024-03-06T22:35:33Z

Note currently spark.sql.shuffle.partitions recommendation only recommends the default value 200. To enable the calculation based on changes in this PR, users have to override argument limitedLogicList passed to getRecommendedProperties.

However, users who use the tools jar cannot access the source code which makes the shuffle partition calculation a dead code block. To improve this, we can either take spark.sql.shuffle.partitions off the limitedLogicList or expose limitedLogicList to the users, e.g. as command line option. Opening the discussion here.

nartal1 · 2024-03-06T23:31:52Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala

-          s"'$lookup' should be increased since spilling occurred.")
+      if (shuffleStagesWithPosSpilling.nonEmpty) {
+        val shuffleSkewStages = appInfoProvider.getShuffleSkewStages
+        if (shuffleSkewStages.exists(id => shuffleStagesWithPosSpilling.contains(id))) {


This check would make sure that the autotuner would not recommend increasing the shuffle partitions if there is atleast one stage with skewed data. Was wondering instead of checking for atleast one, can we still recommend increasing number of shuffle pariitions until a certain threshold (Ex: ratio of number of number of stages with skew to total number of stages with spills ).
This can be a follow-on as in improvement if it takes more time to test.

Interesting. What would be a good way to determine the threshold?

Also, since both are lists of stage ids, can we use intersect for better performance?

Suggested change

if (shuffleSkewStages.exists(id => shuffleStagesWithPosSpilling.contains(id))) {

if (shuffleSkewStages.intersect(shuffleStagesWithPosSpilling).nonEmpty) {

@nartal1 Thanks! I think this would require more discussion/testing. I can file a follow up issue to track this when we are ready to merge this PR.

@parthosa Thanks! I used exists because it will short-circuit after it finds an element which is true for the condition, but intersect will go over the entire sequence. In that case, I think exists has better performance than intersect.

In theory a.intersect(b) should have complexity of O(min(a, b)) and a.exists(b.contains) can have complexity of O(a*b).

Although exists can short circuit, the inner contains cause a full iteration of the list every time.

P.S:getShuffleStagesWithPosSpilling already tries to find all the IDs. If we want a btetr performance then we will need to use another API that returns True/False if at least one exists. I did not want to request that change initially to keep the API standard for the first phase.

Back to the performance discussion here, In worst case b.contains is O(n) because b is a sequence. This is easily can be improved if we change the returnType of getShuffleStagesWithPosSpilling to set instead of sequence. this yields into a faster lookup.

a.intersect(b) the problem here that those are also sequences, which means that O(a*b). Even if we change it to Set, the output of the operation is the length of the intersection which is bigger than an operation that tries to answer "True/False"

For the current approach, we need the full list of getShuffleStagesWithPosSpilling's output to compare with shuffleSkewStages to check if they have any intersection. I will update getShuffleStagesWithPosSpilling and getShuffleSkewStages to return type of Set[Long] for better performance.

amahussein

Thanks @cindyyuanjiang

The argument of the function is disabled by default.
This means that the Profiler calls the AutoTuner while setting limitedLogicList to 'spark.sql.shuffle.partitions'. As a result the final behavior won't execute any of the code changes.

The unitTests overrides the default argument but the Profile code has not changed.

https://github.com/NVIDIA/spark-rapids-tools/blob/main/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/Profiler.scala#L547
https://github.com/NVIDIA/spark-rapids-tools/blob/main/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/Profiler.scala#L141

So, either

you change the default argument "limitedLogicList" of getRecommendedProperties to be empty list https://github.com/NVIDIA/spark-rapids-tools/blob/main/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala#L993 ;
or override all the calls to getRecommendedProperties in production code to pass an empty sequence for limitedLogicList

amahussein

Is it possible to use set instead of seq for better performance.

amahussein · 2024-03-11T17:47:49Z

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala

+  override def getShuffleStagesWithPosSpilling: Seq[Long] = Seq()
+  override def getShuffleSkewStages: Seq[Long] = Seq()


The return type could be changed to Set to have optimize lookups.

Thanks, updated to Set.

cindyyuanjiang · 2024-03-11T21:03:41Z

The argument of the function is disabled by default. This means that the Profiler calls the AutoTuner while setting limitedLogicList to 'spark.sql.shuffle.partitions'. As a result the final behavior won't execute any of the code changes.

Thanks @amahussein for the feedback! I will update argument limitedLogicList of getRecommendedProperties to be empty list so the code would execute shuffle partitions recommendation calculation.

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

amahussein

Thanks @cindyyuanjiang

nartal1

LGTM. Thanks @cindyyuanjiang !

check for shuffle stage and data skew in shuffle partitions recommend…

b855321

…ation Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

amahussein assigned cindyyuanjiang Jan 10, 2024

amahussein added core_tools Scope the core module (scala) bug Something isn't working labels Jan 10, 2024

cindyyuanjiang added 2 commits January 10, 2024 11:11

fixed data skew with shuffle stage var

f2b26ec

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

refactored code

94e36ed

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

cindyyuanjiang marked this pull request as ready for review January 20, 2024 01:20

parthosa reviewed Jan 22, 2024

View reviewed changes

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala Outdated Show resolved Hide resolved

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ApplicationSummaryInfo.scala Outdated Show resolved Hide resolved

amahussein requested changes Jan 22, 2024

View reviewed changes

amahussein closed this Jan 24, 2024

Merge branch 'dev' into shuffle-partition

b23f28b

cindyyuanjiang reopened this Feb 22, 2024

refactored code to improve shuffle partition recommendation

e5958a0

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

cindyyuanjiang requested review from amahussein and parthosa February 26, 2024 18:57

parthosa reviewed Feb 26, 2024

View reviewed changes

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala Outdated Show resolved Hide resolved

use more concise function call

86bf788

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

parthosa previously approved these changes Feb 27, 2024

View reviewed changes

amahussein requested changes Feb 27, 2024

View reviewed changes

amahussein reviewed Feb 27, 2024

View reviewed changes

cindyyuanjiang added 3 commits February 27, 2024 13:14

Merge branch 'dev' into shuffle-partition

cef91c2

update function shuffleStagesWithPosSpilling.contains for better perf

674ba5e

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

added print statements

8fa47a6

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

removed print statements

d11672f

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

cindyyuanjiang dismissed parthosa’s stale review via d11672f March 5, 2024 01:42

cindyyuanjiang added 3 commits March 4, 2024 17:55

removed print

806e962

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

added test cases

78def6d

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

update function comments

6ad07cd

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

cindyyuanjiang requested review from nartal1, parthosa and amahussein March 6, 2024 22:36

nartal1 reviewed Mar 6, 2024

View reviewed changes

amahussein requested changes Mar 11, 2024

View reviewed changes

amahussein reviewed Mar 11, 2024

View reviewed changes

addressed review feedback

161247d

Signed-off-by: cindyyuanjiang <cindyj@nvidia.com>

cindyyuanjiang requested review from nartal1 and amahussein March 11, 2024 21:21

amahussein approved these changes Mar 11, 2024

View reviewed changes

nartal1 approved these changes Mar 12, 2024

View reviewed changes

cindyyuanjiang merged commit f10073b into NVIDIA:dev Mar 12, 2024
13 checks passed

cindyyuanjiang deleted the shuffle-partition branch March 12, 2024 01:06

		app.skewInfo.collect({
		case row if (row.taskShuffleReadMB > 3 * row.avgShuffleReadMB) => row.stageId

	if (shuffleSkewStages.exists(id => shuffleStagesWithPosSpilling.contains(id))) {
	if (shuffleSkewStages.intersect(shuffleStagesWithPosSpilling).nonEmpty) {

		override def getShuffleStagesWithPosSpilling: Seq[Long] = Seq()
		override def getShuffleSkewStages: Seq[Long] = Seq()

Consider additional factors in spark.sql.shuffle.partitions recommendation in Autotuner #722

Consider additional factors in spark.sql.shuffle.partitions recommendation in Autotuner #722

Conversation

cindyyuanjiang commented Jan 10, 2024 • edited Loading

kuhushukla commented Jan 11, 2024

amahussein commented Jan 19, 2024

cindyyuanjiang commented Jan 19, 2024

cindyyuanjiang commented Jan 19, 2024

cindyyuanjiang commented Jan 20, 2024 • edited Loading

amahussein left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

parthosa left a comment • edited Loading

Choose a reason for hiding this comment

amahussein left a comment

Choose a reason for hiding this comment

cindyyuanjiang commented Feb 27, 2024

amahussein left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cindyyuanjiang commented Mar 6, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amahussein Mar 11, 2024 • edited Loading

Choose a reason for hiding this comment

cindyyuanjiang Mar 11, 2024 • edited Loading

Choose a reason for hiding this comment

amahussein left a comment

Choose a reason for hiding this comment

amahussein left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cindyyuanjiang commented Mar 11, 2024

amahussein left a comment

Choose a reason for hiding this comment

nartal1 left a comment

Choose a reason for hiding this comment

cindyyuanjiang commented Jan 10, 2024 •

edited

Loading

cindyyuanjiang commented Jan 20, 2024 •

edited

Loading

parthosa left a comment •

edited

Loading

cindyyuanjiang commented Mar 6, 2024 •

edited

Loading

amahussein Mar 11, 2024 •

edited

Loading

cindyyuanjiang Mar 11, 2024 •

edited

Loading

amahussein left a comment •

edited

Loading