[SPARK-29285][Shuffle] Temporary shuffle files should be able to handle disk failures

[yaooqinn]

What changes were proposed in this pull request?

The getFile method in DiskBlockManager may return a file with an existing subdirectory. But when a disk failure occurs on that subdirectory. this file is inaccessible.
Then the FileNotFoundException like the following usually tear down the entire task, which is a bit heavy.

java.io.FileNotFoundException: /mnt/dfs/4/yarn/local/usercache/da_haitao/appcache/application_1568691584183_1953115/blockmgr-cc4689f5-eddd-4b99-8af4-4166a86ec30b/10/temp_shuffle_79be5049-d1d5-4a81-8e67-4ef236d3834f (No such file or directory)
	at java.io.FileOutputStream.open0(Native Method)
	at java.io.FileOutputStream.open(FileOutputStream.java:270)
	at java.io.FileOutputStream.<init>(FileOutputStream.java:213)
	at org.apache.spark.storage.DiskBlockObjectWriter.initialize(DiskBlockObjectWriter.scala:103)
	at org.apache.spark.storage.DiskBlockObjectWriter.open(DiskBlockObjectWriter.scala:116)
	at org.apache.spark.storage.DiskBlockObjectWriter.write(DiskBlockObjectWriter.scala:249)
	at org.apache.spark.shuffle.sort.ShuffleExternalSorter.writeSortedFile(ShuffleExternalSorter.java:209)
	at org.apache.spark.shuffle.sort.ShuffleExternalSorter.closeAndGetSpills(ShuffleExternalSorter.java:416)
	at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.closeAndWriteOutput(UnsafeShuffleWriter.java:230)
	at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:190)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
	at org.apache.spark.scheduler.Task.run(Task.scala:109)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
	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)

This change pre-touch the temporary file to check whether the parent directory is available or not. If NOT, we may try another possibly heathy disk util we reach the max attempts.

Why are the changes needed?

Re-running the whole task is much heavier than pick another heathy disk to output the temporary results.

Does this PR introduce any user-facing change?

NO

How was this patch tested?

ADD UT

[SparkQA]

Test build #111556 has finished for PR 25962 at commit ced8539.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #111562 has finished for PR 25962 at commit 1fc204f.

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

[SparkQA]

Test build #111584 has finished for PR 25962 at commit 7b080fb.

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

[yaooqinn]

cc @cloud-fan

[yaooqinn]

cc @squito

[cloud-fan]

where do we use it before?

[squito]

its used in Utils.createDirectory, which is only used in the standalone cluster

[yaooqinn]

It's used by DiskBlockManager.createLocalDirs https://github.com/apache/spark/pull/25962/files#diff-b25a153e05a044779df448f024c3a3afL141

[cloud-fan]

is it OK to leave the file created? This is different from before

[squito]

This is a really good question. It looks safe to me, I don't think any of the writers check for file existence before opening a FileOutputStream etc. I also checked IndexBlockResolver, which has some checks on pre-existing files -- but those are checks on the final destination files, not the temporary intermediate files.

[yaooqinn]

Yes, there is only one temp file be created or none, after or before this change

[squito]

I see what you're trying to do here, but does this really buy you much? If you have one bad disk, even if you can prevent temporary files from going to that disk, the final destination files still have a really high chance of going to that disk, don't they?

[squito]

its used in Utils.createDirectory, which is only used in the standalone cluster

[squito]

This is a really good question. It looks safe to me, I don't think any of the writers check for file existence before opening a FileOutputStream etc. I also checked IndexBlockResolver, which has some checks on pre-existing files -- but those are checks on the final destination files, not the temporary intermediate files.

[yaooqinn]

I see what you're trying to do here, but does this really buy you much? If you have one bad disk, even if you can prevent temporary files from going to that disk, the final destination files still have a really high chance of going to that disk, don't they?

@squito Yes, as the temp file name is random, it still has a chance to go to that bad disk. But with 10 times max retries, the probability can go very low. And it is worth preventing one task from failure and rescheduling after it has done all the calculation process right before the commit process, especially when the task is heavy, skewed...

In our 2000 nodes Hadoop cluster, which with 12 disks/node, this approach reduce the number of that exception a lot.

[cloud-fan]

@yaooqinn I think @squito was asking about the final file, which has a fixed name. How do you avoid putting the final file in bad disks?

[yaooqinn]

@cloud-fan @squito sorry for the irrelevant answer. Yes, the final files are not handled here because they have fixed names for each taskAttemt(since 3.0.0). If we fail with a final file exception, we need to reschedule the task by driver.

[cloud-fan]

This makes sense to me. Temp files are not only used by shuffle, but also external sorter, external hash map, etc. So avoiding putting temp files in bad disks does help. @squito what do you think?

[tgravescs]

In our 2000 nodes Hadoop cluster, which with 12 disks/node, this approach reduce the number of that exception a lot.

So the only time Hadoop should show you this bad disk is if yarn doesn't detect it or it if goes bad during the running of the container. YARN has a specific feature to detect bad disks and will not give that to the container if they are bad. So in your case are you executors very long running? Are you using the yarn feature?
I'm not necessarily against this idea as disks can go bad while executors are running but just want to check to see how much this is really happening. What happens when we go to rename/merge the temp file to final location? the shuffle file name is static so should hash to same dir every time unless we are adding different dir. I can't remember that code off the top of my head. With the external shuffle service, the application registers what directories its using such that the external shuffle service can use those to find the files again, I'm wondering if the temp ones might work but then fail later on the static names.

[squito]

OK, I can see how it can help a little to add this safety for temporary files. It doesn't really prevent task failure completely, but it'll help avoid it sometimes, anyway.

@tgravescs I don't think yarn's disk checker will help that much here -- as you've said, it only matters when starting a new container, and executors tend to survive past one disk failure. Blacklisting should still give you correctness even without the fix here -- but this is trying a simple heuristic which might work sometimes, even if it is not guaranteed to work.

[tgravescs]

Yes it depends on how often your executors are created/destroyed, if using dynamic allocation and a lot of long tail it could be cycling those fairly often and yarn disk checker should help, if not it won't. Lots of jobs it won't help by itself.

I mostly wonder how much this helps because it might help with the temp shuffle file, but then it might fail later creating real shuffle file. Is this ok, maybe, but it's potentially changing from failing fast to failing later. if there is a long time between those then you potentially taking longer. There are a lot of mights in that sentence and I don't have any concrete idea how much it will help or hurt.
Has this actually been run on real jobs and have you seen a benefit?

[yaooqinn]

So the only time Hadoop should show you this bad disk is if yarn doesn't detect it or it if goes bad during the running of the container. YARN has a specific feature to detect bad disks and will not give that to the container if they are bad. So in your case are you executors very long running? Are you using the yarn feature?

Yarn disk health check is on. Yes, it does help much for our whole 40k-50k Spark jobs daily. But it does not help much for those with long-lived (30min or longer) executors, which might be big ETL jobs.

What happens when we go to rename/merge the temp file to final location? the shuffle file name is static so should hash to same dir every time unless we are adding different dir.

When using multi disks, the temp file, and the final file more likely to pick different disks. If the final one picks the bad disk, the task attempt is doomed to failure. But if only the temp one picks, it is still worth saving it.

Yes it depends on how often your executors are created/destroyed, if using dynamic allocation and a lot of long tail it could be cycling those fairly often and yarn disk checker should help, if not it won't. Lots of jobs it won't help by itself.

Most of our spark jobs are dynamic allocation on. But executor recycling is not granular enough to handle disk failures, as the disk check of yarn is only just one periodic task.

Is this ok, maybe, but it's potentially changing from failing fast to failing later. if there is a long time between those then you potentially taking longer.

This can happen if and only if the temp and the final files pick the same disk. But comparing to reschedule an entire task, how long can success or fail to rename/move a file be?

Has this actually been run on real jobs and have you seen a benefit?

We have applied this to that cluster for more than 3 months. I have not performed very accurate statistics for such an exception. But before this, the users come to for help with this kind of failure once every 2 or 3 days on average. Since then, there is none.

[tgravescs]

thanks for the details, sounds like anecdotally this has helped. I'm ok with the idea of the change.

[cloud-fan]

It's possible that after 10 retries we still not able to find the healthy disk. Can we think of how to remove the flakiness of this test?

[yaooqinn]

add a precondition for this assert？like if tempShuffleFile2.parent belongs to rootDir1?

[SparkQA]

Test build #112985 has finished for PR 25962 at commit dce7bb8.

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

[yaooqinn]

@cloud-fan please check whether this is ok or not

[jiangxb1987]

LGTM

[yaooqinn]

@jiangxb1987 thanks a lot.

[yaooqinn]

@jiangxb1987 Can we have this be merged. It looks be approved by all the reviewers here. Thank you very much.

[cloud-fan]

thanks, merging to master!

[squito]

sorry I should have mentioned this earlier -- these two functions are exactly the same except for the constructor, right? Could have been refactored. (not a big deal, and not worth a followup just for this.)

[hvanhovell]

@yaooqinn this is causing a performance regression for short running queries that only write to a few shuffle files, because now every shuffle file is created eagerly. In this case I think the cure is worse than the disease, so I think we should revert this. cc @cloud-fan @tgravescs @squito

[tgravescs]

to clarify you are saying that many of your tasks don't actually create shuffle files so the fact they get created eagerly is causing overhead? Can you quantify how much?

[hvanhovell]

@tgravescs yeah I am. The #28072 quantifies how much the overhead can be.