[
https://issues.apache.org/jira/browse/SPARK-18105?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17444386#comment-17444386
]
Wei Zhang edited comment on SPARK-18105 at 11/16/21, 1:16 PM:
--------------------------------------------------------------
In our case, it is strongly related with `spark.file.transferTo`. Set
`spark.file.transferTo` to false has saved our life.
We have been tortured by this problem for almost about a year, and we got about
hundreds of tasks with this `FetchFailedException: Decompression error:
Corrupted block detected` error everyday. Recently the heat is burning up and
got us kind of on fire since we are using Spark in more scenarios.
One of the stack I got is:
{code:java}
org.apache.spark.shuffle.FetchFailedException: Decompression error: Corrupted
block detected at
org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:569)
at
org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:474)
at
org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:66)
at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:435)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:441) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409) at
org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)
at
org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at
org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221)
at
org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1165)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1156)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1091)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1156)
at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:882)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:357) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:308) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:310) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:310) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:310) at
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) at
org.apache.spark.scheduler.Task.run(Task.scala:123) at
org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1405) at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414) 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)
Caused by: java.io.IOException: Decompression error: Corrupted block detected
at com.github.luben.zstd.ZstdInputStream.read(ZstdInputStream.java:111) at
java.io.BufferedInputStream.fill(BufferedInputStream.java:246) at
java.io.BufferedInputStream.read1(BufferedInputStream.java:286) at
java.io.BufferedInputStream.read(BufferedInputStream.java:345) at
org.apache.spark.util.Utils$$anonfun$1.apply$mcZ$sp(Utils.scala:402) at
org.apache.spark.util.Utils$$anonfun$1.apply(Utils.scala:397) at
org.apache.spark.util.Utils$$anonfun$1.apply(Utils.scala:397) at
org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1405) at
org.apache.spark.util.Utils$.copyStreamUpTo(Utils.scala:409) at
org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:467)
... 32 more{code}
We have Spark version 2.2.2, 2.4.7, 3.0.1 and 3.1.2，and all have this issue. I
tried methods in this post and others, nothing can make the non-repeatable
error disappear. After weeks' struggle and going through the code with one of
our almost repeatable case, we are able to find that `spark.file.transferTo` is
the one. We turned off this flag in our cluster, and this error literally
disappears – None of this error any more.
Normally we won't suspect flags like `spark.file.transferTo`, in that the only
difference is you are using `FileStream` or `FileChannel#transferTo` in shuffle
spill merge phase. And if that's the problem, it probably relates to JVM or
kernel bug. `transferTo` is more efficient than `FileStream` in that it copies
file segment with zero-copy, but empirically from our experience, this error
definitely IS caused by this flag.
So for those who are stilled bothered by this problem, `spark.file.transferTo`
is one thing to try. And if [~dongjoon] or somebody else is interested in this,
I can offer some more info if I can. Currently, we are using linux kernel
4.19.95 and JDK 1.8.0_171-b11.
was (Author: zhangweilst):
In our case, it is strongly related with `spark.file.transferTo`. Set
`spark.file.transferTo` to false has saved our life.
We have been tortured by this problem for almost about a year, and we got about
hundreds of tasks with this `FetchFailedException: Decompression error:
Corrupted block detected` error everyday. Recently the heat is burning up and
got us kind of on fire since we are using Spark in more scenarios.
One of the stack I got is:
{code:java}
org.apache.spark.shuffle.FetchFailedException: Decompression error: Corrupted
block detected at
org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:569)
at
org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:474)
at
org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:66)
at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:435)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:441) at
scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409) at
org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31)
at
org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at
org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:221)
at
org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:299)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1165)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1156)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1091)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1156)
at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:882)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:357) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:308) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:310) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:310) at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:310) at
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) at
org.apache.spark.scheduler.Task.run(Task.scala:123) at
org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1405) at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414) 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)
Caused by: java.io.IOException: Decompression error: Corrupted block detected
at com.github.luben.zstd.ZstdInputStream.read(ZstdInputStream.java:111) at
java.io.BufferedInputStream.fill(BufferedInputStream.java:246) at
java.io.BufferedInputStream.read1(BufferedInputStream.java:286) at
java.io.BufferedInputStream.read(BufferedInputStream.java:345) at
org.apache.spark.util.Utils$$anonfun$1.apply$mcZ$sp(Utils.scala:402) at
org.apache.spark.util.Utils$$anonfun$1.apply(Utils.scala:397) at
org.apache.spark.util.Utils$$anonfun$1.apply(Utils.scala:397) at
org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1405) at
org.apache.spark.util.Utils$.copyStreamUpTo(Utils.scala:409) at
org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:467)
... 32 more{code}
We have Spark version 2.2.2, 2.4.7, 3.0.1 and 3.1.2，and all have this issue. I
tried methods in this post and others, nothing can make the non-repeatable
error disappear. After weeks' struggle and going through the code with one of
our almost repeatable case, we are able to find that `spark.file.transferTo` is
the one. We turned off this flag in our cluster, and this error literally
disappears – None of this error any more.
Normally we won't suspect flags like `spark.file.transferTo`, in that the only
difference is you are using `FileStream` or `FileChannel#transferTo` in shuffle
spill merge phase. And if that's the problem, it probably relates to JVM or
kernel bug. `transferTo` is more efficient than `FileStream` in that it copies
file segment with zero-copy, but empirically from our experience, this error
definitely IS caused by this flag.
So for those who are stilled bothered by this problem, `spark.file.transferTo`
is one thing to try. And if [~dongjoon] or somebody else is interested in this,
I can offer some more info if I can.
> LZ4 failed to decompress a stream of shuffled data
> --------------------------------------------------
>
> Key: SPARK-18105
> URL: https://issues.apache.org/jira/browse/SPARK-18105
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 3.0.1, 3.1.1
> Reporter: Davies Liu
> Priority: Major
> Attachments: TestWeightedGraph.java
>
>
> When lz4 is used to compress the shuffle files, it may fail to decompress it
> as "stream is corrupt"
> {code}
> Caused by: org.apache.spark.SparkException: Job aborted due to stage failure:
> Task 92 in stage 5.0 failed 4 times, most recent failure: Lost task 92.3 in
> stage 5.0 (TID 16616, 10.0.27.18): java.io.IOException: Stream is corrupted
> at
> org.apache.spark.io.LZ4BlockInputStream.refill(LZ4BlockInputStream.java:220)
> at
> org.apache.spark.io.LZ4BlockInputStream.available(LZ4BlockInputStream.java:109)
> at java.io.BufferedInputStream.read(BufferedInputStream.java:353)
> at java.io.DataInputStream.read(DataInputStream.java:149)
> at com.google.common.io.ByteStreams.read(ByteStreams.java:828)
> at com.google.common.io.ByteStreams.readFully(ByteStreams.java:695)
> at
> org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$3$$anon$1.next(UnsafeRowSerializer.scala:127)
> at
> org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$3$$anon$1.next(UnsafeRowSerializer.scala:110)
> at scala.collection.Iterator$$anon$13.next(Iterator.scala:372)
> at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
> at
> org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:30)
> at
> org.apache.spark.InterruptibleIterator.next(InterruptibleIterator.scala:43)
> at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
> at
> org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.sort_addToSorter$(Unknown
> Source)
> at
> org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown
> Source)
> at
> org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
> at
> org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:370)
> at
> org.apache.spark.sql.execution.datasources.DynamicPartitionWriterContainer.writeRows(WriterContainer.scala:397)
> at
> org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(InsertIntoHadoopFsRelationCommand.scala:143)
> at
> org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(InsertIntoHadoopFsRelationCommand.scala:143)
> at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
> at org.apache.spark.scheduler.Task.run(Task.scala:86)
> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> {code}
> https://github.com/jpountz/lz4-java/issues/89
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
