01.Ray分布式框架介绍

Leefs 2025-03-15 PM 883℃ 0条

[TOC]

### 一、Ray框架简介

Ray是一个开源的分布式计算框架，用于构建和扩展分布式应用。它提供了简单的API，使得开发者可以轻松地编写并行和分布式代码，而无需担心底层的复杂性。

Ray支持多种编程范式，包括任务并行、Actor模型、分布式对象存储等。

现在的机器学习等任务在性能（任务高速响应），灵活性（计算任务差异），可靠性方面都有很高要求，因此设计一个新的计算框架Ray，提出了逻辑中心控制状态板（GCS）的概念，采用分片的分布式存储系统，提供一种自底向上的分布式调度器来实现。

![01.Ray分布式框架介绍01.png](https://lilinchao.com/usr/uploads/2025/03/3668917006.png)

由上面的Ray 框架图来看，Ray Core即是底层的核心实现，使得开发者能基于此，在本地机器、K8s集群或者各种云上开发构建可扩展的分布式应用。

在上层，Ray 提供了一系列的原生库，用于开发和部署机器学习任务，如可扩展的数据处理库、分布式训练库、超参调优库、模型服务以及分布式强化学习库等。

机器学习应用或者数据应用增长速度远快于单个节点或者单个处理器的能力，必须使用分布式技术来处理这些工作，然而直接编写分布式应用非常困难，现在利用Ray 框架，可以方便的开发分布式任务，而不需要担心分布式组件通信、服务部署、服务发现、监控和异常恢复等。

### 二、Ray架构设计

Ray架构分为**应用层**和**系统层**组成，应用层提供了Ray API，系统层保障Ray的高可扩展和容错性；

![01.Ray分布式框架介绍02.png](https://lilinchao.com/usr/uploads/2025/03/1817390755.png)

作为分布式计算系统，Ray仍旧遵循了典型的Master-Slave的设计：Master负责全局协调和状态维护，Slave执行分布式计算任务。

不过和传统的分布式计算系统不同的是，Ray使用了**混合任务调度**的思路。

在集群部署模式下，Ray启动了以下关键组件：

+ **GlobalScheduler**：Master上启动了一个全局调度器，用于接收本地调度器提交的任务，并将任务分发给合适的本地任务调度器执行。

+ **RedisServer**：Master上启动了一到多个RedisServer用于保存分布式任务的状态信息（ControlState），包括对象机器的映射、任务描述、任务debug信息等。

+ **LocalScheduler**：每个Slave上启动了一个本地调度器，用于提交任务到全局调度器，以及分配任务给当前机器的Worker进程。

+ **Worker**：每个Slave上可以启动多个Worker进程执行分布式任务，并将计算结果存储到ObjectStore。

+ **ObjectStore**：每个Slave上启动了一个ObjectStore存储只读数据对象，Worker可以通过共享内存的方式访问这些对象数据，这样可以有效地减少内存拷贝和对象序列化成本。ObjectStore底层由Apache Arrow实现。

+ **Plasma**：每个Slave上的ObjectStore都由一个名为Plasma的对象管理器进行管理，它可以在Worker访问本地ObjectStore上不存在的远程数据对象时，主动拉取其它Slave上的对象数据到当前机器。

应用层对应了三种类型的进程：**驱动进程、工作器进程、行动器进程组成；**

+ **驱动器 (Driver )**

执行用户程序的进程，所有操作都需要由主进程来驱动。

+ **工作器 (Worker )**

执行由驱动器或其他工作器调用的任务（远程函数）的无状态的进程。工作器是在系统层分配任务时自动启动的。当声明一个远程函数时，该函数将被自动发送到所有的工作器中。在同一个工作器中，任务是串行地执行的，工作器并不维护其任务与任务之间的局部状态，即在工作器中，一个远程函数执行完后，其局部作用域的所有变量将不再能被其他任务所访问。

+ **行动器 (Actor )**

行动器被调用时只执行其所暴露的方法。行动器由工作器或驱动器显式地进行实例化。与工作器相同的是，行动器也会串行地执行任务，不同的是行动器上执行的每个方法都依赖于其前面所执行的方法所变更的状态。

### 三、Ray的执行原理

![01.Ray分布式框架介绍03.png](https://lilinchao.com/usr/uploads/2025/03/3877683845.png)

Ray 集群是由一个或者多个worker 节点组成，其中一个节点被指定为head 节点。每个节点都有:

+ 一个或者多个worker进程，要么是无状态的remote function，要么是actor。
+ raylet 进程，用于管理每个节点的共享资源，与worker 进程不同的是，raylet在所有并发的job中是共享的，主要有2个组件运行在单独的线程
  + scheduler 组件，负责资源管理，任务调度和完成，将Task的参数存储在Object Store中；目前支持多种调度策略。
  + object store 组件，存储和转移大对象。

集群内部大部分都是通过gRPC 调用通信的，保障集群内部通信效率。

+ **分布式调度（Distributed Scheduler）**

运行在每个节点的Raylet进程上，负责资源管理，task 放置的位置， 保证 task 运行所需要的参数对象可以从分布式对象存储的对象中获取。

每个 Raylet 组件会跟踪本地节点的资源， 当一个资源请求被同意之后，Raylet 就会减少本地可用资源，一旦资源被使用完之后返还回来了，Raylet 又会增加本地的可用资源， 所以 Raylet 有一个强一致性的本地的可用资源的视图。

同时，Raylet 也接收来自GCS 服务的一些关于其它节点的资源使用的信息， 这个信息（最终一致性）对于分布式调度是很有用。

比如，在集群范围内，可以多种不同策略（数据位置、节点亲和性等）的调度。GCS 会定期（默认100ms）从每个节点的 Raylet 拉取本地节点的可用资源，然后将这些信息通过广播的方式告知所有节点的 Raylet。

+ **分布式对象存储(Distributed Object Store)**

分布式共享内存存储，在同一个节点上可以实现不同的woker进程之间能够零拷贝（zero copy）地访问共同的数据，不同节点上也可以引用。

具体实现采用了Apache Arrow中的Plasma store，负责存储和转移大对象，如果plasma store满了之后，会通过LRU 机制剔除陈旧的对象溢出到外部存储（disk或者s3等）。

+ **全局状态控制器(Global Controler Service)**

之前的只在head节点，v2版本后支持了容错，可以运行在任意节点或多个节点。

主要包含节点管理，资源管理，PlacementGroup管理

另外有些额外的辅助服务，如AutoScaler、Job Commit、监控等，使得集群管理更加简单。

**根本上说，Ray是一个RPC（远程过程调用）框架，加上一个actor框架，以及一个对象存储，它允许你在不同函数和actor之间通过引用（reference）高效传递数据。**

### 四、Ray两种计算模式

#### 4.1 任务Task

任务执行为无状态的，任务无法修改作为本地变量传入的值，Ray远程函数为无副作用的；

**编写任务流程**：

1. 注册任务：在注册为任务的函数上添加`@ray.remote`修饰器
2. 提交任务：在调用`@ray.remote`修饰器的函数时需带上`.remote()`
3. 非阻塞提交：提交任务后立即返回`ObjectRef`对象
4. 阻塞获取结果：通过`ray.get`传入返回的`ObjectRef`对象获取函数返回值

#### 4.2 行动器Actor

有状态的的计算任务，行动器方法调用可能会修改行动器状态，属于有副作用的函数，因此同一行动器下的方法需按顺序串行调用；

**编写行动器流程**：

1. 注册行动器：在注册行为器的类上加上`@ray.remote`装饰器
2. 实例化行动器：实例化类实例时需在类名后加上`.remote()`
3. 提交方法调用：调用行动器方法需加上`.remote()`
4. 非阻塞提交：提交后返回一个`ObjectRef`对象，同一行动器实例下方法会按提交顺序执行
5. 阻塞获取结果：通过`ray.get`传入`ObjectRef`获取结果

### 五、Ray的安装和启动

#### 5.1 Ray集群安装组件选择

Ray框架有多个组成部分，可单独或组合安装，组件包括Ray Core、Ray Data、Ray Data、Ray Train、Ray Tune、Ray Serve、Dashboard等；

+ Ray Core提供了分布式应用的核心的基础部分的支持如Tasks、Actors、Objects等；

+ Ray Train是一个可扩展的机器学习库，用于分布式训练和微调。

其支持PyTorch、TensorFlow、Keras、XGBoost、LightGBM、Hugging Face Transformers等框架；

由于是一个Python的框架，Ray可以直接使用pip进行安装和管理：

| Command                       | Installed components                                         |
| ----------------------------- | ------------------------------------------------------------ |
| pip install -U "ray"          | Core                                                         |
| pip install -U "ray[default]" | Core, Dashboard, Cluster Launcher                            |
| pip install -U "ray[data]"    | Core, Data                                                   |
| pip install -U "ray[train]"   | Core, Train                                                  |
| pip install -U "ray[tune]"    | Core, Tune                                                   |
| pip install -U "ray[serve]"   | Core, Dashboard, Cluster Launcher, Serve                     |
| pip install -U "ray[rllib]"   | Core, Tune, RLlib                                            |
| pip install -U "ray[air]"     | Core, Dashboard, Cluster Launcher, Data, Train, Tune, Serve  |
| pip install -U "ray[all]"     | Core, Dashboard, Cluster Launcher, Data, Train, Tune, Serve, RLlib |

+ 如果想使用Ray所有功能，或者对存储空间没有要求，直接pip install -U "ray[all]"即可。
+ 各操作系统均可安装（Linux、MacOS、Windows），支持python 3.7-3.10等多版本

#### 5.2 Ray集群安装

**（1）安装Anaconda环境**

```sh
# 下载安装脚本
$ wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
# 添加执行权限
$ chmod u+x Miniconda3-latest-Linux-x86_64.sh 
# 运行安装脚本
$ ./Miniconda3-latest-Linux-x86_64.sh
```

**（2）安装python**

为ray准备一个python环境，以python3.7.13示例：

```sh
# 创建一个名为ray，版本为3.7.13的python环境
$ conda create --name ray python=3.7.13
# 激活名为ray的python环境
$ conda activate ray
```

安装完之后，最好重新登录一下，或者执行一下`source ~/.bashrc`使得环境变量生效

**（3）安装Ray**

安装指定版本的ray环境，这里为了加快速度指定了阿里的镜像源

```sh
$ pip install -i https://mirrors.aliyun.com/pypi/simple --trusted-host mirrors.aliyun.com -U 'ray[default]'==2.7.2
```

需要注意的是，在所有需要构建集群的设备上，需要统一Python和Ray的版本，因此建议先使用conda创建同样的虚拟环境之后，再安装统一版本的ray。

否则在添加集群节点的时候就有可能出现如下问题：

```
RuntimeError: Version mismatch: The cluster was started with:
    Ray: 2.7.2
    Python: 3.7.13
This process on node 172.17.0.2 was started with:
    Ray: 2.7.2
    Python: 3.7.5
```

#### 5.3 Ray集群启动

Ray的架构遵循master-slave的模式。Head Node 可以认为是Master，其他的Node为worker。在集群部署时，Head Node需要首先启动ray start --head, 其他机器依次启动worker,注意需要指定head Node的地址确定关系。

**（1）启动head节点**

在`192.168.12.101`上启动Head节点:

```sh
$ ray start --head --dashboard-host='0.0.0.0' --dashboard-port=8265
```

正常会看到如下输出：

```
Local node IP: 192.168.12.101
2025-03-02 18:33:11,977 INFO services.py:1250 -- View the Ray dashboard at http://192.168.12.101:8265
--------------------
Ray runtime started.
--------------------
Next steps
  To connect to this Ray runtime from another node, run
    ray start --address='192.168.12.101:6379' --redis-password='5241590000000000'
...
```

输出信息包含了2个关键信息，需要别注意：

- `View the Ray dashboard at http://192.168.12.101:8265`：web服务的地址
- `ray start --address='192.168.12.101:6379' --redis-password='524159000'`：head的地址和密码

**（2）启动worker节点**

在`192.168.12.102`上，按照上面的步骤将python和ray安装好，注意它们的版本必须保持一致。

（另外，Worker不是必须的，因为Head节点本身就具有worker角色）

```sh
# 连接指定的Head地址
$ ray start --address='192.168.12.101:6379' --redis-password='5241590000000000'
```

**（3）Dashboad**

访问dashboard：`http://192.168.12.101:8265`

![01.Ray分布式框架介绍04.png](https://lilinchao.com/usr/uploads/2025/03/1600452752.png)

标签: Ray

非特殊说明，本博所有文章均为博主原创。

如若转载，请注明出处：https://lilinchao.com/archives/2951.html

上一篇 06.Pytorch张量索引操作

下一篇 02.Ray简单使用示例

01.Ray分布式框架介绍

评论已关闭

栏目分类

标签云

友情链接申请

01.Ray分布式框架介绍

评论已关闭

 栏目分类

标签云

友情链接申请

栏目分类

标签云

友情链接申请