本期推荐的Ray是由UC Berkeley RISELab推出的高性能分布式执行框架。

Ray是UC Berkeley大学 RISE lab（前AMP lab）开源的新一代分布式应用框架（刚发布的时候定位是高性能分布式计算框架，20年中修改定位为分布式应用框架），通过一套引擎解决复杂场景问题，通过动态计算及状态共享提高效率，实现研发、运行时、容灾一体化。

功能特性

Ray 包含以下用于加速机器学习工作负载的库：

• Tune：可扩展的超参数调整
• RLlib：可扩展的强化学习
• Train：分布式深度学习（测试版）
• Datasets：分布式数据加载和计算

以及用于将 ML 和分布式应用程序投入生产的库：

• Serve：可扩展和可编程服务
• Workflows：快速、持久的应用程序流程（alpha）

业务目标

Ray的定位是分布式应用框架，主要目标是使能分布式应用的开发和运行。

Ray的API让开发者可以轻松的在单个分布式应用中组合多个libraries，例如，Ray的tasks和Actors可能会call into 或called from在Ray上运行的分布式训练（e.g. torch.distributed）或者在线服务负载; 在这种场景下,Ray是作为一个“分布式胶水”系统，因为它提供通用API接口并且性能足以支撑许多不同工作负载类型。

系统设计

逻辑架构

关键概念

• Task：Ray 使任意函数能够在单独的 Python 工作者上异步执行。这些异步 Ray 函数称为“任务”。Ray 使任务能够根据 CPU、GPU 和自定义资源指定其资源需求。集群调度程序使用这些资源请求在集群中分发任务以进行并行执行。
• Object：在 Ray 中，任务和参与者在对象上创建和计算。我们将这些对象称为远程对象，因为它们可以存储在 Ray 集群中的任何位置，并且我们使用对象引用来引用它们。远程对象缓存在 Ray 的分布式共享内存 对象库中，集群中的每个节点都有一个对象库。在集群设置中，远程对象可以存在于一个或多个节点上，与谁拥有对象 ref(s) 无关。
• Actor：Actor 将 Ray API 从函数（任务）扩展到类。演员本质上是有状态的工人（或服务）。当一个新的actor被实例化时，一个新的worker被创建，并且actor的方法被调度在那个特定的worker上并且可以访问和改变那个worker的状态。与任务一样，actor 支持 CPU、GPU 和自定义资源要求。
• Placement Groups： 放置组允许用户以原子方式跨多个节点保留资源组（即，群调度）。然后可以使用它们来调度 Ray 任务和 Actor，它们尽可能靠近局部（PACK）或分散（SPREAD）。安置组通常用于帮派调度演员，但也支持任务。
• Environment Dependencies：当 Ray 在远程机器上执行任务和参与者时，它们的环境依赖项（例如，Python 包、本地文件、环境变量）必须可供代码运行。

集群设计

如上图所示，Ray集群包括一组同类的worker节点和一个集中的全局控制存储（GCS）实例。部分系统元数据由GCS管理，GCS是基于可插拔数据存储的服务，这些元数据也由worker本地缓存，例如Actor的地址。 GCS管理的元数据访问频率较低，但可能被群集中的大多数或所有worker使用，例如，群集的当前节点成员身份。这是为了确保GCS性能对于应用程序性能影响不大。

快速开始

并行执行 Python 函数

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

使用Ray的模型

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)]
[c.increment.remote() for c in counters]
futures = [c.read.remote() for c in counters]
print(ray.get(futures))

Tune快速入门

Tune是一个用于任何规模的超参数调整的库。

• 用不到 10 行代码启动多节点分布式超参数扫描。
• 支持任何深度学习框架，包括 PyTorch、PyTorch Lightning、TensorFlow 和 Keras。
• 使用TensorBoard可视化结果。
• 在可扩展的 SOTA 算法中进行选择，例如基于人口的训练 (PBT)、Vizier 的中值停止规则、HyperBand/ASHA。
• Tune 与许多优化库（如Facebook Ax、HyperOpt和贝叶斯优化）集成，使您能够透明地扩展它们。

要运行此示例，您需要安装以下内容：

$ pip install "ray[tune]"

此示例运行并行网格搜索以优化示例目标函数。

from ray import tune


def objective(step, alpha, beta):
    return (0.1 + alpha * step / 100)**(-1) + beta * 0.1


def training_function(config):
    # Hyperparameters
    alpha, beta = config["alpha"], config["beta"]
    for step in range(10):
        # Iterative training function - can be any arbitrary training procedure.
        intermediate_score = objective(step, alpha, beta)
        # Feed the score back back to Tune.
        tune.report(mean_loss=intermediate_score)


analysis = tune.run(
    training_function,
    config={
        "alpha": tune.grid_search([0.001, 0.01, 0.1]),
        "beta": tune.choice([1, 2, 3])
    })

print("Best config: ", analysis.get_best_config(metric="mean_loss", mode="min"))

# Get a dataframe for analyzing trial results.
df = analysis.results_df

如果安装了 TensorBoard，则自动可视化所有试验结果：

tensorboard --logdir ~/ray_results

RLlib快速入门

RLlib是一个工业级的强化学习 (RL) 库，建立在 Ray 之上。它为各种行业和研究应用程序提供了高可扩展性和统一的 API 。

$ pip install "ray[rllib]" tensorflow  # or torch

import gym
from ray.rllib.agents.ppo import PPOTrainer


# Define your problem using python and openAI's gym API:
class SimpleCorridor(gym.Env):
    """Corridor in which an agent must learn to move right to reach the exit.

    ---------------------
    | S | 1 | 2 | 3 | G |   S=start; G=goal; corridor_length=5
    ---------------------

    Possible actions to chose from are: 0=left; 1=right
    Observations are floats indicating the current field index, e.g. 0.0 for
    starting position, 1.0 for the field next to the starting position, etc..
    Rewards are -0.1 for all steps, except when reaching the goal (+1.0).
    """

    def __init__(self, config):
        self.end_pos = config["corridor_length"]
        self.cur_pos = 0
        self.action_space = gym.spaces.Discrete(2)  # left and right
        self.observation_space = gym.spaces.Box(0.0, self.end_pos, shape=(1,))

    def reset(self):
        """Resets the episode and returns the initial observation of the new one.
        """
        self.cur_pos = 0
        # Return initial observation.
        return [self.cur_pos]

    def step(self, action):
        """Takes a single step in the episode given `action`

        Returns:
            New observation, reward, done-flag, info-dict (empty).
        """
        # 向左走
        if action == 0 and self.cur_pos > 0:
            self.cur_pos -= 1
        # 向右走
        elif action == 1:
            self.cur_pos += 1
        # 结束时设置`done`标志
        done = self.cur_pos >= self.end_pos
        # 达到目标时+1，否则-1。
        reward = 1.0 if done else -0.1
        return [self.cur_pos], reward, done, {}


# 使用有 20 个字段（包括 S 和 G）
trainer = PPOTrainer(
    config={
        # Env class to use (here: our gym.Env sub-class from above).
        "env": SimpleCorridor,
        # Config dict to be passed to our custom env's constructor.
        "env_config": {
            # Use corridor with 20 fields (including S and G).
            "corridor_length": 20
        },
        # Parallelize environment rollouts.
        "num_workers": 3,
    })

# 训练 n 次迭代并报告结果（平均情节奖励）。
# 因为我们必须在环境中移动至少 19 次才能达到目标
# 每一步都给我们-0.1的奖励（除了最后的最后一步：+1.0），
# 我们可以期望达到 -0.1*18 + 1.0 = -0.8 的最佳情节奖励
for i in range(5):
    results = trainer.train()
    print(f"Iter: {i}; avg. reward={results['episode_reward_mean']}")

训练后，您可能希望在您的环境中执行动作计算（推理）。这是有关如何执行此操作的最小示例。

# 根据给定的环境观察执行推理（动作计算）。
# 请注意，我们在这里使用的环境略有不同（len 10 而不是 20），
# 但是，这应该仍然有效，因为代理已经（希望）学到了
# “永远走对了！”
env = SimpleCorridor({"corridor_length": 10})
# 获取初始观察值（应为：[0.0] 为起始位置）。
obs = env.reset()
done = False
total_reward = 0.0
# 播放一集。
while not done:
    # 给定当前观察值，计算单个动作
    # 从环境。
    action = trainer.compute_single_action(obs)
    # 在环境中应用计算的操作。
    obs, reward, done, info = env.step(action)
    # 总结奖励用于报告目的。
    total_reward += reward
# 报告结果。
print(f"Played 1 episode; total-reward={total_reward}")

Ray Serve快速开始

Ray Serve是一个基于 Ray 的可扩展模型服务库。它是：

• 与框架无关：使用相同的工具包来服务从使用 PyTorch 或 Tensorflow 和 Keras 等框架构建的深度学习模型到 Scikit-Learn 模型或任意业务逻辑的所有内容。
• Python First：在纯 Python 中以声明方式配置模型服务，无需 YAML 或 JSON 配置。
• 面向性能：打开批处理、流水线和 GPU 加速以提高模型的吞吐量。
• 组合原生：允许您通过将多个模型组合在一起来驱动单个预测来创建“模型管道”。
• 水平可扩展：随着您添加更多机器，Serve 可以线性扩展。使您的 ML 支持的服务能够处理不断增长的流量。

要运行此示例，您需要安装以下内容：

$ pip install scikit-learn
$ pip install "ray[serve]"

更多 Ray ML 库

• Ray AI Runtime (alpha)
• Distributed Scikit-learn / Joblib
• Distributed LightGBM on Ray
• Distributed multiprocessing.Pool
• Ray Collective Communication Lib
• Distributed PyTorch Lightning Training on Ray
• Using Ray with Pytorch Lightning
• Distributed XGBoost on Ray
• XGBoost-Ray with Dask
• XGBoost-Ray with Modin

—END—

开源协议：Apache-2.0 License