高性能分布式执行框架——Ray

Ray是UC BerkeleyRISELab新生产的高性能分布式执行框架,它应用了和价值观分布式总结系统不平等的架构和对分布式统计的抽象模式,具有比Spark更美妙的测算性能。

Ray近日还处在实验室阶段,最新版本为0.2.2版本。即使Ray自称是面向AI应用的分布式总计框架,不过它的架构具有通用的分布式统计抽象。本文对Ray举办简要的牵线,帮助我们更快地了然Ray是什么样,如有描述不当的地点,欢迎不吝指正。

一、简单开首

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

# 导入ray,并初始化执行环境
import ray
ray.init()

# 定义ray remote函数
@ray.remote
def hello():
    return "Hello world !"

# 异步执行remote函数,返回结果id
object_id = hello.remote()

# 同步获取计算结果
hello = ray.get(object_id)

# 输出计算结果
print hello

在Ray里,通过Python注解@ray.remote定义remote函数。使用此表明声明的函数都会自带一个默认的不二法门remote,通过此措施发起的函数调用都是以提交分布式任务的法子异步执行的,函数的返回值是一个对象id,使用ray.get置于操作可以共同获取该id对应的对象。熟知Java里的Future机制的话对此相应并不生疏,或许会有人疑惑这和平日的异步函数调用没什么大的界别,不过这里最大的差异是,函数hello是分布式异步执行的。

remote函数是Ray分布式统计抽象中的主旨概念,通过它开发者拥有了动态定制总括依赖(任务DAG)的力量。比如:

@ray.remote
def A():
    return "A"

@ray.remote
def B():
    return "B"

@ray.remote
def C(a, b):
    return "C"

a_id = A.remote()
b_id = B.remote()
c_id = C.remote(a_id, b_id)
print ray.get(c_id)

事例代码中,对函数A、B的调用是一心并行执行的,然则对函数C的调用看重于A、B函数的回到结果。Ray可以确保函数C需要等待A、B函数的结果的确总结出来后才会实施。虽然将函数A、B、C类比为DAG的节点的话,那么DAG的边就是函数C参数对函数A、B总计结果的依靠,自由的函数调用情势允许Ray能够肆意地定制DAG的布局和测算看重关系。此外,提及一点的是Python的函数可以定义函数具有三个重返值,这也使得Python的函数更天然具备了DAG节点多入和多出的特性。

Java 1

二、系统架构

Ray是使用什么的架构对分布式统计做出如上抽象的呢,一下交付了Ray的系统架构(来自Ray散文,参考文献1)。

Java 2

用作分布式总括系统,Ray仍然遵照了卓绝的Master-Slave的计划性:Master负责全局协调和情景维护,Slave执行分布式统计任务。可是和观念的分布式总结系统不同的是,Ray使用了错落任务调度的笔触。在集群部署格局下,Ray启动了以下重点零部件:

  1. GlobalScheduler:Master上启动了一个大局调度器,用于收纳本地调度器提交的职责,并将任务分发给方便的本土任务调度器执行。
  2. RedisServer:Master上启动了一到三个RedisServer用于保存分布式任务的意况音讯(ControlState),包括对象机器的照射、任务描述、任务debug信息等。
  3. LocalScheduler:每个Slave上启动了一个本土调度器,用于提交任务到全局调度器,以及分配任务给当下机械的Worker进程。
  4. Worker:每个Slave上可以启动六个Worker进程执行分布式任务,并将总计结果存储到ObjectStore。
  5. ObjectStoreJava,:每个Slave上启动了一个ObjectStore存储只读数据对象,Worker可以经过共享内存的不二法门访问这一个目标数据,这样可以使得地减小内存拷贝和目标连串化成本。ObjectStore底层由Apache
    Arrow实现。
  6. Plasma:每个Slave上的ObjectStore都由一个名为Plasma的目标管理器举行保管,它能够在Worker访问本地ObjectStore上不设有的长途数据对象时,主动拉取此外Slave上的目的数据到当下机械。

亟需表明的是,Ray的舆论中提及,全局调度器可以启动一到六个,而眼前Ray的兑现文档里探讨的情节都是基于一个大局调度器的景观。我怀疑可能是Ray尚在建设中,一些机制还未周密,后续读者可以小心此处的底细变化。

Ray的天职也是由此类似Spark中Driver的定义的情势展开付出的,有所不同的是:

  1. 斯帕克(Spark)的Driver提交的是任务DAG,一旦付出则不足改变。
  2. 而Ray提交的是更细粒度的remote
    function,任务DAG倚重关系由函数倚重关系自由定制。

舆论给出的架构图里从未画出Driver的定义,由此我在其基础上做了有些改动和壮大。

Java 3

Ray的Driver节点和和Slave节点启动的零件几乎一样,然而却有以下分别:

  1. Driver上的工作过程DriverProcess一般只有一个,即用户启动的PythonShell。Slave可以依照需要成立四个WorkerProcess。
  2. Driver只可以交给任务,却无法收到来自全局调度器分配的职责。Slave可以交到任务,也得以吸收全局调度器分配的任务。
  3. Driver可以积极绕过全局调度器给Slave发送Actor调用任务(此处设计是否合理尚不商量)。Slave只可以收到全局调度器分配的精打细算任务。

三、大旨操作

依据以上架构,咱们大概商量一下Ray中举足轻重的操作和流程。

1. ray.init()

在PythonShell中,使用ray.init()能够在地点启动ray,包括Driver、HeadNode(Master)和几何Slave。

import ray
ray.init()

假诺是直连已有些Ray集群,只需要指定RedisServer的地方即可。

ray.init(redis_address="<redis-address>")

本地启动Ray拿到的出口如下:

>>> 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

======================================================================
View the web UI at http://localhost:8888/notebooks/ray_ui62614.ipynb?token=7c253b0fd66fe41294d9f2c6739e3f002c1e76f6f59b99f5
======================================================================

{'object_store_addresses': [ObjectStoreAddress(name='/tmp/plasma_store73540254', manager_name='/tmp/plasma_manager78072648', manager_port=39874)], 'redis_address': '127.0.0.1:58807', 'local_scheduler_socket_names': ['/tmp/scheduler98624129'], 'webui_url': 'http://localhost:8888/notebooks/ray_ui62614.ipynb?token=7c253b0fd66fe41294d9f2c6739e3f002c1e76f6f59b99f5', 'node_ip_address': '127.0.0.1'}
>>> 

当地启动Ray时,可以看看Ray的WebUI的造访地址。

2. ray.put()

使用ray.put()可以将Python对象存入本地ObjectStore,并且异步重临一个唯一的ObjectID。通过该ID,Ray可以访问集群中任一个节点上的靶子(远程对象通过查阅Master的对象表得到)。

对象一旦存入ObjectStore便不可变更,Ray的remote函数可以将一贯将该对象的ID作为参数传入。使用ObjectID作为remote函数参数,可以有效地缩减函数参数的写ObjectStore的次数。

@ray.remote
def f(x):
    pass

x = "hello"

# 对象x往ObjectStore拷贝里10次
[f.remote(x) for _ in range(10)]

# 对象x仅往ObjectStore拷贝1次
x_id = ray.put(x)
[f.remote(x_id) for _ in range(10)]

3. ray.get()

使用ray.get()可以由此ObjectID获取ObjectStore内的靶子并将之转换为Python对象。对于数组类型的对象,Ray使用共享内存机制裁减数量的正片成本。而对于其余对象则需要将数据从ObjectStore拷贝到过程的堆内存中。

一经调用ray.get()操作时,对象尚未创造好,则get操作会阻塞,直到对象成立完成后赶回。get操作的首要流程如下:

  1. Driver或者Worker进程首先到ObjectStore内请求ObjectID对应的目的数据。
  2. 假设地点ObjectStore没有相应的目的数据,本地对象管理器Plasma会检讨Master上的靶子表查看对象是否存储另外节点的ObjectStore。
  3. 如若目的数据在另外节点的ObjectStore内,Plasma会发送网络请求将目的数据拉到本地ObjectStore。
  4. 若果目的数据还一直不开创好,Master会在对象创造完成后通报请求的Plasma读取。
  5. 设若目的数据现已被有着的ObjectStore移除(被LRU策略删除),本地调度器会基于职责血缘关系执行对象的再次成立工作。
  6. 尽管目的数据在本地ObjectStore可用,Driver或者Worker进程会通过共享内存的章程直接将对象内存区域映射到自己的经过地址空间中,并反体系化为Python对象。

另外,ray.get()可以两回性读取四个对象的数据:

result_ids = [ray.put(i) for i in range(10)]
ray.get(result_ids)  # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

4. @ray.remote

Ray中应用注脚@ray.remote可以申明一个remote
function。remote函数时Ray的骨干任务调度单元,remote函数定义后会立刻被序列化存储到RedisServer中,并且分配了一个唯一的ID,那样就保证了集群的有所节点都足以观看这一个函数的概念。

不过,这样对remote函数定义有了一个神秘的要求,即remote函数内要是调用了其余的用户函数,则必须提前定义,否则remote函数不可能找到相应的函数定义内容。

remote函数内也可以调用其余的remote函数,Driver和Slave每一遍调用remote函数时,其实都是向集群提交了一个盘算任务,从这边也能够见见Ray的分布式总括的自由性。

Ray中调用remote函数的重要流程如下:

  1. 调用remote函数时,首先会创建一个职责指标,它蕴含了函数的ID、参数的ID或者值(Python的主导对象直接传值,复杂对象会先经过ray.put()操作存入ObjectStore然后归来ObjectID)、函数再次回到值对象的ID。
  2. 职责目的被发送到本地调度器。
  3. 当地调度器决定任务目标是在当地调度仍然发送给全局调度器。假诺任务目的的倚重性(参数)在该地的ObejctStore已经存在且当地的CPU和GPU总结资源充沛,那么地点调度器将任务分配给地点的WorkerProcess执行。否则,任务目标被发送给全局调度器并蕴藏到任务表(TaskTable)中,全局调度器依照当下的任务意况信息决定将任务发给集群中的某一个本地调度器。
  4. 地面调度器收到任务目标后(来自当地的任务如故全局调度分配的任务),会将其放入一个任务队列中,等待总括资源和当地依赖知足后分配给WorkerProcess执行。
  5. Worker收到任务目的后实施该任务,并将函数重回值存入ObjectStore,并革新Master的目的表(ObjectTable)音讯。

@ray.remote讲明有一个参数num_return_vals用来表明remote函数的再次回到值个数,基于此实现remote函数的多重临值机制。

@ray.remote(num_return_vals=2)
def f():
    return 1, 2

x_id, y_id = f.remote()
ray.get(x_id)  # 1
ray.get(y_id)  # 2

@ray.remote注脚的另一个参数num_gpus能够为天职指定GPU的资源。使用内置函数ray.get_gpu_ids()可以博得当前任务可以采用的GPU音讯。

@ray.remote(num_gpus=1)
def gpu_method():
    return "This function is allowed to use GPUs {}.".format(ray.get_gpu_ids())

5. ray.wait()

ray.wait()操作帮忙批量的职责等待,基于此可以实现两次性取得七个ObjectID对应的数量。

# 启动5个remote函数调用任务
results = [f.remote(i) for i in range(5)]
# 阻塞等待4个任务完成,超时时间为2.5s
ready_ids, remaining_ids = ray.wait(results, num_returns=4, timeout=2500)

上述例子中,results包含了5个ObjectID,使用ray.wait操作可以从来守候有4个任务完成后重回,并将成功的数目对象放在首个list类型重回值内,未形成的ObjectID放在第二个list再次回到值内。即使设置了晚点时间,那么在逾期时间结束后仍未等到预期的再次回到值个数,则已逾期完成时的重回值为准。

6. ray.error_info()

使用ray.error_info()可以获取任务履行时发出的错误音讯。

>>> import time
>>> @ray.remote
>>> def f():
>>>     time.sleep(5)
>>>     raise Exception("This task failed!!")
>>> f.remote()
Remote function __main__.f failed with:

Traceback (most recent call last):
  File "<stdin>", line 4, in f
Exception: This task failed!!


  You can inspect errors by running

      ray.error_info()

  If this driver is hanging, start a new one with

      ray.init(redis_address="127.0.0.1:65452")
>>> ray.error_info()
[{'type': 'task', 'message': 'Remote function \x1b[31m__main__.f\x1b[39m failed with:\n\nTraceback (most recent call last):\n  File "<stdin>", line 4, in f\nException: This task failed!!\n', 'data': '{\'function_id\': "Hm\\xde\\x93\'\\x91\\xce\\x13ld\\xf4O\\xd7\\xce\\xc2\\xe1\\x151\\x1e3", \'function_name\': u\'__main__.f\'}'}]

7. Actor

Ray的remote函数只可以处理无状态的乘除需求,有气象的乘除需求需要使用Ray的Actor实现。在Python的class定义前应用@ray.remote可以表明Actor。

@ray.remote
class Counter(object):
    def __init__(self):
        self.value = 0

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

行使如下情势开创Actor对象。

a1 = Counter.remote()
a2 = Counter.remote()

Ray创设Actor的流水线为:

  1. Master采用一个Slave,并将Actor制造任务分发给它的地方调度器。
  2. 创办Actor对象,并举办它的构造函数。

从流程可以观望,Actor对象的创导时互动的。

经过调用Actor对象的方法应用Actor。

a1.increment.remote()  # ray.get returns 1
a2.increment.remote()  # ray.get returns 1

调用Actor对象的办法的流水线为:

  1. 率先创设一个职责。
  2. 该任务被Driver直接分配到创制该Actor对应的地头执行器执行,这多少个操作绕开了大局调度器(Worker是否也可以使用Actor直接分配任务尚存疑问)。
  3. 回去Actor方法调用结果的ObjectID。

为了保证Actor状态的一致性,对同一个Actor的章程调用是串行执行的。

四、安装Ray

假定只是使用Ray,可以应用如下命令直接设置。

pip intall ray

尽管急需编译Ray的新颖源码举办安装,依照如下步骤举办(马克斯OS):

# 更新编译依赖包
brew update
brew install cmake pkg-config automake autoconf libtool boost wget
pip install numpy cloudpickle funcsigs click colorama psutil redis flatbuffers cython --ignore-installed six
# 下载源码编译安装
git clone https://github.com/ray-project/ray.git
cd ray/python
python setup.py install
# 测试
python test/runtest.py

# 安装WebUI需要的库[可选]
pip install jupyter ipywidgets bokeh

# 编译Ray文档[可选]
cd ray/doc
pip install -r requirements-doc.txt
make html
open _build/html/index.html

自家在MacOS上安装jupyter时,碰到了Python的setuptools库无法进步的意况,原因是MacOS的安全性设置问题,能够应用如下形式缓解:

  1. 重启电脑,启动时按住Command+R跻身Mac珍视模式。
  2. 开辟命令行,输入指令csrutils disable关闭系统安全策略。
  3. 重启电脑,继续安装jupyter。
  4. 设置到位后,重复如上的点子执行csrutils enable,再一次重启即可。

进去PythonShell,输入代码本地启动Ray:

import ray
ray.init()

浏览器内开辟WebUI界面如下:

Java 4

参考资料

  1. Ray论文:Real-Time Machine Learning: The Missing
    Pieces
  2. Ray开发手册:http://ray.readthedocs.io/en/latest/index.html
  3. Ray源代码:https://github.com/ray-project/ray

相关文章