[SPARK-37695][CORE][SHUFFLE] Skip diagnosis ob merged blocks from push-based shuffle #34961

pan3793 · 2021-12-20T13:04:50Z

What changes were proposed in this pull request?

Skip diagnosis ob merged blocks from push-based shuffle

Why are the changes needed?

Because SPARK-36284 has not been addressed yet, skip it to suppress exceptions.

21/12/19 18:46:37 WARN TaskSetManager: Lost task 166.0 in stage 1921.0 (TID 138855) (beta-spark5 executor 218): java.lang.AssertionError: assertion failed: Expected ShuffleBlockId, but got shuffleChunk_464_0_5645_0
	at scala.Predef$.assert(Predef.scala:223)
	at org.apache.spark.storage.ShuffleBlockFetcherIterator.diagnoseCorruption(ShuffleBlockFetcherIterator.scala:1043)
	at org.apache.spark.storage.BufferReleasingInputStream.$anonfun$tryOrFetchFailedException$1(ShuffleBlockFetcherIterator.scala:1308)
	at scala.Option.map(Option.scala:230)
	at org.apache.spark.storage.BufferReleasingInputStream.tryOrFetchFailedException(ShuffleBlockFetcherIterator.scala:1307)
	at org.apache.spark.storage.BufferReleasingInputStream.read(ShuffleBlockFetcherIterator.scala:1293)
	at java.io.BufferedInputStream.fill(BufferedInputStream.java:246)
	at java.io.BufferedInputStream.read(BufferedInputStream.java:265)
	at java.io.DataInputStream.readInt(DataInputStream.java:387)
	at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.readSize(UnsafeRowSerializer.scala:113)
	at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.<init>(UnsafeRowSerializer.scala:120)
	at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2.asKeyValueIterator(UnsafeRowSerializer.scala:110)
	at org.apache.spark.shuffle.BlockStoreShuffleReader.$anonfun$read$2(BlockStoreShuffleReader.scala:98)
	at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
	at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.sort_addToSorter_0$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.smj_findNextJoinRows_0$(Unknown Source)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:778)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
	at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:179)
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
	at org.apache.spark.scheduler.Task.run(Task.scala:136)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:507)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1468)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:510)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Does this PR introduce any user-facing change?

Yes, suppress the exceptions.

How was this patch tested?

Run 1T TPCDS manually.

Ngone51 · 2021-12-20T13:10:18Z

core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala

+          .filterNot { _ => blockId.isShuffleChunk }
+          .map { checkedIn =>
+            iterator.diagnoseCorruption(checkedIn, address, blockId)
+          }


Shall we do the check inside diagnoseCorruption() to ensure other calls on it are also safe?

This reverts commit a853f8e.

Ngone51 · 2021-12-21T01:29:39Z

core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala

+      case shuffleBlockChunk: ShuffleBlockChunkId =>
+        // TODO SPARK-36284 Add shuffle checksum support for push-based shuffle
+        val diagnosisResponse = s"BlockChunk $shuffleBlockChunk is corrupted but corruption " +
+          s"diagnosis failed due to lack of shuffle checksum support for push-based shuffle."


"...diagnosis is skipped due to..."?

Ngone51 · 2021-12-21T01:31:30Z

core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala

@@ -1037,39 +1037,48 @@ final class ShuffleBlockFetcherIterator(
      blockId: BlockId): String = {
    logInfo("Start corruption diagnosis.")
    val startTimeNs = System.nanoTime()


This line could move to ShuffleBlockId case only.

Ngone51 · 2021-12-21T01:32:11Z

LGTM excepts 2 minor comments.

Ngone51 · 2021-12-21T01:32:18Z

cc @mridulm

yaooqinn · 2021-12-21T07:21:38Z

ok to test

yaooqinn

LGTM

SparkQA · 2021-12-21T08:32:53Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50902/

SparkQA · 2021-12-21T09:30:46Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50902/

SparkQA · 2021-12-21T10:50:36Z

Test build #146427 has finished for PR 34961 at commit ce42f59.

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

AmplabJenkins · 2021-12-21T13:44:10Z

Can one of the admins verify this patch?

…h-based shuffle ### What changes were proposed in this pull request? Skip diagnosis ob merged blocks from push-based shuffle ### Why are the changes needed? Because SPARK-36284 has not been addressed yet, skip it to suppress exceptions. ``` 21/12/19 18:46:37 WARN TaskSetManager: Lost task 166.0 in stage 1921.0 (TID 138855) (beta-spark5 executor 218): java.lang.AssertionError: assertion failed: Expected ShuffleBlockId, but got shuffleChunk_464_0_5645_0 at scala.Predef$.assert(Predef.scala:223) at org.apache.spark.storage.ShuffleBlockFetcherIterator.diagnoseCorruption(ShuffleBlockFetcherIterator.scala:1043) at org.apache.spark.storage.BufferReleasingInputStream.$anonfun$tryOrFetchFailedException$1(ShuffleBlockFetcherIterator.scala:1308) at scala.Option.map(Option.scala:230) at org.apache.spark.storage.BufferReleasingInputStream.tryOrFetchFailedException(ShuffleBlockFetcherIterator.scala:1307) at org.apache.spark.storage.BufferReleasingInputStream.read(ShuffleBlockFetcherIterator.scala:1293) at java.io.BufferedInputStream.fill(BufferedInputStream.java:246) at java.io.BufferedInputStream.read(BufferedInputStream.java:265) at java.io.DataInputStream.readInt(DataInputStream.java:387) at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.readSize(UnsafeRowSerializer.scala:113) at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.<init>(UnsafeRowSerializer.scala:120) at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2.asKeyValueIterator(UnsafeRowSerializer.scala:110) at org.apache.spark.shuffle.BlockStoreShuffleReader.$anonfun$read$2(BlockStoreShuffleReader.scala:98) at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31) at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.sort_addToSorter_0$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.smj_findNextJoinRows_0$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:778) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:179) at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) at org.apache.spark.scheduler.Task.run(Task.scala:136) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:507) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1468) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:510) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) ``` ### Does this PR introduce _any_ user-facing change? Yes, suppress the exceptions. ### How was this patch tested? Run 1T TPCDS manually. Closes #34961 from pan3793/SPARK-37695. Authored-by: Cheng Pan <chengpan@apache.org> Signed-off-by: yi.wu <yi.wu@databricks.com> (cherry picked from commit 57ca75f) Signed-off-by: yi.wu <yi.wu@databricks.com>

Ngone51 · 2021-12-23T04:02:55Z

Thanks, merged to master, branch-3.2.

…h-based shuffle ### What changes were proposed in this pull request? Skip diagnosis ob merged blocks from push-based shuffle ### Why are the changes needed? Because SPARK-36284 has not been addressed yet, skip it to suppress exceptions. ``` 21/12/19 18:46:37 WARN TaskSetManager: Lost task 166.0 in stage 1921.0 (TID 138855) (beta-spark5 executor 218): java.lang.AssertionError: assertion failed: Expected ShuffleBlockId, but got shuffleChunk_464_0_5645_0 at scala.Predef$.assert(Predef.scala:223) at org.apache.spark.storage.ShuffleBlockFetcherIterator.diagnoseCorruption(ShuffleBlockFetcherIterator.scala:1043) at org.apache.spark.storage.BufferReleasingInputStream.$anonfun$tryOrFetchFailedException$1(ShuffleBlockFetcherIterator.scala:1308) at scala.Option.map(Option.scala:230) at org.apache.spark.storage.BufferReleasingInputStream.tryOrFetchFailedException(ShuffleBlockFetcherIterator.scala:1307) at org.apache.spark.storage.BufferReleasingInputStream.read(ShuffleBlockFetcherIterator.scala:1293) at java.io.BufferedInputStream.fill(BufferedInputStream.java:246) at java.io.BufferedInputStream.read(BufferedInputStream.java:265) at java.io.DataInputStream.readInt(DataInputStream.java:387) at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.readSize(UnsafeRowSerializer.scala:113) at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.<init>(UnsafeRowSerializer.scala:120) at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2.asKeyValueIterator(UnsafeRowSerializer.scala:110) at org.apache.spark.shuffle.BlockStoreShuffleReader.$anonfun$read$2(BlockStoreShuffleReader.scala:98) at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31) at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.sort_addToSorter_0$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.smj_findNextJoinRows_0$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage10.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:778) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:179) at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) at org.apache.spark.scheduler.Task.run(Task.scala:136) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:507) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1468) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:510) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) ``` ### Does this PR introduce _any_ user-facing change? Yes, suppress the exceptions. ### How was this patch tested? Run 1T TPCDS manually. Closes apache#34961 from pan3793/SPARK-37695. Authored-by: Cheng Pan <chengpan@apache.org> Signed-off-by: yi.wu <yi.wu@databricks.com> (cherry picked from commit 57ca75f) Signed-off-by: yi.wu <yi.wu@databricks.com>

Skip diagnosis ob merged blocks from push-based shuffle

a853f8e

github-actions bot added the CORE label Dec 20, 2021

Ngone51 reviewed Dec 20, 2021

View reviewed changes

pan3793 added 2 commits December 20, 2021 21:22

Move to diagnoseCorruption

424d633

Revert "Skip diagnosis ob merged blocks from push-based shuffle"

886ac05

This reverts commit a853f8e.

Ngone51 reviewed Dec 21, 2021

View reviewed changes

Address comments

ce42f59

yaooqinn approved these changes Dec 21, 2021

View reviewed changes

Ngone51 approved these changes Dec 23, 2021

View reviewed changes

Ngone51 closed this in 57ca75f Dec 23, 2021

pan3793 deleted the SPARK-37695 branch December 27, 2021 12:15

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-37695][CORE][SHUFFLE] Skip diagnosis ob merged blocks from push-based shuffle #34961

[SPARK-37695][CORE][SHUFFLE] Skip diagnosis ob merged blocks from push-based shuffle #34961

pan3793 commented Dec 20, 2021 •

edited

Loading

Ngone51 Dec 20, 2021

pan3793 Dec 20, 2021

Ngone51 Dec 21, 2021

pan3793 Dec 21, 2021

Ngone51 Dec 21, 2021

pan3793 Dec 21, 2021

Ngone51 commented Dec 21, 2021

Ngone51 commented Dec 21, 2021

yaooqinn commented Dec 21, 2021

yaooqinn left a comment

SparkQA commented Dec 21, 2021

SparkQA commented Dec 21, 2021

SparkQA commented Dec 21, 2021

AmplabJenkins commented Dec 21, 2021

Ngone51 commented Dec 23, 2021

[SPARK-37695][CORE][SHUFFLE] Skip diagnosis ob merged blocks from push-based shuffle #34961

[SPARK-37695][CORE][SHUFFLE] Skip diagnosis ob merged blocks from push-based shuffle #34961

Conversation

pan3793 commented Dec 20, 2021 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Ngone51 Dec 20, 2021

Choose a reason for hiding this comment

pan3793 Dec 20, 2021

Choose a reason for hiding this comment

Ngone51 Dec 21, 2021

Choose a reason for hiding this comment

pan3793 Dec 21, 2021

Choose a reason for hiding this comment

Ngone51 Dec 21, 2021

Choose a reason for hiding this comment

pan3793 Dec 21, 2021

Choose a reason for hiding this comment

Ngone51 commented Dec 21, 2021

Ngone51 commented Dec 21, 2021

yaooqinn commented Dec 21, 2021

yaooqinn left a comment

Choose a reason for hiding this comment

SparkQA commented Dec 21, 2021

SparkQA commented Dec 21, 2021

SparkQA commented Dec 21, 2021

AmplabJenkins commented Dec 21, 2021

Ngone51 commented Dec 23, 2021

pan3793 commented Dec 20, 2021 •

edited

Loading