本文介绍向EMR Serverless Spark的Ray集群提交任务的三种方式：交互式开发、SDK提交和命令行提交，并给出各方式的适用场景与示例代码。-开源大数据平台 E-MapReduce(EMR)-阿里云帮助中心

前提条件

已创建并启动Ray集群，且已在集群的调用信息页面获取提交任务所需的地址和Token。详情请参见Ray集群管理。

选择提交方式

Ray集群支持以下三种任务提交方式，请根据使用场景选择。

提交方式	适用场景	运行环境	连接地址
交互式开发	快速入门和调试，可逐段执行代码并即时查看结果。	工作空间内的Notebook无需额外配置；从本地连接需Python 3.12并安装`ray[client]`。	gRPC地址
SDK提交	将Ray作业嵌入工程代码中提交，适合由应用程序触发的批任务。	Python 3.12，安装`ray[default]`。	公网或内网调用地址
命令行提交	与外部调度系统集成，或以脚本方式提交批任务。	Python 3.12，安装`ray[default]`。	公网调用地址

说明

SDK提交和命令行提交均以Ray Job形式运行批任务，两者的差异仅在于触发入口。

交互式开发

适合快速入门和调试。在EMR Serverless Spark控制台的Notebook环境中直接连接Ray集群（无需额外配置网络）。如从本地连接，需确保本地机器与Ray集群之间网络互通，且环境为Python 3.12。

安装Ray客户端：
```
pip install ray[client]==2.47.1
```
在集群的调用信息页面，获取gRPC 地址和Token。

使用以下示例代码连接Ray集群，进行交互式开发：

import ray
import os

def get_metadata():
    headers = {"ray-token": "<yourToken>"}
    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("平方结果:", results)

成功连接 Ray 集群并执行计算任务后，终端输出类似信息：

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

平方结果: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

SDK提交

适合将Ray作业嵌入工程代码中提交。本地环境要求：Python 3.12。

安装Ray Job Submission客户端库：
```
pip install "ray[default]==2.47.1"
```
在集群的调用信息页面，获取公网调用地址和Token。

使用以下示例代码连接Ray集群并提交作业：

import time
from ray.job_submission import JobSubmissionClient

custom_headers = {
    "ray-token": "<yourToken>"
}

client = JobSubmissionClient("<公网调用地址>", headers=custom_headers)

# 或者用内网，需要保证提交端在同一region vpc下，建议内网提交，更加稳定
# 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)

(.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!

（可选）如需在作业中访问挂载目录中的文件，可在创建Ray集群时挂载纳管文件目录，然后在提交作业时通过挂载路径引用文件。以下示例假设已将OSS目录挂载到/mnt/myoss路径下：

import time
from ray.job_submission import JobSubmissionClient

custom_headers = {
    "ray-token": "<yourToken>"
}

client = JobSubmissionClient("<公网调用地址>", headers=custom_headers)

# 通过挂载路径直接引用OSS中的脚本文件作为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)

说明

在Ray任务代码中，也可以直接通过挂载路径读写文件。例如，读取/mnt/myoss/test.txt或写入文件到/mnt/myoss/目录下，写入的文件会同步到对应的OSS路径中。

以下示例展示如何在Ray任务中读写挂载目录中的文件：

# 下面是提交到ray集群的代码

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 '写入成功'

print("--- 开始运行远程任务 ---")
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)

命令行提交

适合与外部系统集成。本地环境要求：Python 3.12。

安装Ray Job Submission客户端库：
```
pip install "ray[default]==2.47.1"
```
在集群的调用信息页面，获取公网调用地址和Token。

执行以下命令提交Ray作业：

# ray job submit前必须设置ray address和headers
export RAY_ADDRESS='<公网调用地址>'               
export RAY_JOB_HEADERS='{"ray-token": "<yourToken>"}'

## 提交任务
### working dir 支持本地 & Ray core distributed sort
ray job submit --working-dir "." -- python  test-ray-core-sort.py

### 支持OSS读写
ray job submit --working-dir "." -- python test-saving-data-test.py

### 支持OSS-HDFS读写
ray job submit --working-dir "." -- python test-saving-data-oss-hdfs-test.py

### 代码支持Mount读写
ray job submit --working-dir "." -- python  test-saving-data-test-mount.py

更多命令行参数，请参见Ray Job Submission CLI官方文档。