利用Python实现分布式计算：Ray框架入门与集群部署

沈知远

发布时间：2026-10-16| 7人浏览过

原创

本文详细介绍如何利用Python的Ray框架实现高效分布式计算。内容涵盖Ray的核心特性、安装方法、单机分布式代码示例，以及基于Docker的多节点集群搭建与任务调度实战，帮助开发者快速掌握分布式并行处理技术。

前言

在处理计算密集型任务时，除了使用多进程技术外，分布式计算是另一种高效方案。本文将介绍如何使用Python结合Ray库来实现分布式计算。

什么是 Ray

Ray是一个基于Python的分布式计算框架，采用动态图计算模型，提供简洁通用的API以创建分布式应用。其易用性体现在只需通过装饰器修改少量代码，即可将单机Python代码轻松转化为分布式应用，目前广泛应用于机器学习领域。

Ray 的特色：

1、提供构建和运行分布式应用程序的简单原语。
2、支持并行化单机代码，且代码修改量极少甚至为零。
3、Ray Core 包含由应用程序、库和工具组成的大型生态系统，以支持复杂的应用程序。比如 Tune、RLlib、RaySGD、Serve、Datasets、Workflows。

安装 Ray

最简单的安装官方版本的方式：

pip install -U ray
pip install 'ray[default]'

如果是 Windows 系统，要求必须安装Visual C++ runtime

使用 Ray

一个装饰器就搞定分布式计算：

import ray
ray.init()
@ray.remote
def f(x):
    return x * x
futures = [f.remote(i) for i in range(4)]
print(ray.get(futures)) # [0, 1, 4, 9]

先执行ray.init()，然后在要执行分布式任务的函数前加一个装饰器@ray.remote就实现了分布式计算。装饰器@ray.remote也可以装饰一个类：

import ray
ray.init()

@ray.remote
class Counter(object):
    def __init__(self):
        self.n = 0
    def increment(self):
        self.n += 1
    def read(self):
        return self.n
counters = [Counter.remote() for i in range(4)]
tmp1 = [c.increment.remote() for c in counters]
tmp2 = [c.increment.remote() for c in counters]
tmp3 = [c.increment.remote() for c in counters]
futures = [c.read.remote() for c in counters]
print(ray.get(futures)) # [3, 3, 3, 3]

当然了，上述的分布式计算依然是在自己的电脑上进行的，只不过是以分布式的形式。程序执行的过程中，你可以输入http://127.0.0.1:8265/#/查看分布式任务的执行情况：

那么如何实现 Ray 集群计算呢？接着往下看。

使用 Ray 集群

Ray 的优势之一是能够在同一程序中利用多台机器。当然，Ray 可以在一台机器上运行，因为通常情况下，你只有一台机器。但真正的力量是在一组机器上使用 Ray。

Ray 集群由一个头节点和一组工作节点组成。需要先启动头节点，给 worker 节点赋予头节点地址，组成集群：

你可以使用 Ray Cluster Launcher 来配置机器并启动多节点 Ray 集群。你可以在 AWS、GCP、Azure、Kubernetes、阿里云、内部部署和 Staroid 上甚至在你的自定义节点提供商上使用集群启动器。

Ray 集群还可以利用 Ray Autoscaler，它允许 Ray 与云提供商交互，以根据规范和应用程序工作负载请求或发布实例。

现在，我们来快速演示下 Ray 集群的功能，这里是用 Docker 来启动两个 Ubuntu 容器来模拟集群：

环境 1: 172.17.0.2 作为 head 节点
环境 2: 172.17.0.3 作为 worker 节点，可以有多个 worker 节点

具体步骤：

1. 下载 ubuntu 镜像

docker pull ubuntu

2. 启动 ubuntu 容器，安装依赖

启动第一个

docker run -it --name ubuntu-01 ubuntu bash

启动第二个

docker run -it --name ubuntu-02 ubuntu bash

检查下它们的 IP 地址：

$ docker inspect -f "{{ .NetworkSettings.IPAddress }}" ubuntu-01
172.17.0.2
$ docker inspect -f "{{ .NetworkSettings.IPAddress }}" ubuntu-02
172.17.0.3

然后分别在容器内部安装 python、pip、ray

apt update && apt install python3 
apt install python3-pip
pip3 install ray

3. 启动 head 节点和 worker 节点

选择在其中一个容器作为 head 节点，这里选择 172.17.0.2，执行：

ray start --head --node-ip-address 172.17.0.2

默认端口是 6379，你可以使用 --port 参数来修改默认端口，启动后的结果如下：

忽略掉警告，可以看到给出了一个提示，如果要把其他节点绑定到该 head，可以这样：

ray start --address='172.17.0.2:6379' --redis-password='5241590000000000'

在另一个节点执行上述命令，即可启动 worker 节点：

如果要关闭，执行：

ray stop

4、执行任务

随便选择一个节点，执行下面的脚本，修改下 ray.init() 函数的参数：

from collections import Counter
import socket
import time
import ray
ray.init(address='172.17.0.2:6379', _redis_password='5241590000000000')

print('''This cluster consists o    f
    {} nodes in total
    {} CPU resources in total
'''.format(len(ray.nodes()), ray.cluster_resources()['CPU']))
@ray.remote
def f():
    time.sleep(0.001)
    # Return IP address.
    return socket.gethostbyname(socket.gethostname())
object_ids = [f.remote() for _ in range(10000)]
ip_addresses = ray.get(object_ids)
print('Tasks executed')
for ip_address, num_tasks in Counter(ip_addresses).items():
    print('    {} tasks on {}'.format(num_tasks, ip_address))

执行结果如下：