Ray on spark implementation

[WeichenXu123]

Signed-off-by: Weichen Xu [email protected]

Why are these changes needed?

REP: ray-project/enhancements#14

Commands to run tests:

Prerequisite

• Only supports linux system e.g. ubuntu, does not support macOS for now.
• Spark version >= 3.3

Testing on local machine (Requires linux system)

1. Install ray dev version in editable mode.

first, merge latest master into ray-on-spark branch, then

cd ray-repo, then
Install latest ray dev version: pip install -U https://s3-us-west-2.amazonaws.com/ray-wheels/latest/ray-3.0.0.dev0-cp38-cp38-manylinux2014_x86_64.whl (for other python version see https://docs.ray.io/en/latest/ray-overview/installation.html#install-nightlies) , then run python python/ray/setup-dev.py -y , it will link ray source directory to your local ray repo code directory.

3. Run tests pytest python/ray/tests/spark/test_ray_on_spark.py -s

Testing on databricks runtime

Install my PR branch ray package on databricks notebook:
https://e2-dogfood.staging.cloud.databricks.com/?o=6051921418418893#notebook/2110795073425386/command/2110795073425387
Using my testing notebook:
https://e2-dogfood.staging.cloud.databricks.com/?o=6051921418418893#notebook/2110795073420548/command/2110795073420549

Debugging tips

Checking ray processes output logs is painful. These files are scattered on every spark cluster worker nodes. For easier testing, we can create a spark cluster with only one worker machine, then,
By default, we can check following local disk path:

"Ray start" script output:

/tmp/ray-logs-{head-port}-XXXX

Other ray processes log output:

/tmp/ray-temp-{head-port}-XXXX/session_latest/logs

There will be a warning message output like:

You can check ray head / worker starting script logs under local disk path /tmp/ray-logs-50233-070d, and you can check ray processes logs under local disk path /tmp/ray-temp-50233-070d/session_latest/logs.

But note the log path is local path on every spark cluster nodes. So, for non-driver nodes, you have to write a spark job to collect those files, like:

# Check files under the dir
def mapper(_):
  import os
  return os.listdir("/tmp/ray-temp-50233-070d/session_latest")

print(sc.parallelize([1], 1).map(mapper).collect()[0])

# read specific files.
def mapper(_):
  import os
  with open("/tmp/ray-temp-50233-070d/session_latest/logs/raylet.1.err") as f:
    return f.read()

print(sc.parallelize([1], 1).map(mapper).collect()[0])

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[WeichenXu123]

CC @jjyao Ready for first pass reviewing :)

[WeichenXu123]

Provide a param to control the ray_temp_dir path, this is useful when the default temp dir disk capacity is not sufficient.

[WeichenXu123]

@ericl @jjyao

Could you help add an option for ray start script to make ray node delete temp directory when ray node is killed ?

[ericl]

I don't think this is something we could offer except as best effort, which suggests it isn't the right mechanism for cleanup. Instead, why not set the temp dir to a known location and remove that in a wrapper script after the Ray worker exits?

[WeichenXu123]

@ericl Yes. Make sense. I use a wrapper script and it register a SIGTERM handler and in handler it deletes the temp dir.