pan3793 edited a comment on pull request #35076: URL: https://github.com/apache/spark/pull/35076#issuecomment-1007354842
Hi @otterc I got more information for this issue. Add assertion and debug log in `RemoteBlockPushResolver`(ESS side) ```java public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) { ... for (AppShufflePartitionInfo partition: shuffleMergePartitions.values()) { synchronized (partition) { try { // This can throw IOException which will marks this shuffle partition as not merged. partition.finalizePartition(); bitmaps.add(partition.mapTracker); reduceIds.add(partition.reduceId); sizes.add(partition.getLastChunkOffset()); } catch (IOException ioe) { logger.warn("Exception while finalizing shuffle partition {}_{} {} {}", msg.appId, msg.appAttemptId, msg.shuffleId, partition.reduceId, ioe); } finally { partition.closeAllFilesAndDeleteIfNeeded(false); } } + assert partition.dataFile.length() == partition.lastChunkOffset; + assert partition.indexFile.file.length() == partition.indexFile.getPos(); + assert partition.metaFile.file.length() == partition.metaFile.getPos(); + logger.info("shuffle partition {}_{} {} {}, chunk_size={}, meta_length={}, data_length={}", + msg.appId, msg.appAttemptId, msg.shuffleId, partition.reduceId, + partition.indexFile.getPos() / 8 - 1, + partition.metaFile.getPos(), + partition.lastChunkOffset); } mergeStatuses = new MergeStatuses(msg.shuffleId, msg.shuffleMergeId, bitmaps.toArray(new RoaringBitmap[0]), Ints.toArray(reduceIds), Longs.toArray(sizes)); } ... } ``` Add assertion and debug log in `IndexShuffleBlockResolver`(Reducer side) ```scala override def getMergedBlockData( blockId: ShuffleMergedBlockId, dirs: Option[Array[String]]): Seq[ManagedBuffer] = { val indexFile = getMergedBlockIndexFile(conf.getAppId, blockId.shuffleId, blockId.shuffleMergeId, blockId.reduceId, dirs) val dataFile = getMergedBlockDataFile(conf.getAppId, blockId.shuffleId, blockId.shuffleMergeId, blockId.reduceId, dirs) val metaFile = getMergedBlockMetaFile(conf.getAppId, blockId.shuffleId, blockId.shuffleMergeId, blockId.reduceId, dirs) // Load all the indexes in order to identify all chunks in the specified merged shuffle file. val size = indexFile.length.toInt val offsets = Utils.tryWithResource { new DataInputStream(Files.newInputStream(indexFile.toPath)) } { dis => val buffer = ByteBuffer.allocate(size) dis.readFully(buffer.array) buffer.asLongBuffer } // Number of chunks is number of indexes - 1 val numChunks = size / 8 - 1 + if (numChunks == 0) { + val indexBackupPath = java.nio.file.Paths.get(s"/tmp/${indexFile.toPath.getFileName}") + val dataBackupPath = java.nio.file.Paths.get(s"/tmp/${dataFile.toPath.getFileName}") + val metaBackupPath = java.nio.file.Paths.get(s"/tmp/${metaFile.toPath.getFileName}") + logError(s"$blockId chunk_size is 0, " + + s"index_file is $indexFile, backup to $indexBackupPath" + + s"data_file is $dataFile, backup to $dataBackupPath" + + s"meta_file is $metaFile, backup to $metaBackupPath") + Files.copy(indexFile.toPath, indexBackupPath) + Files.copy(dataFile.toPath, dataBackupPath) + Files.copy(metaFile.toPath, metaBackupPath) + assert(false) } for (index <- 0 until numChunks) yield { new FileSegmentManagedBuffer(transportConf, dataFile, offsets.get(index), offsets.get(index + 1) - offsets.get(index)) } } ``` Then I run TPCDS several rounds and reproduce the exception. Assertion failed in reduce task side. ```log 01-07 19:42:08 INFO [dag-scheduler-event-loop] scheduler.DAGScheduler@61: ShuffleMapStage 453 (save at QueryRunner.scala:98) failed in 1.811 s due to Job aborted due to stage failure: Task 122 in stage 453.0 failed 4 times, most recent failure: Lost task 122.3 in stage 453.0 (TID 278831) (beta-spark4 executor 562): java.lang.AssertionError: assertion failed at scala.Predef$.assert(Predef.scala:208) at org.apache.spark.shuffle.IndexShuffleBlockResolver.getMergedBlockData(IndexShuffleBlockResolver.scala:504) at org.apache.spark.storage.BlockManager.getLocalMergedBlockData(BlockManager.scala:777) at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:945) at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:85) at org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:29) ``` ESS logs ``` 2022-01-07 19:40:46,795 INFO shuffle.RemoteBlockPushResolver: shuffle partition application_1640143179334_0148_-1 126 4877, chunk_size=1, meta_length=18, data_length=157 ``` Reduce task backup merged shuffle files ``` root@beta-spark4:/tmp# ls -l shuffleMerged_application_1640143179334_0148_126_0_4877* -rw-r--r-- 1 root root 16036 Jan 7 19:41 shuffleMerged_application_1640143179334_0148_126_0_4877.data -rw-r--r-- 1 root root 8 Jan 7 19:41 shuffleMerged_application_1640143179334_0148_126_0_4877.index -rw-r--r-- 1 root root 0 Jan 7 19:41 shuffleMerged_application_1640143179334_0148_126_0_4877.meta ``` So, the ESS and reduce task running on same machine, and ESS closed the 'data', 'index', 'meta' files and reported the there size as `chunk_size=1, meta_length=18, data_length=157`, these metadata also return to driver and pass to reduce task, but when reduce task read the file from disk, the data is not match! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org