Submit jobs to a Ray cluster

Method 1: 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)

   

Method 2: 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) 

   

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)

Method 3: 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: Submit a job that performs a 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.