dask / dask/dask-yarn

Dask Scheduler host/port Not Written to Skein Key-Value Storage When YARN Application Restarts

Open
#135 5 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
69
Forks
40
PR merge metrics
No merged PRs in 30d

Description

We are running a Dask cluster using Dask-Yarn on AWS EMR. I am trying to make it more robust so we're running the Scheduler in local mode. To make the system more robust we allow multiple attempts of the YARN Application using `skein.ApplicationSpec(..., max_attempts=2)`.

In the event that a node fails, and that node is running the ApplicationMaster, then the YARN application will make a second attempt - it will restart itself and all the workers. However, the Scheduler will *not* see these new workers. It will sit there happily waiting on workers and blocking our application flow, since there's pending work to be done, but no workers to do it on.

**The problem is that when the YARN Application retries after the ApplicationMaster has failed, the scheduler's address and port number are not written to Key/Value store and the new workers don't know where to connect**. This line (https://github.com/dask/dask-yarn/blob/master/dask_yarn/core.py#L558) appears to not re-run on the 2nd application attempt.

To verify this, on the 1st YARN application attempt you can navigate to `http://appmaster:20888/proxy/attempt_num/kv` and see the `dask.scheduler` and `dask.dashboard` urls listed in the KV storage. Then if you shutdown the node running the ApplicationMaster the application will start a 2nd attempt. Navigate to `http://new_appmaster:20888/proxy/second_attempt_num/kv` and you'll see that nothing is listed in the KV storage, and no workers in the Dask dashboard.

**Question(s)**:

1) Is there a workaround to this problem so that my YARN Application can be robust to failures and restart?
2) Is there a way for my Scheduler to recognize that it has no workers and eventually timeout?
3) Is there a way to use the `distributed.as_completed()` method with Futures that somehow timeout in the event that the Scheduler loses all its workers and they never come back?

**Environment**:

- Dask version: 2.24.0, Dask-yarn: 0.8.1
- Python version: 3.6
- Operating System: AWS Linux

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.