Ray quick start

Prerequisites

A Ray cluster is created and running, and you have obtained the endpoint and token from the Endpoint Information page of the cluster. For more information, see Manage Ray clusters.

Choose a submission method

A Ray cluster supports the following three submission methods. Choose one based on your scenario.

Interactive development

Ideal for quick starts and debugging. You can connect to a Ray cluster directly from the Notebook environment in the EMR Serverless Spark console without additional network configuration. To connect from your local machine, ensure it has network connectivity to the Ray cluster and a Python 3.12 environment.

1. Install the Ray client:

   pip install ray[client]==2.47.1

2. On the invocation information page of the cluster, get the gRPC address and token.

3. Use the following sample code to connect to the Ray cluster for interactive development:

   import ray
   import os
   
   def get_metadata():
       headers = {"ray-token": "<your_token>"}
       return [(key.lower(), value) for key, value in headers.items()]
   
   ray.init(address="<your_gRPC_address>", _metadata=get_metadata())
   
   import time
   @ray.remote
   def square(x):
       time.sleep(0.1)
       return x * x
   
   futures = [square.remote(i) for i in range(10)]
   
   results = ray.get(futures)
   print("Square results:", results)

   After successfully connecting to the Ray cluster and running the compute task, the terminal displays output similar to the following:

   2025-12-22 17:32:04,514 INFO util.py:154 — Missing packages: ['ipywidgets']. Run `pip install -U ipywidgets`, then restart the notebook server for rich notebook output.
   2025-12-22 17:32:04,537 INFO client_builder.py:242 — Passing the following kwargs to ray.init() on the server: log_to_driver
   2025-12-22 17:32:05,842 WARNING utils.py:1565 — Python patch version mismatch: The cluster was started with:
       Ray: 2.47.1
       Python: 3.12.9
   This process on Ray Client was started with:
       Ray: 2.47.1
       Python: 3.12.12
   
   Square results: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

SDK submission

Suitable for embedding Ray job submissions in your application code. Your local environment must have Python 3.12 installed.

1. Install the Ray job submission client library:

   pip install "ray[default]==2.47.1"

2. On the invocation information page of the cluster, get the public endpoint and token.

3. Use the following sample code to connect to the Ray cluster and submit a job:

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<your_token>"
   }
   
   client = JobSubmissionClient("<your_public_endpoint>", headers=custom_headers)
   
   # Alternatively, you can use the internal network. This requires the client to be in the same region and VPC as the cluster. 
   # Submitting over the internal network is recommended for improved stability.
   # client = JobSubmissionClient("http://emr-spark-ray-gateway-cn-beijing-internal.spark.emr.aliyuncs.com", headers=custom_headers)
   
   job_id = client.submit_job(
       entrypoint="python -c 'print(\"Hello from Ray Client!\")'"
   )
   
   print(f"Submitted job with ID: {job_id}")
   
   while True:
       status = client.get_job_status(job_id)
       print(f"Job status: {status}")
       if status.is_terminal():
           break
       time.sleep(1)
   
   logs = client.get_job_logs(job_id)
   print("Job logs:")
   print(logs) 

   After running the script, the terminal displays output similar to the following:

   (.venv) xxx % python '/Users/xxx/xxx.py'
   Submitted job with ID: raysubmit_9Q5JTEgG8tqLEJb4
   Job status: PENDING
   Job status: PENDING
   Job status: SUCCEEDED
   Job logs:
   2026-04-03 17:01:15,126 INFO job_manager.py:531 -- Runtime env is setting up.
   Hello from Ray Client!

4. (Optional) To access files in a mounted directory from your job, mount a managed file directory when you create the Ray cluster. Then, reference the files using their mount path when you submit the job. The following example assumes you have mounted an OSS directory to the /mnt/myoss path:

   import time
   from ray.job_submission import JobSubmissionClient
   
   custom_headers = {
       "ray-token": "<your_token>"
   }
   
   client = JobSubmissionClient("<your_public_endpoint>", headers=custom_headers)
   
   # Use the mount path to directly reference a script file in OSS as the entrypoint.
   job_id = client.submit_job(
       entrypoint="python /mnt/myoss/main.py"
   )
   
   print(f"Submitted job with ID: {job_id}")
   
   while True:
       status = client.get_job_status(job_id)
       print(f"Job status: {status}")
       if status.is_terminal():
           break
       time.sleep(1)
   
   logs = client.get_job_logs(job_id)
   print("Job logs:")
   print(logs)

   Note

   In your Ray job code, you can also read from and write to files directly using their mount paths. For example, you can read from /mnt/myoss/test.txt or write a file to the /mnt/myoss/ directory. Files written to this directory are synchronized to the corresponding OSS path.

   The following example shows how to read from and write to files in a mounted directory within a Ray job:

   # The following code is submitted to the Ray cluster.
   
   import ray
   import time
   
   ray.init()
   
   @ray.remote
   def read_from_mount_path(i):
       with open('/mnt/myoss/test.txt', 'r', encoding='utf-8') as f:
           content = f.read()
       return content
   
   @ray.remote
   def write_to_mount_path(i):
       content = "Hello world " + str(i)
       with open('/mnt/myoss/output' + str(i) + '.txt', 'w', encoding='utf-8') as f:
           f.write(content)
       return 'Write succeeded'
   
   print("--- Starting remote tasks ---")
   start_time = time.time()
   obj_refs = [read_from_mount_path.remote(i) for i in range(4)]
   obj_refs2 = [write_to_mount_path.remote(i) for i in range(4)]
   results = ray.get(obj_refs)
   results2 = ray.get(obj_refs2)

Command-line submission

Ideal for integrating with external systems. Your local environment must have Python 3.12 installed.

1. Install the Ray job submission client library:

   pip install "ray[default]==2.47.1"

2. On the invocation information page of the cluster, get the public endpoint and token.

3. Run the following commands to submit a Ray job:

   # You must set the RAY_ADDRESS and RAY_JOB_HEADERS environment variables before you run `ray job submit`.
   export RAY_ADDRESS='<your_public_endpoint>'               
   export RAY_JOB_HEADERS='{"ray-token": "<your_token>"}'
   
   ## Submit the job
   ### Example: working-dir supports local paths & Ray core distributed sort.
   ray job submit --working-dir "." -- python  test-ray-core-sort.py
   
   ### Supports reading from and writing to OSS.
   ray job submit --working-dir "." -- python test-saving-data-test.py
   
   ### Supports reading from and writing to OSS-HDFS.
   ray job submit --working-dir "." -- python test-saving-data-oss-hdfs-test.py
   
   ### The code supports reading from and writing to mounted directories.
   ray job submit --working-dir "." -- python  test-saving-data-test-mount.py

For more command-line parameters, see the Ray job submission CLI official documentation.