Rayismultitaskingofdisturbingpython.

并行和分布式计算是现代应用程序的主要内容。我们需要利用多个核心或多台机器来加速应用程序或大规模运行它们。网络爬虫和搜索所使用的基础设施并不是在某人笔记本电脑上运行的单线程程序&＃xff0c;而是相互通信和交互的服务的集合。

ray的api接口教程&＃xff1a;https://ray.readthedocs.io/en/latest/api.html

为什么要使用 Ray&＃xff1f;

        很多教程解释了如何使用 Python 的多进程模块&＃xff08;https://docs.python.org/2/library/multiprocessing.html&＃xff09;。遗憾的是&＃xff0c;多进程模块在处理现代应用程序的要求方面存在严重的短板。这些要求包括以下这些内容&＃xff1a;

在多台计算机上运行相同的代码。

构建有状态且可以与之通信的微服务和 actor。

优雅地处理机器故障。

有效处理大对象和数值数据。

AI的开源框架

与深度学习框架的关系&＃xff1a; Ray与TensorFlow&＃xff0c;PyTorch和MXNet等深度学习框架完全兼容&＃xff0c;在许多应用中与Ray一起使用一个或多个深度学习框架是很自然的&＃xff08;例如&＃xff0c;我们的强化学习库使用TensorFlow和PyTorch&＃xff09;。

与其他分布式系统的关系&＃xff1a;今天使用了许多流行的分布式系统&＃xff0c;但是其中大多数并不是用AI应用程序构建的&＃xff0c;并且缺乏支持所需的性能以及表示AI应用程序的API。从今天的分布式系统来看&＃xff0c;它们缺少以下功能&＃xff08;以各种组合方式&＃xff09;&＃xff1a;

支持毫秒级任务和每秒数百万个任务
嵌套并行&＃xff08;在任务内并行化任务&＃xff0c;例如超参数搜索内部的并行模拟&＃xff09;&＃xff08;见下图&＃xff09;
在运行时动态确定任意任务依赖关系&＃xff08;例如&＃xff0c;为了避免等待缓慢的工作人员&＃xff09;
在共享可变状态下运行的任务&＃xff08;例如&＃xff0c;神经网络权重或模拟器&＃xff09;
支持异构资源&＃xff08;CPU&＃xff0c;GPU等&＃xff09;

有两种使用Ray的主要方法&＃xff1a;通过其较低级别的API和更高级别的库。较高级别的库建立在较低级别的API之上。目前这些包括Ray RLlib&＃xff0c;一个可扩展的强化学习库和Ray.tune&＃xff0c;一个高效的分布式超参数搜索库。

ray.init(redis_address&＃61;"123.45.67.89:6379")

这些过程包括&＃xff1a;

• 有很多 worker 进程并行执行 Python 函数&＃xff08;大概是每个 CPU 核心对应一个 worker&＃xff09;。

• 用于将“任务”分配给 worker&＃xff08;以及其他计算机&＃xff09;的调度程序进程。任务是 Ray 调度的工作单元&＃xff0c;对应于一个函数调用或方法调用。

• 共享内存对象存储库&＃xff0c;用于在 worker 之间有效地共享对象&＃xff08;无需创建副本&＃xff09;。

• 内存数据库&＃xff0c;用于存储在发生机器故障时重新运行任务所需的元数据。

Ray worker 是独立的进程&＃xff0c;而不是线程&＃xff0c;因为在 Python 中存在全局解释器锁&＃xff0c;所以对多线程的支持非常有限。

Ray低级API

Ray API的目标是自然地表达非常普遍的计算模式和应用程序&＃xff0c;而不受像MapReduce这样的固定模式的限制。

动态任务图

Ray应用程序或作业中的基础基元是一个动态任务图。这与TensorFlow中的计算图非常不同。而在TensorFlow中&＃xff0c;一个计算图代表一个神经网络&＃xff0c;并且在单个应用程序中执行多次&＃xff0c;在Ray中&＃xff0c;任务图代表整个应用程序&＃xff0c;并且只执行一次。任务图不是事先知道的。它是在应用程序运行时动态构建的&＃xff0c;执行一个任务可能会触发创建更多任务。

任意的Python函数都可以作为任务执行&＃xff0c;并且可以任意依赖其他任务的输出。下面的例子给出了说明。

要将 Python 函数 f 转换为一个“远程函数”&＃xff08;可以远程和异步执行的函数&＃xff09;&＃xff0c;可以使用 &＃64;ray.remote 装饰器来声明这个函数。然后函数调用 f.remote() 将立即返回一个 future&＃xff08;future 是对最终输出的引用&＃xff09;&＃xff0c;实际的函数执行将在后台进行&＃xff08;我们将这个函数执行称为任务&＃xff09;。

要将一个任务的输出作为输入提供给后续任务&＃xff0c;只需将第一个任务返回的 future 作为参数传给第二个任务。Ray 的调度程序会自动考虑任务依赖关系。在第一个任务完成之前不会执行第二个任务&＃xff0c;第一个任务的输出将自动被发送给执行第二个任务的机器。

import ray
import time,datetime# Start Ray.
ray.init()import numpy as np# 定义两个远程函数。
# 这些函数的调用创建了远程执行的任务&＃64;ray.remote
def create_matrix(size):return np.random.normal(size&＃61;size)
&＃64;ray.remote
def multiply_matrices(x, y):return np.dot(x, y)result_ids &＃61; []
for i in range(400):# 开始两个并行的任务&＃xff0c;这些会立即返回futures并在后台执行x_id &＃61; create_matrix.remote([1000, 1000])print(datetime.datetime.now())y_id &＃61; create_matrix.remote([1000, 1000])print(datetime.datetime.now())# 开始第三个任务&＃xff0c;但这并不会被提前计划&＃xff0c;直到前两个任务都完成了.result_ids.append(multiply_matrices.remote(x_id, y_id))print(datetime.datetime.now())
# 获取结果。这个结果直到第三个任务完成才能得到。只有get创建以后所有的任务才开始创建执行。
z_id &＃61; ray.get(result_ids)
print(z_id)

机器耗能&＃xff1a;

有效的使用聚合函数

下图是两个聚合过程和相应的函数。以线性方式聚合值与以树形结构方式聚合值的对比

右图方式的聚合函数会比左图方式的聚合更高校&＃xff0c;因为在一个任务

import time
&＃64;ray.remote
def add(x, y):time.sleep(1)return x &＃43; y
# Aggregate the values slowly. This approach takes O(n) where n is the
# number of values being aggregated. In this case, 7 seconds.
id1 &＃61; add.remote(1, 2)
id2 &＃61; add.remote(id1, 3)
id3 &＃61; add.remote(id2, 4)
id4 &＃61; add.remote(id3, 5)
id5 &＃61; add.remote(id4, 6)
id6 &＃61; add.remote(id5, 7)
id7 &＃61; add.remote(id6, 8)
result &＃61; ray.get(id7)
# Aggregate the values in a tree-structured pattern. This approach
# takes O(log(n)). In this case, 3 seconds.
id1 &＃61; add.remote(1, 2)
id2 &＃61; add.remote(3, 4)
id3 &＃61; add.remote(5, 6)
id4 &＃61; add.remote(7, 8)
id5 &＃61; add.remote(id1, id2)
id6 &＃61; add.remote(id3, id4)
id7 &＃61; add.remote(id5, id6)
result &＃61; ray.get(id7)# Slow approach.
values &＃61; [1, 2, 3, 4, 5, 6, 7, 8]
while len(values) > 1:values &＃61; [add.remote(values[0], values[1])] &＃43; values[2:]
result &＃61; ray.get(values[0])
# Fast approach.
values &＃61; [1, 2, 3, 4, 5, 6, 7, 8]
while len(values) > 1:values &＃61; values[2:] &＃43; [add.remote(values[0], values[1])]
result &＃61; ray.get(values[0])

简单开始

首先来看一下最简单的Ray程序是如何编写的。

# 导入ray&＃xff0c;并初始化执行环境
import ray
ray.init()# 定义ray remote函数
&＃64;ray.remote
def hello():return "Hello world !"# 异步执行remote函数&＃xff0c;返回结果id
object_id &＃61; hello.remote()# 同步获取计算结果
hello &＃61; ray.get(object_id)# 输出计算结果
print hello

在Ray里&＃xff0c;通过Python注解&＃64;ray.remote定义remote函数。使用此注解声明的函数都会自带一个默认的方法remote&＃xff0c;通过此方法发起的函数调用都是以提交分布式任务的方式异步执行的&＃xff0c;函数的返回值是一个对象id&＃xff0c;使用ray.get内置操作可以同步获取该id对应的对象。熟悉Java里的Future机制的话对此应该并不陌生&＃xff0c;或许会有人疑惑这和普通的异步函数调用没什么大的区别&＃xff0c;但是这里最大的差异是&＃xff0c;函数hello是分布式异步执行的。

remote函数是Ray分布式计算抽象中的核心概念&＃xff0c;通过它开发者拥有了动态定制计算依赖&＃xff08;任务DAG&＃xff09;的能力。比如&＃xff1a;

&＃64;ray.remote
def A():return "A"&＃64;ray.remote
def B():return "B"&＃64;ray.remote
def C(a, b):return "C"a_id &＃61; A.remote()
b_id &＃61; B.remote()
c_id &＃61; C.remote(a_id, b_id)
print ray.get(c_id)

例子代码中&＃xff0c;对函数A、B的调用是完全并行执行的&＃xff0c;但是对函数C的调用依赖于A、B函数的返回结果。Ray可以保证函数C需要等待A、B函数的结果真正计算出来后才会执行。如果将函数A、B、C类比为DAG的节点的话&＃xff0c;那么DAG的边就是函数C参数对函数A、B计算结果的依赖&＃xff0c;自由的函数调用方式允许Ray可以自由地定制DAG的结构和计算依赖关系。另外&＃xff0c;提及一点的是Python的函数可以定义函数具有多个返回值&＃xff0c;这也使得Python的函数更天然具备了DAG节点多入和多出的特点。

二、系统架构

Ray是使用什么样的架构对分布式计算做出如上抽象的呢&＃xff0c;一下给出了Ray的系统架构&＃xff08;来自Ray论文&＃xff0c;参考文献1&＃xff09;。

作为分布式计算系统&＃xff0c;Ray仍旧遵循了典型的Master-Slave的设计&＃xff1a;Master负责全局协调和状态维护&＃xff0c;Slave执行分布式计算任务。不过和传统的分布式计算系统不同的是&＃xff0c;Ray使用了混合任务调度的思路。在集群部署模式下&＃xff0c;Ray启动了以下关键组件&＃xff1a;

1. GlobalScheduler&＃xff1a;Master上启动了一个全局调度器&＃xff0c;用于接收本地调度器提交的任务&＃xff0c;并将任务分发给合适的本地任务调度器执行。
2. RedisServer&＃xff1a;Master上启动了一到多个RedisServer用于保存分布式任务的状态信息&＃xff08;ControlState&＃xff09;&＃xff0c;包括对象机器的映射、任务描述、任务debug信息等。
3. LocalScheduler&＃xff1a;每个Slave上启动了一个本地调度器&＃xff0c;用于提交任务到全局调度器&＃xff0c;以及分配任务给当前机器的Worker进程。
4. Worker&＃xff1a;每个Slave上可以启动多个Worker进程执行分布式任务&＃xff0c;并将计算结果存储到ObjectStore。
5. ObjectStore&＃xff1a;每个Slave上启动了一个ObjectStore存储只读数据对象&＃xff0c;Worker可以通过共享内存的方式访问这些对象数据&＃xff0c;这样可以有效地减少内存拷贝和对象序列化成本。ObjectStore底层由Apache Arrow实现。
6. Plasma&＃xff1a;每个Slave上的ObjectStore都由一个名为Plasma的对象管理器进行管理&＃xff0c;它可以在Worker访问本地ObjectStore上不存在的远程数据对象时&＃xff0c;主动拉取其它Slave上的对象数据到当前机器。

需要说明的是&＃xff0c;Ray的论文中提及&＃xff0c;全局调度器可以启动一到多个&＃xff0c;而目前Ray的实现文档里讨论的内容都是基于一个全局调度器的情况。我猜测可能是Ray尚在建设中&＃xff0c;一些机制还未完善&＃xff0c;后续读者可以留意此处的细节变化。

Ray的任务也是通过类似Spark中Driver的概念的方式进行提交的&＃xff0c;有所不同的是&＃xff1a;

1. Spark的Driver提交的是任务DAG&＃xff0c;一旦提交则不可更改。
2. 而Ray提交的是更细粒度的remote function&＃xff0c;任务DAG依赖关系由函数依赖关系自由定制。

论文给出的架构图里并未画出Driver的概念&＃xff0c;因此我在其基础上做了一些修改和扩充。

Ray的Driver节点和和Slave节点启动的组件几乎相同&＃xff0c;不过却有以下区别&＃xff1a;

1. Driver上的工作进程DriverProcess一般只有一个&＃xff0c;即用户启动的PythonShell。Slave可以根据需要创建多个WorkerProcess。
2. Driver只能提交任务&＃xff0c;却不能接收来自全局调度器分配的任务。Slave可以提交任务&＃xff0c;也可以接收全局调度器分配的任务。

Driver可以主动绕过全局调度器给Slave发送Actor调用任务&＃xff08;此处设计是否合理尚不讨论&＃xff09;。Slave只能接收全局调度器分配的计算任务

三、核心操作

基于以上架构&＃xff0c;我们简单讨论一下Ray中关键的操作和流程。

1. ray.init()

在PythonShell中&＃xff0c;使用ray.init()可以在本地启动ray&＃xff0c;包括Driver、HeadNode&＃xff08;Master&＃xff09;和若干Slave。

import ray
ray.init()


如果是直连已有的Ray集群&＃xff0c;只需要指定RedisServer的地址即可。

ray.init(redis_address&＃61;"")

本地启动Ray得到的输出如下&＃xff1a;

>>> ray.init()
Waiting for redis server at 127.0.0.1:58807 to respond...
Waiting for redis server at 127.0.0.1:23148 to respond...
Allowing the Plasma store to use up to 13.7439GB of memory.
Starting object store with directory /tmp and huge page support disabled
Starting local scheduler with 8 CPUs, 0 GPUs&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;
View the web UI at http://localhost:8888/notebooks/ray_ui62614.ipynb?token&＃61;7c253b0fd66fe41294d9f2c6739e3f002c1e76f6f59b99f5
&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;{&＃39;object_store_addresses&＃39;: [ObjectStoreAddress(name&＃61;&＃39;/tmp/plasma_store73540254&＃39;, manager_name&＃61;&＃39;/tmp/plasma_manager78072648&＃39;, manager_port&＃61;39874)], &＃39;redis_address&＃39;: &＃39;127.0.0.1:58807&＃39;, &＃39;local_scheduler_socket_names&＃39;: [&＃39;/tmp/scheduler98624129&＃39;], &＃39;webui_url&＃39;: &＃39;http://localhost:8888/notebooks/ray_ui62614.ipynb?token&＃61;7c253b0fd66fe41294d9f2c6739e3f002c1e76f6f59b99f5&＃39;, &＃39;node_ip_address&＃39;: &＃39;127.0.0.1&＃39;}

本地启动Ray时&＃xff0c;可以看到Ray的WebUI的访问地址。

2. ray.put()