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

Ray是UC Berkeley RISELab新推出的高性能分布式执行框架，它使用了和传统分布式计算系统不一样的架构和对分布式计算的抽象方式，具有比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节点多入和多出的特点。

二、系统架构

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

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

1. GlobalScheduler：Master上启动了一个全局调度器，用于接收本地调度器提交的任务，并将任务分发给合适的本地任务调度器执行。
2. RedisServer：Master上启动了一到多个RedisServer用于保存分布式任务的状态信息（ControlState），包括对象机器的映射、任务描述、任务debug信息等。
3. LocalScheduler：每个Slave上启动了一个本地调度器，用于提交任务到全局调度器，以及分配任务给当前机器的Worker进程。
4. Worker：每个Slave上可以启动多个Worker进程执行分布式任务，并将计算结果存储到ObjectSto