# 基于RayCluster Data的数据处理 #### 操作场景 AI DataLake是华为云提供的全托管的湖仓一体大数据分析服务，提供端到端的数据管理与分析能力，支持多种引擎的作业开发，满足多样化的大数据处理需求。 本文以通过创建计算资源池、配置RayCluster类型的端点、连接数据、使用API提交作业为例，演示通过AI DataLake分析数据的操作指导。 #### 操作流程 开始使用如下样例前，请务必按[准备工作]指导完成必要操作。 1. [规划并创建OBS桶]：创建一个用于存储作业脚本的OBS桶。 2. [创建工作空间]：创建一个工作空间。 3. [创建计算资源池]：创建运行作业的计算资源。 4. [创建运行作业的Ray端点]：创建运行作业的Ray端点，并关联资源空间以获得运行作业的资源。 5. [使用API提交作业]：使用API封装rest请求提交AI DataLake作业至Ray端点运行，本入门以提交一个简单文本处理的AI DataLake作业为例进行演示。 #### 准备工作 - 已注册账号并实名认证，且账号不能处于欠费或冻结状态。 - 已开通AI DataLake服务并授权使用云服务资源。 - 已开通OBS权限并进行了委托确认。 - 已构建镜像并上传至SWR中，并将镜像注册至AI DataLake。 - 已经获取IAM Token，详细操作请参见[获取IAM用户Token](https://support.huaweicloud.com/api-iam/iam_30_0001.html)。 #### 步骤一：规划并创建OBS桶 AI DataLake Ray通过OBS服务实现作业提交与数据存储，需要先在OBS控制台进行桶及文件夹创建，并导入样例数据或作业脚本。 1. 登录[华为云OBS管理控制台](https://console.huaweicloud.com/obs/)。 2. 选择"桶列表 \> 创建桶"，进入创建桶页面，配置相关参数后单击"立即创建"。 - 桶名称：根据界面要求设置桶名称。 - 其他参数根据实际情况选择。 3. 在桶列表页面，单击已创建的桶名称。 4. 将作业脚本压缩为.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() ``` #### 步骤二：创建工作空间 1. 登录[AI DataLake管理控制台](https://console.huaweicloud.com/fabric/)。 2. 在左侧导航栏中，单击工作空间区域。 3. 在下拉列表中选择"创建工作空间"。 4. 配置工作空间的相关参数。 表1工作空间的基本信息 | 类型 | 参数 | 示例 | 说明 | |:---|:---|:---|:---| | 基础信息 | 工作空间名称 | workspace_raytest | 工作空间的具体名称，同一个账号下的工作空间不可重名。 - 名称只能包含字母、数字、中划线、下划线。 - 名称字符长度为4\~32位。 | | 基础信息 | 描述 | 智能驾驶数据分析 | 对工作空间的简要描述，最长不超过255个字符。 | | 基础信息 | 企业项目 | default | 如果所建工作空间属于企业项目，可选择对应的企业项目。 企业项目是一种云资源管理方式，企业项目管理服务提供统一的云资源按项目管理，以及项目内的资源管理、成员管理。 关于如何设置企业项目请参考《[企业管理用户指南](https://support.huaweicloud.com/usermanual-em/zh-cn_topic_0108763975.html)》。 说明： 只有开通了企业管理服务的用户才显示该参数。 | | 多模数据管理 | LakeFormation实例 | lakeformation_for_raytest | 选择当前工作空间关联的LakeFormation实例，每个工作空间需关联1个LakeFormation实例，空间创建后已关联的实例不支持修改。 - 选择已创建的LakeFormation实例，用于统一管理元数据和数据访问控制。 - 如无可用实例，请先创建LakeFormation实例，具体操作请参考[创建LakeFormation实例](https://support.huaweicloud.com/usermanual-lakeformation/lakeformation_03_00007.html)。 | 5. 确认所有配置信息无误，单击"立即创建"，完成工作空间创建。 工作空间创建成功后，您可以在工作空间管理的列表中查看新创建的工作空间。 #### 步骤三：创建计算资源池 1. 在AI DataLake管理控制台页面，切换页面左上角的工作空间为[步骤二：创建工作空间]新创建的空间。 2. 单击左侧导航栏的"资源管理"，选择"计算资源池"，单击右上方"购买"，进入购买计算资源池页面。 3. 在"购买计算资源池"界面，填写具体参数。 表2购买计算资源池参数说明 | 参数 | 示例 | 说明 | |:---|:---|:---| | 计费模式 | 按需计费 | - **包年/包月：** 预付费模式，按订单的购买周期计费。在购买周期内资源独享，空闲（无作业运行）时不会释放，使用体验更佳，价格比按需计费模式更优惠。 适用于可预估资源使用周期的场景，例如已完成开发进入生产阶段的项目，推荐使用包年/包月计费模式。 - **按需计费**：即后付费模式，按实际使用量计费，在购买周期内资源独享，空闲时资源不被释放。 | | 资源池名称 | resource-pool-for-raytest | 计算资源池的具体名称。 - 名称只能包含数字、小写英文字母和中划线，且只能以字母开头、以字母或数字结尾。 - 输入长度不能超过63个字符。 | | CPU资源 | 本例配置8个通用计算标准型实例。 | CPU是通用型计算资源，适用于各种类型的计算任务。 - CPU资源的特点：通用性强，适合各种类型的计算任务。相比于GPU和NPU的成本更低。擅长处理顺序执行的任务。 - CPU资源的适用场景：ETL数据抽取、转换、加载；轻量计算场景，例如小规模数据处理、脚本运行，日志分析场景，例如日志采集、解析、统计分析。 - CPU资源的具体规格请参考[产品规格](https://support.huaweicloud.com/productdesc-aidatalake/aidatalake_08_0011.html) | | GPU资源 | 本例不购买GPU实例 | GPU是图形处理器，适用于图形渲染和大规模并行计算场景，适合深度学习训练和科学计算。GPU拥有成百上千个计算核心，可以同时处理大量简单计算任务。 - GPU资源的特点：支持并行计算，适用于批处理作业场景，数千个核心同时计算，处理效率高。天然适合深度学习、神经网络、AI计算场景。 - GPU资源的适用场景：深度学习，例如神经网络训练、模型调优场景；图形处理场景，例如图像识别、目标检测；视频处理场景，例如视频分析、转码、视频渲染场景，等其他AI科学计算场景。 - GPU资源的具体规格请参考[产品规格](https://support.huaweicloud.com/productdesc-aidatalake/aidatalake_08_0011.html) | | NPU资源 | 本例不购买NPU实例 | NPU适用于AI计算场景。NPU资源采用架构优化和指令集，专门加速AI推理任务，具有高能效比的特点。 - NPU资源的特点：AI计算场景专用，具备高性能、低延迟、推理成本更优的特点。 - NPU资源的适用场景：AI计算场景、图像识别场景，推荐系统等AI计算相关场景。 - NPU资源的具体规格请参考[产品规格](https://support.huaweicloud.com/productdesc-aidatalake/aidatalake_08_0011.html) | | AI DataLake网络 | 自定义 | 选择AI DataLake资源池所属网络，该网络基于虚拟私有云（VPC）进行封装。如果不存在可使用的网络，也可单击"创建网络"进行创建。 **一个工作空间仅支持创建一个网络。** | 4. 参数填写完成后，单击"立即购买"，在界面上确认当前配置是否正确。 5. 单击"提交"完成创建。等待资源池状态变成"可用"表示当前资源池创建成功。 #### 步骤四：创建运行作业的Ray端点 1. 在AI DataLake管理控制台页面，切换页面左上角的工作空间为[步骤二：创建工作空间]新创建的空间。 2. 在左侧导航栏选择"引擎端点 \> AI 计算引擎 Ray"进入Ray端点列表页面。 3. 单击页面右上角的"创建端点"，配置以下参数并单击"立即创建"。 表3创建Ray引擎端点 | 参数 | 示例 | 参数说明 | |:---|:---|:---| | 端点类型 | RayCluster | 选择端点类型。 RayCluster：常驻模式，可多次提交Ray作业。 | | 端点名称 | Ray-Test | 输入端点名称，是端点的唯一标识符，不可与已存在的端点重名，且创建后不支持修改。 名称只能包含小写字母、数字、中划线，且只能以字母开头，以字母或数字结尾。 输入长度不能超过63个字符。 | | 端点显示名称 | 开发团队专用Ray端点-01 | 输入端点显示名称，创建后可修改，用于用户界面展示，方便用户识别和记忆。 - 名称只能包含中文、英文、数字、中划线。 - 输入长度不能超过63个字符。 | | 资源使用模式 | 预留资源 | 选择资源使用模式。 预留资源：预留模式通过预留专属计算资源，确保业务所需的计算资源的稳定性，同时获得最优的单价成本。 | | 选择计算资源池 | resource_pool_for_raytest | 在下拉框中选择[步骤三：创建计算资源池]已创建的资源池。 | | 集群配置 | - Head:64vCPU,128GiB - Workergroups:64vCPU,128GiB | 配置对应Ray集群的Head节点规格、Workergroup规格，可创建多个Workergroup，根据业务需求选择。 规格选择列表中可以看到所有的规格，选择的规格可根据创建的Ray资源向下兼容，例如创建了一个aidatalake.cpu.x86.general.1c2g的资源，那么在选择head规格的时候可以选择aidatalake.cpu.x86.general.1c2g，之后配置vCPU与内存资源。 | | 镜像配置 | 自定义镜像"Ray" | 选择集群镜像版本。 预置镜像：预置镜像为AI DataLake提供的基础镜像版本。 自定义镜像：自定义镜像为用户自行注册的SWR镜像。 | | 存储配置 | - 选择临时存储：sfs-turbo-01 - 存储路径：/ - 挂载路径：/home/service/works | 基于共享存储（SFS Turbo），通过挂载路径规范容器内访问路径，并配合存储路径为Head与Worker节点划定独立的物理子目录，从而在实现跨节点数据互通的同时，解决多节点数据写入冲突的问题。 | 4. 端点创建后，可在列表中查看相关信息，端点状态变为"运行中"后即可提交作业到该端点中运行。 #### 步骤五：使用API提交作业 本示例通过requests库展示一个简单示例，例如"request.py"，模拟通过OpenAPI从创建Ray Job到提交Ray Job的完整流程，执行脚本"request.py"即可提交一个文本处理AI DataLake Job至指定Ray集群。 请求脚本示例如下： ``` import requests import json import urllib3 import time # 禁用SSL警告 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # 公共配置 TOKEN = "XXX" # IAM的token WORKSPACE_ID = "3fa4b5a6-9208-4254-b823-119aa29c****" # 工作空间的ID HEADERS = {"X-Auth-Token": TOKEN, "Content-Type": "application/json"} ENDPOINT = "https://fabric.cn-southwest-2.myhuaweicloud.com" # 实际使用时这个ENDPOINT为对应region endpoint的值，此处为贵阳一地址 # 提交运行URL URL_SUBMIT_RUN = f"https://{ENDPOINT}/v2/workspaces/{WORKSPACE_ID}/ray-jobs" def submit_run(): payload = { "name": "Ray-Test", "description": "description", "endpoint_name": "Ray-Test", #需替换为RayCluster端点名称 "config": { "entrypoint": "python3 -m estimate_pi 1000000", "runtime_env": { "working_dir": "obs://ray-shared-csc/ray-job-code.zip", #需替换为作业脚本存储的obs路径，且必须为.zip文件 "py_modules": [ "string" ], "pip": [ "numpy=1.16.1" # 可选值，填写值为涉及pip依赖库，若不涉及可不填 ], "env_vars": { "additionalProp1": "value1", # 可选值，填写值为环境变量，若不涉及可不填 "additionalProp2": "value2", "additionalProp3": "value3" } } } } try: response = requests.post( URL_SUBMIT_RUN, headers=HEADERS, json=payload, verify=False ) if response.status_code in [200, 201, 202]: print( f"提交运行成功！响应码：{response.status_code}。响应内容：{response.text}" ) else: print(f"提交运行失败 (Code: {response.status_code})") print(response.text) except Exception as e: print(f"请求异常: {e}") if __name__ == "__main__": submit_run() ``` 提交作业后，您可以查看作业，操作如下： 1. 在AI DataLake管理控制台页面的左侧导航栏，选择"作业开发 \> 作业运行历史"，进入作业运行历史页面。 2. 单击"AI计算引擎Ray"页签，查看作业信息。