Documentation for how to use Ray on Yarn

[robertnishihara]

We should provide instructions for how to use Ray on a Yarn cluster.

@imzhenyu, you have successfully deployed Ray on a Yarn cluster, right? What changes to Ray were required? Do you have any instructions for how to do this?

[imzhenyu]

Yes, our Ray cluster runs on Yarn. There are indeed some changes made to make it possible (esp. when we also want to have FO). I'll ask people in the team to provide more details later.

[imzhenyu]

Just discussed with the team, we may probably come up two steps for this: (1). open source ray-app-master on yarn (w/o FO); (2). change code in ray and ray-app-master to enable FO. What do you think?

[robertnishihara]

@imzhenyu that sounds good. Is "ray-app-master on yarn" a repository? Or a set of config files? What kind of codes changes are needed? Thanks!

[imzhenyu]

@robertnishihara there are not many files, so we can probably put it under ray/yarn folder, or ray/devops/yarn folder, whichever you like. The code change is mostly like assigning a pre-defined ID to each local scheduler (as db client ID) due to the some of the FO changes in Ant. But after a short discussion with team members yesterday, I reallize that it may not be inevitable so we can omit that. There are still some FO logic in ray-on-yarn though to have nodes restarted after failure using Yarn's APIs.

[robertnishihara]

Ok, that sounds good. Thanks!

[jcrist]

Apologies for the developer evangelism/long-post below, I think you may find work I've done recently useful

---

An alternative to adding custom Java code would be to use Skein, a tool/library I wrote to make deploying applications on YARN easier. Skein is what Dask uses for its YARN deployments. It uses a declarative specification (either written as a yaml file, or using the python api), and handles all the YARN stuff without the user ever touching java.

I looked at writing a proof-of-concept for deploying Ray on YARN using Skein, but ran into a few issues (more on that later). So the following is pseudo-code, but hopefully the logic is still clear:

name: ray

services:
    ray-head:
        instances: 1
        resources:
            # The resources for the head node.
            vcores: 2
            memory: 2048
        files:
            # A packaged python environment using `conda-pack`. Note that Skein
            # doesn't require any specific way of distributing files, but this
            # is a good one for python projects.
            environment: environment.tar.gz
        commands:
            # Activate the packaged conda environment
            - source environment/bin/activate
            # This command would start all the processes needed on the ray head
            # node, and register any addresses needed by the workers with the
            # Skein key-value store. It should then block until killed with sigterm
            # (at which point all started processes should also die).
            - start-ray-head

    ray-worker:
        # The number of instances to start initially. This can be scaled
        # dynamically later.
        instances: 4
        resources:
            # The resources for the worker node
            vcores: 2
            memory: 2048
        files:
            environment: environment.tar.gz
        depends:
            # Don't start any worker nodes until the head node is started
            - ray-head
        commands:
            # Activate the packaged conda environment
            - source environment/bin/activate
            # This command would start all the processes needed on a ray worker
            # node, getting any addresses it needs for e.g. the head node from
            # the skein key-value store. It should then block until killed with
            # sigterm (at which point all started processes should also die).
            - start-ray-worker

A few key points:

• The above specification describes two services: ray-head and ray-worker. Both are composed of what files are needed to run them, and what commands are needed to start each service. The full specification format is here

• In the above pseudocode, conda-pack is used to distribute a python environment containing Ray, its dependencies, and any other libraries the user might want on each node. The tarball works transparently with YARN's resource localization mechanism, and allows the user to distribute the necessary resources without having IT deploy stuff manually. Note that skein is agnostic to this, so files could already be available locally, or distributed using any other mechanism. For more information see https://jcrist.github.io/skein/distributing-files.html.

• The commands to start each service are pseudocode (I couldn't figure out how to write them, but someone familiar with Ray could easily). start-ray-head needs to start all processes needed for the ray head node and register any runtime configuration details in the skein key-value store. This would be things like the redis address. start-ray-worker can then get this address from the key-value store and use it as needed. Both commands then need to block indefinitely until shutdown, and if killed with SIGTERM shutdown all started processes.

  This is the main problem I ran into when trying to write a proof-of-concept. The ray start --block command doesn't kill subprocesses when killed. When YARN stops a container it sends SIGTERM (and later SIGKILL if that doesn't work). Killing the top-level process should result in cleanup of all child processes, which currently doesn't seem to be the case (kill -15 $PID_OF_RAY_START_BLOCK doesn't cleanup child processes).

Given the above specification, starting, scaling, and stopping the cluster using the skein cli is as simple as:

# Start the application, and record the application id
$ appid=`skein application start spec.yaml`

# Scale up to 10 workers
$ skein container scale $appid --service ray-worker --number 10

# Shutdown the applicaiton
$ skein application shutdown $appid

Alternatively, this is all doable using the native python api, which is what dask-yarn uses.

---

Skein is tested on clusters with and without kerberos enabled (as well as a slew of other configurations), eliminating a class of bugs you'd need to test for if you wrote your own application master. It also significantly shortens the amount of code you need to support yarn (dask-yarn is ~500 lines, including docstrings).

There are arguments against using it (it's new-ish, an added abstraction layer, etc...), but I truly believe for non-jvm projects it's at least a good option to think about when looking at YARN support.

I'd be happy to help someone work on adding YARN support to Ray using Skein, but I don't currently have bandwidth to work on this personally. Just thought I'd let you know another option exists.

[robertnishihara]

@jcrist thanks for the note! I agree that it makes sense for ray start --block to kill its child processes when killed. Did you run into other issues or is that the only one?

[jovany-wang]

@jcrist
Sorry that I know a little about Skein.
Does this a popular solution for making deploying apps on yarn?

[raulchen]

@jcrist thank you for providing this info. It definitely great if deployment on yarn can be language-independent.
cc @mylinyuzhi

[jcrist]

Did you run into other issues or is that the only one?

That was the main one. I also had trouble building Ray (afaict there's no way to build Ray and not also build arrow?), but that's more my issue than yours.

I've opened issue #2615 for the --block behavior. If this is fixed I should be able to work up a quick script to demonstrate deploying on YARN using skein.

Does this a popular solution for making deploying apps on yarn?

Not popular (it's very new), but we've been happy with it for Dask and it may have some use for others.

[SongGuyang]

@jcrist Hi, I have one question.
How can I set some dynamic args or env when start ray process using skein. For example, I want to set the logging directory like this:

ray start --head --logging_directory ApplicationConstants.LOG_DIR_EXPANSION_VAR/ray.log

In this cmd, "ApplicationConstants.LOG_DIR_EXPANSION_VAR" is dynamic variable which is got from yarn in appMaster.

[jcrist]

ApplicationConstants.LOG_DIR_EXPANSION_VAR is just the string "<LOG_DIR>" (all the dynamic environment variables are just strings that are replaced at runtime. See here). You can write this as:

ray start --head --logging_directory <LOG_DIR>/ray.log