使用Ray集群

纳管文件高级配置示例

• 挂载时对读写一致要求高，使用工作空间角色鉴权：

  {
    "fs.jindo.args": "-oattr_timeout=0 -oentry_timeout=0 -onegative_timeout=0 -oauto_cache -ono_symlink"
  }

• 挂载时对读写一致要求高，且使用自定义AccessKey鉴权：

  {
    "authType": "AK",
    "accessKeyId": "LTxxxxxxxxx",
    "accessKeySecret": "xxxxxxxxxxxxxx",
    "fs.jindo.args": "-oattr_timeout=0 -oentry_timeout=0 -onegative_timeout=0 -oauto_cache -ono_symlink"
  }

高级配置示例

以下示例配置了启动时下载OSS文件和初始化Python环境：

{
  "userDefinedFiles": "oss://mybucket/hello.py,oss://mybucket2/test/test.jar",
  "userRequirementsFile": "oss://mybucket/env/requirements.txt"
}

方式一：代码提交（批任务）

本地环境要求：Python 3.12。

1. 安装Ray Job Submission客户端库：

   pip install "ray[default]==2.47.1"

2. 在集群的调用信息页面，获取公网调用地址和Token。

3. 使用以下示例代码连接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) 

   

方式二：命令行提交（批任务）

本地环境要求：Python 3.12。

1. 安装Ray Job Submission客户端库：

   pip install "ray[default]==2.47.1"

2. 在集群的调用信息页面，获取公网调用地址和Token。

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

   # ray job submit前必须设置ray address和headers
   export RAY_ADDRESS='<公网调用地址>'               
   export RAY_JOB_HEADERS='{"ray-token": "<yourToken>"}'
   
   ## 提交任务
   ### working dir 支持本地 & Ray core distrubuted 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官方文档。

方式三：交互式开发

与Ray集群相同地域的VPC内访问，且环境为Python 3.12。

1. 安装Ray客户端：

   pip install ray[client]==2.47.1

2. 在集群的调用信息页面，获取gRPC 地址和Token。

3. 使用以下示例代码连接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)

   

Kerberos身份认证

EMR Serverless Ray支持以Kerberos身份访问OSS和OSS-HDFS。

读写DLF（Paimon表）

EMR Serverless Ray支持通过PyPaimon读写DLF中的Paimon表。

以下示例代码演示如何通过Ray读写DLF中的Paimon表：

import ray
import pandas as pd

data = [
    (10, 'pic10.jpg'),
    (11, 'pic11.png'),
    (12, 'pic12.gif')
]

# 转换为DataFrame
df = pd.DataFrame(data, columns=['id', 'pic'])
ds = ray.data.from_pandas(df)

# 写入Paimon表（仅支持VPC网络）
ds.write_paimon(
    table_identifier="default.my_table",
    catalog_kwargs={
        'metastore': 'rest',
        'warehouse': '<DLF Catalogs名称>',
        'uri': 'http://<region>-vpc.dlf.aliyuncs.com',
        'dlf.region': '<region>'
    }
)

# 读取Paimon表
ds1 = ray.data.read_paimon(
    table_identifier="default.my_table",
    catalog_kwargs={
        'metastore': 'rest',
        'warehouse': '<DLF Catalogs名称>',
        'uri': 'http://<region>-vpc.dlf.aliyuncs.com',
        'dlf.region': '<region>'
    }
)

records = ds1.take_all()
for i, record in enumerate(records, 1):
    print(f"记录 {i}: {record}")

如果遇到权限问题，请参见DLF无权限问题排查。

读写Paimon File System

以下示例代码演示如何通过Ray读取OSS上的Paimon表：

import ray

# 读取Paimon表
ds = ray.data.read_paimon(
    table_identifier="default.calculate_tbl",
    catalog_kwargs={
        "warehouse": "oss://<bucket>/paimon_warehouse",
        "fs.oss.endpoint": "oss-<region>.aliyuncs.com",
        "fs.oss.region": "<region>"
    }
)

records = ds.take_all()
for i, record in enumerate(records, 1):
    print(f"记录 {i}: {record}")