BigDL-2.x RayOnSpark实战：在Spark集群上运行Ray程序

BigDL-2.x是一个强大的分布式深度学习框架，它允许开发者在Apache Spark集群上无缝运行Ray程序，实现高效的分布式计算。通过RayOnSpark技术，用户可以充分利用Spark集群的资源，同时享受Ray带来的灵活分布式编程模型，为大规模机器学习和深度学习任务提供了强大的支持。## 🚀 什么是RayOnSpark？RayOnSpark是BigDL-2.x中一项创新技术，它实

尚学红Vandal

961人浏览 · 2026-04-05 12:16:57

尚学红Vandal · 2026-04-05 12:16:57 发布

BigDL-2.x RayOnSpark实战：在Spark集群上运行Ray程序

【免费下载链接】BigDL-2.x BigDL: Distributed TensorFlow, Keras and PyTorch on Apache Spark/Flink & Ray 项目地址: https://gitcode.com/gh_mirrors/bi/BigDL-2.x

BigDL-2.x是一个强大的分布式深度学习框架，它允许开发者在Apache Spark集群上无缝运行Ray程序，实现高效的分布式计算。通过RayOnSpark技术，用户可以充分利用Spark集群的资源，同时享受Ray带来的灵活分布式编程模型，为大规模机器学习和深度学习任务提供了强大的支持。

🚀 什么是RayOnSpark？

RayOnSpark是BigDL-2.x中一项创新技术，它实现了在Spark集群上部署和运行Ray应用程序的能力。这意味着你可以直接利用现有的Spark集群资源来运行基于Ray的分布式应用，无需单独维护Ray集群，大大简化了分布式计算架构的复杂性。

图：BigDL的端到端分布式内存管道，展示了数据 ingestion、特征转换、模型训练与评估以及模型优化与推理的完整流程

🔧 环境准备与初始化

要在Spark集群上运行Ray程序，首先需要初始化SparkContext。BigDL提供了便捷的初始化方法，可以根据环境自动选择本地模式或YARN集群模式：

from bigdl.dllib.nncontext import init_spark_on_local, init_spark_on_yarn
import os

hadoop_conf_dir = os.environ.get('HADOOP_CONF_DIR')

if hadoop_conf_dir:
    # YARN集群模式初始化
    sc = init_spark_on_yarn(
        hadoop_conf=hadoop_conf_dir,
        conda_name=os.environ.get("ZOO_CONDA_NAME", "zoo"),
        num_executors=2,
        executor_cores=4,
        executor_memory="2g",
        driver_memory="2g",
        driver_cores=1,
        extra_executor_memory_for_ray="3g"
    )
else:
    # 本地模式初始化
    sc = init_spark_on_local(cores=8, conf={"spark.driver.memory": "2g"})

🎯 在Spark上启动Ray集群

初始化SparkContext后，接下来需要在Spark集群上启动Ray：

import ray
from bigdl.orca.ray import OrcaRayContext

# 在Spark集群上初始化Ray
ray_ctx = OrcaRayContext(sc=sc, object_store_memory="4g")
ray_ctx.init()

这段代码会在Spark集群上自动启动Ray服务，包括Ray的主节点和工作节点，无需手动配置复杂的Ray集群参数。

💡 实战：分布式参数服务器示例

下面我们通过一个分布式参数服务器的例子，展示如何在Spark集群上运行Ray程序。这个例子实现了一个简单的分布式异步随机梯度下降算法。

定义参数服务器Actor

首先，我们定义一个参数服务器Actor，用于存储和更新模型参数：

import numpy as np

@ray.remote
class ParameterServer(object):
    def __init__(self, dim):
        self.parameters = np.zeros(dim)
    
    def get_parameters(self):
        return self.parameters
    
    def update_parameters(self, update):
        self.parameters += update

创建参数服务器实例

dim = 10  # 参数维度
ps = ParameterServer.remote(dim)  # 创建参数服务器实例

定义Worker任务

接下来，定义一个Worker任务，它将从参数服务器获取参数，计算更新，并将更新发送回参数服务器：

import time

@ray.remote
def worker(ps, dim, num_iters):
    for _ in range(num_iters):
        # 获取最新参数
        parameters = ray.get(ps.get_parameters.remote())
        # 计算更新
        update = 1e-3 * parameters + np.ones(dim)
        # 更新参数
        ps.update_parameters.remote(update)
        # 模拟实际工作负载
        time.sleep(0.5)

启动Worker并执行

# 启动两个Worker
worker_results = [worker.remote(ps, dim, 100) for _ in range(2)]

# 查看参数更新结果
print(ray.get(ps.get_parameters.remote()))

📊 进阶：参数服务器分片

当模型规模增大或Worker数量增加时，单个参数服务器可能成为瓶颈。这时可以将参数服务器分片，提高系统吞吐量：

@ray.remote
class ParameterServerShard(object):
    def __init__(self, sharded_dim):
        self.parameters = np.zeros(sharded_dim)
    
    def get_parameters(self):
        return self.parameters
    
    def update_parameters(self, update):
        self.parameters += update

total_dim = (10 ** 8) // 8  # 参数总维度
num_shards = 2  # 分片数量

# 创建多个参数服务器分片
ps_shards = [ParameterServerShard.remote(total_dim // num_shards) for _ in range(num_shards)]