CS267 · 第 13 课

Ray 分布式框架

从课程项目到全球分布式计算平台：Ion Stoica 讲述 Ray 如何通过极简 API 与统一计算模型，让 Python 开发者轻松编写分布式 AI 应用。

---

Ray 的诞生背景

Ray 起源于 2016 年加州大学伯克利分校的 ODI（Observatory on Distributed Intelligence）项目，由 Ion Stoica 和 Joseph E. Gonzalez 指导的学生团队（包括 Philipp Moritz、Robert Nishihara 等）最初作为课程研究项目的产物。其核心动机是：强化学习等新兴 AI 工作负载需要一种比 MapReduce / Spark 更灵活的分布式编程模型。

为什么需要 Ray？AI 工作负载的三大趋势

Ion Stoica 指出，推动 Ray 需求的三大趋势是：

趋势	含义	对框架的要求
AI 需求 > 硬件增长	AI 计算需求每年增长 10x，远超摩尔定律	必须能横向扩展到数千 GPU
异构集群	单机多卡 → 多机多卡 → CPU/GPU/TPU 混合	统一管理异构资源和调度
复杂 AI 工作负载	训练、推理、数据处理、超参搜索交织	统一框架支持多阶段 pipeline

Ray 的核心设计理念：统一计算框架

Ray 的目标是成为一个统一的计算框架，用同一套 API 表达从单机 Python 程序到数千节点的分布式应用。其设计哲学是"minimalist API"——用最少的概念覆盖最多的场景。

极简 API：六行代码入门

Ray 的 API 核心仅由几个函数和装饰器组成：

import ray

# 1. 连接到 Ray 集群（或本地启动）
ray.init()

# 2. 将普通函数转为分布式任务（Task）
@ray.remote
def compute(x):
    return x * x

# 3. 异步提交任务，返回 Future
futures = [compute.remote(i) for i in range(10)]

# 4. 获取结果
results = ray.get(futures)

# 5. 将对象存入共享内存 Object Store
ref = ray.put(large_object)

# 6. 等待部分结果（不阻塞全部）
done, pending = ray.wait(futures, num_returns=3)

API	作用	特点
ray.init()	初始化 Ray 运行时	连接集群或本地启动
@ray.remote	装饰器：函数→Task，类→Actor	声明式并行
.remote()	异步调用 remote 函数/方法	返回 ObjectRef (Future)
ray.get()	阻塞等待并获取结果	可传入列表批量获取
ray.put()	将对象存入 Object Store	获得引用，零拷贝共享
ray.wait()	非阻塞等待部分结果	支持超时和 num_returns

---

FAST 计算模型

Ray 用四个核心抽象（FAST）统一所有分布式计算模式：

抽象	英文	语义	类比
Function	Task	无状态、幂等的远程函数执行	类似函数调用，无副作用
Actor	Actor	有状态的分布式对象，方法串行执行	带邮箱的 Actor 模型
Schedule	调度器	全局 + 两级调度，感知数据位置	混合调度架构
Transport	Object Store	共享内存对象存储，零拷贝传输	分布式共享内存

Python 示例：从函数到 Task，从类到 Actor

import ray

ray.init()

# ===== Task：无状态函数 =====
@ray.remote
def add(a, b):
    return a + b

# 提交 1000 个任务，自动并行
refs = [add.remote(i, i) for i in range(1000)]
results = ray.get(refs)

# ===== Actor：有状态对象 =====
@ray.remote
class Counter:
    def __init__(self):
        self.value = 0

    def increment(self):
        self.value += 1
        return self.value

# 创建 10 个独立 Actor 实例
counters = [Counter.remote() for _ in range(10)]
# 每个 Actor 内部状态独立
refs = [c.increment.remote() for c in counters]
print(ray.get(refs))  # [1, 1, 1, ...]

共享内存 Object Store：零拷贝的魔法

Ray 的核心创新之一是跨节点的共享内存 Object Store，基于 Apache Arrow Plasma 对象存储实现：

---

Ray 生态系统

基于 Ray Core，社区构建了一系列高层库，覆盖 ML 全生命周期：

库	功能	典型场景
Ray Core	Task/Actor/API 基础	自定义分布式应用
Ray Data	分布式数据加载/预处理	ETL、数据湖读取
Ray Train	分布式训练	PyTorch/TF 多机训练
Ray Tune	超参数搜索	大规模 HP 优化
Ray Serve	模型服务/在线推理	低延迟模型部署
RLlib	强化学习	分布式 RL 训练

Compiled Graph：静态 DAG 的低开销执行

Ray v2.0 引入 Compiled Graph（原名 Ray DAG），适用于延迟敏感的场景：

• 用户用 @ray.remote 声明静态计算图（DAG）
• Ray 在启动时预分配资源、预编译执行计划
• 运行时直接执行，跳过调度开销
• 适合在线推理（要求 P99 延迟极低）

应用场景示例

场景	Ray 如何支持
在线推理	Compiled Graph + Serve，静态 DAG 实现微秒级 P99 延迟
模型微调	Train + Data，分布式数据并行 + 流水线并行
多模态搜索	Data 处理异构数据，Train 嵌入模型，Serve 在线查询

---

Ray 架构深度解析

全局控制存储（GCS）

GCS（Global Control Store）是 Ray 集群的元数据中枢：

• 存储所有 Actor 的位置、任务调度信息、对象位置映射
• 基于 Redis 的改进版，支持 TTL 和故障检测
• 集群中唯一的全局状态源（single source of truth）

分布式调度器

Ray 采用两级调度架构：p>

层级	职责	位置
全局调度器	跨节点调度，考虑数据位置和资源约束	GCS 内
本地调度器	节点内任务排队、CPU/GPU 资源分配	每个 Node 上

容错机制

Ray 提供两种互补的容错策略：

机制	原理	适用场景
Lineage-based	每个对象记录其计算 lineage（依赖图），丢失时重算	Task（无状态，可重放）
Ownership	Actor 由 Owner 管理，Owner 失效则 Actor 不可恢复	Actor（有状态，需外部存储 checkpoint）

---

为什么需要 Actor？（深度分析）

Ion Stoica 强调，Actor 模型解决了 Task 无法表达的几种关键模式：

场景	为什么 Task 不够	Actor 如何解决
参数服务器	需维护全局模型状态	Actor 持有模型，方法串行更新
环境模拟	RL 环境需连续状态	Actor 封装环境，step() 保持状态
GPU 亲和性	Task 无绑定，GPU 切换开销大	Actor 绑定 GPU，避免 context switch
流水线阶段	需长期运行的计算阶段	Actor 常驻内存，低延迟响应

传递 Actor Handle：构建分布式应用

Ray 允许将一个 Actor 的 handle 作为参数传递给另一个 Task/Actor，实现复杂的分布式拓扑：

@ray.remote
class Worker:
    def __init__(self, param_server_handle):
        self.ps = param_server_handle  # 持有参数服务器的引用

    def train_step(self, batch):
        # 通过 handle 远程调用参数服务器
        weights = ray.get(self.ps.get_weights.remote())
        grads = compute_gradients(weights, batch)
        self.ps.apply_gradients.remote(grads)

# 创建参数服务器 Actor
param_server = ParameterServer.remote()

# 将 handle 传给所有 Worker
workers = [Worker.remote(param_server) for _ in range(8)]
ray.get([w.train_step.remote(batch) for w, batch in zip(workers, batches)])

调度演进：从 Direct Actor Call 到 ESCHER

Ray 的调度策略经历了三代演进：

代	策略	特点
第 1 代	Direct Actor Call	Actor 方法调用直接发送到目标节点，简单但无法全局优化
第 2 代	集中式调度	所有调度决策经 GCS，全局最优但 GCS 成为瓶颈
第 3 代	ESCHER	Ephemeral Resources：调度时动态创建临时资源，实现细粒度负载均衡

---

本课要点

---

知识检查