基于RayJob Data的数据处理

操作场景

本文以通过创建计算资源池、创建RayJob类型的端点、连接数据、使用API提交作业为例，演示通过AI DataLake分析数据的操作指导。

操作流程

开始使用如下样例前，请务必按准备工作指导完成必要操作。

规划并创建OBS桶：创建一个用于存储作业脚本的OBS桶。
创建工作空间：创建一个工作空间并关联LakeFormation实例。
创建计算资源池：创建运行作业的计算资源。
创建运行作业的RayJob端点：创建运行作业的RayJob端点，并关联计算资源池（按需弹性模式除外）以获得运行作业的资源。
使用API提交作业：使用API提交作业至RayJob端点运行，本示例以提交一个简单文本处理的AI DataLake作业为例进行演示。

准备工作

已注册账号并实名认证，且账号不能处于欠费或冻结状态。
已开通AI DataLake服务并授权使用云服务资源。
已开通OBS权限并进行了委托确认。
已经获取IAM Token，详细操作请参见获取IAM用户Token。

步骤一：规划并创建OBS桶

AI DataLake Ray通过OBS服务实现作业提交与数据存储，需要先在OBS控制台进行桶及文件夹创建，并导入样例数据或作业脚本。

登录华为云管理控制台。
在页面左上角单击，选择“存储 > 对象存储服务 OBS”，进入对象存储服务页面。
选择“桶列表 > 创建桶”，进入创建桶页面，配置相关参数后单击“立即创建”。
- 桶名称：根据界面要求设置桶名称。
- 其他参数根据实际情况选择。
在桶列表页面，单击已创建的桶名称。

将作业脚本压缩为.zip文件，然后在“对象”页签单击“上传对象”，上传压缩包至所创建的OBS桶中。

例如文本处理脚本“ray-job-code.py”，内容示例如下：

import ray
import random
import time
import sys

# 连接Ray集群
ray.init()

@ray.remote
def compute_points_in_circle(num_points: int) -> int:
    inside = 0
    for _ in range(num_points):
        x, y = random.random(), random.random()
        if x*x + y*y <= 1:
            inside += 1
    return inside

def estimate_pi(total_points: int, num_workers: int = 4) -> float:
    start = time.time()
    points_per_worker = total_points // num_workers
    # 分布式执行任务
    tasks = [compute_points_in_circle.remote(points_per_worker) for _ in range(num_workers)]
    total_inside = sum(ray.get(tasks))
    pi = 4.0 * total_inside / total_points
    print(f"估算的圆周率: {pi:.6f}, 耗时: {time.time() - start:.2f}秒")
    return pi

if __name__ == "__main__":
    estimate_pi(total_points=int(sys.argv[1]), num_workers=1)
    ray.shutdown()

步骤二：创建工作空间

登录AI DataLake管理控制台。
在左侧导航栏中，单击工作空间区域。
在下拉列表中选择“创建工作空间”。

配置工作空间的相关参数。

表1 工作空间的基本信息
类型	参数	示例	说明
基础信息	工作空间名称	workspace_raytest	工作空间的具体名称，同一个账号下的工作空间不可重名。名称只能包含字母、数字、中划线、下划线。名称字符长度为4~32位。
	描述	智能驾驶数据分析	对工作空间的简要描述，最长不超过255个字符。
	企业项目	default	如果所建工作空间属于企业项目，可选择对应的企业项目。企业项目是一种云资源管理方式，企业项目管理服务提供统一的云资源按项目管理，以及项目内的资源管理、成员管理。关于如何设置企业项目请参考《企业管理用户指南》。说明：只有开通了企业管理服务的用户才显示该参数。
多模数据管理	LakeFormation实例	lakeformation_for_raytest	选择当前工作空间关联的LakeFormation实例，每个工作空间需关联1个LakeFormation实例，空间创建后已关联的实例不支持修改。选择已创建的LakeFormation实例，用于统一管理元数据和数据访问控制。如无可用实例，请先创建LakeFormation实例，具体操作请参考创建LakeFormation实例。

确认所有配置信息无误，单击“立即创建”，完成工作空间创建。
工作空间创建成功后，您可以在工作空间管理的列表中查看新创建的工作空间。

步骤三：创建计算资源池

在AI DataLake管理控制台页面，切换页面左上角的工作空间为步骤二：创建工作空间新创建的空间。
单击左侧导航栏的“资源管理”，选择“计算资源池”，单击右上方“购买”，进入购买计算资源池页面。

在“购买计算资源池”界面，填写具体参数。

表2 购买计算资源池参数说明
参数	示例	说明
计费模式	按需计费	包年/包月：预付费模式，按订单的购买周期计费。在购买周期内资源独享，空闲（无作业运行）时不会释放，使用体验更佳，价格比按需计费模式更优惠。适用于可预估资源使用周期的场景，例如已完成开发进入生产阶段的项目，推荐使用包年包月计费模式。按需计费：即后付费模式，按实际使用量计费，在购买周期内资源独享，空闲时资源不被释放。
资源池名称	resource-pool-for-raytest	计算资源池的具体名称。名称只能包含数字、小写英文字母和中划线，且只能以字母开头、以字母或数字结尾。输入长度不能超过63个字符。
CPU资源	本例配置8个通用计算标准型实例。	CPU是通用型计算资源，适用于各种类型的计算任务。 CPU资源的特点：通用性强，适合各种类型的计算任务。相比于GPU和NPU的成本更低。擅长处理顺序执行的任务。 CPU资源的适用场景：ETL数据抽取、转换、加载；轻量计算场景，例如小规模数据处理、脚本运行，日志分析场景，例如日志采集、解析、统计分析。 CPU资源的具体规格请参考产品规格
GPU资源	本例不购买GPU实例	GPU是图形处理器适用于图形渲染和大规模并行计算场景，适合深度学习训练和科学计算。GPU拥有成百上千个计算核心，可以同时处理大量简单计算任务。 GPU资源的特点：支持并行计算，适用于批处理作业场景，数千个核心同时计算，处理效率高。天然适合深度学习、神经网络、AI计算场景。 GPU资源的适用场景：深度学习，例如神经网络训练、模型调优场景；图形处理场景，例如图像识别、目标检测；视频处理场景，例如视频分析、转码、视频渲染场景，等其他AI科学计算场景。 GPU资源的具体规格请参考产品规格
NPU资源	本例不购买NPU实例	NPU适用于AI计算场景。NPU资源采用架构优化和指令集，专门加速AI推理任务，具有高能效比的特点。 NPU资源的特点：AI计算场景专用，具备高性能、低延迟、推理成本更优的特点。 NPU资源的适用场景：AI计算场景、图像识别场景，推荐系统等AI计算设计场景。 NPU资源的具体规格请参考产品规格
AI DataLake网络	自定义	选择AI DataLake资源池所属网络，该网络基于虚拟私有云（VPC）进行封装。如果不存在可使用的网络，也可单击“创建网络”进行创建。一个工作空间仅支持创建一个网络。

参数填写完成后，单击“立即购买”，在界面上确认当前配置是否正确。
单击“提交”完成创建。等待资源池状态变成“可用”表示当前资源池创建成功。

步骤四：创建运行作业的RayJob端点

在AI DataLake管理控制台页面，切换页面左上角的工作空间为步骤二：创建工作空间新创建的空间。
在左侧导航栏选择“引擎端点 > AI 计算引擎 Ray”进入Ray端点列表页面。

单击页面右上角的“创建端点”，配置以下参数并单击“立即创建”。

基础信息配置。

表3 基础信息配置说明
参数	示例	参数说明
端点类型	RayJob	选择端点类型为“RayJob”。 RayJob端点提供按需启动、用完即销毁的“任务执行单元”，任务完成后自动释放资源，节省资源成本，但每次提交任务需要启动集群，存在一定的启动延迟。适用于一次性运行、自动清理资源的场景。
端点名称	ray-job-owin	输入端点名称，是端点的唯一标识符，不可与已存在的端点重名，且创建后不支持修改。名称只能包含小写英文字母、数字、中划线，且只能以英文字母开头，以英文字母或数字结尾。输入长度不能超过63个字符。
端点显示名称	ray-job-owin	输入端点显示名称，创建后可修改，用于用户界面展示，方便用户识别和记忆。名称只能包含中文、英文、数字、中划线。输入长度不能超过63个字符。

资源配置。

表4 资源配置说明
参数	示例	参数说明
资源使用模式	预留资源	选择资源使用模式。预留资源：使用预留资源池，资源独享。适合负载稳定、持续运行的业务场景，如生产项目、关键任务。混合模式：基线保障，自动扩容。优先消耗预留资源，高峰期自动触发弹性补位。适用于有规律波动的业务。按需弹性：按任务实际运行时长扣费，无任务不产生费用。适合短期或偶发性的业务需求，如开发测试或临时任务。
选择计算资源池	resource-pool-for-raytest	在下拉框中选择已创建的资源池。如果下拉框中没有可选的资源池，可以单击“购买计算资源”进行购买。具体操作，请参见购买预留资源池。

规格配置。

表5 规格配置说明
参数	参数说明
规格配置	选择对应的资源规格并配置资源数量。资源规格：在下拉框中选择需要配置的资源规格，支持选择CPU、GPU、NPU资源。资源配额 (Min-Max)：配置资源池分配给当前端点的最低可用资源和资源上限，Min值默认置灰，不能配置。Min值需小于等于Max值，且需大于0。您可以单击“添加规格”添加多个规格配置，最多支持添加10个，也可以单击“删除”删除配置，至少需保留一个规格配置。说明：仅资源使用模式选择“预留资源”和“混合模式”时，配置此参数。规格配置后，即可在页面右侧的资源配额区域查看端点关联资源池后配置的资源规格及Min和Max资源数量。

委托配置。

表6 委托配置说明
参数	示例	参数说明
委托配置	ai_datalake_admin_trust	委托给该端点的权限，会在运行时自动换取凭证并刷新至Ray集群，您可以在Ray集群中使用该委托凭证调用对应服务。通过下拉菜单选择已有的委托。创建后不支持修改，若未配置，可单击“新建委托”进行创建。具体操作请参考创建自定义委托。

端点创建后，可在列表中查看相关信息，当端点状态变为“运行中”，即可提交作业到该端点中运行。

步骤五：使用API提交作业

本示例以通过Postman等工具，调用运行作业API，提交一个简单文本处理的RayJob作业为例进行演示。

查看镜像版本。
1. 在AI DataLake管理控制台页面的左侧导航栏，选择“资产管理 > 镜像”。
2. 在“预置镜像”或“自定义镜像”页签下，单击镜像名称。
3. 查看预置镜像或者自定义镜像的版本。
  - 查看预置镜像版本：单击“版本页签”，在“镜像版本”列，查看预置镜像的版本。
  - 查看自定义镜像版本：在“镜像版本”列，查看自定义镜像的版本。
    如果没有自定义镜像，请手动注册。具体操作，请参见注册自定义镜像。

使用Postman等工具，调用运行作业API提交作业。

调用API的具体操作，请参见如何调用API。
运行作业API的参数说明，请参见运行作业。
image_id的值，请通过调用查看自定义镜像列表接口查询获取。

使用API提交作业请求示例如下：

POST https://{endpoint}/v2/workspaces/{workspace_id}/ray-jobs
{
    "name": "rayjob0615-4",     #作业名称
    "description": "description",     #作业描述信息
    "config": {
        "runtime_env":{
            "working_dir":"obs://ray-shared-csc/ray-job-code.zip",  #需替换为作业脚本存储的obs路径，且必须为.zip文件
        },
        "entrypoint": "python3 -m estimate_pi 1000000",
        "image": {
            "image_id": "dd877a59-5cbd-44c5-b8a7-dc464cca****",     #需替换为实际镜像ID
            "image_version": "01ae8ee6-af8a-48a8-895b-8d4a70bf****"   #需替换为实际镜像版本
        },
        "resource_config": {
            "head_resource_spec": {
                "cpu": "1020",
                "memory": "2048",
                "worker_gpu_amount": 0,
                "worker_npu_amount": 0
            },
            "worker_resource_spec": [
                {
                    "cpu": "1000",
                    "memory": "2049",
                    "worker_gpu_amount": 0,
                    "worker_npu_amount": 0,
                    "worker_replicas": 1,
                    "enable_autoscaling": true,
                    "worker_min_replicas": 1,
                    "worker_max_replicas": 1
                }
            ]
        },
        "rayjob_strategy": {
            "queued_timeout": 100000,
            "running_timeout": 100000
        }
    },
    "endpoint_name": "ray-job-owin"    #需替换为实际端点名称
}