Ray 是什么

Ray 是一个开源的分布式计算框架，旨在简化在多种计算资源上运行分布式应用程序的过程。Ray提供了一套简单直观的 API，使得开发者能够轻松地构建、部署和管理大规模的分布式应用程序。基于此分布式计算框架，Ray 提供了诸多面向数据科学和人工智能的上层库。

Ray Core

Ray 的分布式计算底盘，提供了 low level 的编程 API，可以让用户方便的把单机程序转为分布式计算任务。
特点：

• 简单易用：分布式运行的细节被很好的封装，用户无需太多的分布式经验即可上手。
• 代码侵入低：代码改造容易。
• 扩展性好：改造后的 Ray 代码可以从 local 的单一节点扩展至千级别节点。

相关概念：

• Task：面向函数（Function）的接口，用于定义一个函数，该函数可以在集群中分布式地执行。
• Actor：面向类（Class）的接口，用于定义一个类，该类可以在集群中分布式地实例化并执行。
• Object：分布式的不可变（Immutable）对象，用于在分布式的 Task 和 Actor 之间传递数据。

Ray AI Libraries

基于 Ray Core 实现的面向数据可选和人工智能的库，可实现数据读取、加工、训练、推理等一站式 pipeline。

• Data：提供数据加载和转换功能。
• Train：基于 Ray 的分布式模型框架，支持 TF/PyTorch/Horovod，或者 XGB/LightGBM 等训练工具。
• Tune：可扩展的超参数调优，以优化模型性能。
• Serve：用于部署模型进行在线推理服务。
• RLlib：可扩展的分布式强化学习库。

Ray 可以做什么

1. 机器学习：Ray提供了一套完整的机器学习库，包括用于训练和部署机器学习模型的工具。例如，Ray的RLlib库是一个用于强化学习的开源库，可以轻松地在Ray上进行大规模的强化学习训练。
2. 数据处理：Ray可以用于大规模的数据处理任务，如数据清洗、数据转换、数据分析等。Ray的Modin库是一个用于大规模数据处理的开源库，它可以将Pandas的数据处理操作分布到多个节点上，从而实现快速的数据处理。
3. 分布式系统：Ray提供了一套简单的API，可以用于构建和运行分布式系统。例如，Ray的Serve库是一个用于构建和运行分布式服务的开源库，可以轻松地在Ray上部署和扩展服务。
4. 高性能计算：Ray的分布式计算能力可以用于各种高性能计算任务，如科学计算、模拟、优化等。
5. 实时计算：用于实时计算任务，如实时数据分析、实时机器学习等。Ray的实时计算能力主要来自于Actor模型和分布式任务调度能力。

架构

Ray 集群是一组Worker节点和一个Head节点组成。Ray 集群可以是固定大小的，也可以根据在集群上运行的应用程序负载进。请参考官网Ray Clusters Overview

• Head node：集群的控制中心，负责管理整个集群的状态、调度任务以及分配资源。
  • Global Control Store：全局控制存储，用于存储集群状态和元数据信息
  • AutoScaler：自动伸缩器，根据当前的工作负载自动增加或减少工作节点的数量
  • Raylet：负责任务调度和资源管理
• Worker Node：执行实际计算任务的节点

Ray Cluster部署后，便可以在Ray Cluster上提交作业。提交作业有两种方式，具体使用方式可以参考RayCluster快速入门和RayJob快速入门。

主要特点

1. 易用性：Ray提供了一套简单的API，使得开发者可以轻松地编写分布式应用程序。
2. 性能：Ray设计了一种新的调度算法，可以在大规模集群上实现高效的任务调度。
3. 灵活性：Ray可以支持各种不同的计算模型，包括批处理、流处理、图计算等。
4. 可扩展性：Ray可以在数千个节点的集群上运行，支持大规模的计算任务。
5. 集成性：Ray可以与其他流行的数据处理和机器学习框架（如TensorFlow、PyTorch等）无缝集成。

如何使用

Ray的核心API包括以下几个关键概念：

1. ray.init(): 初始化Ray集群。你可以传递一个地址来连接到一个已存在的集群。
2. @ray.remote: 将函数转换为任务，将类转换为actor（演员）。
3. ray.put(): 将值放入Ray的对象存储。
4. ray.get(): 从对象存储中获取值。返回你之前放入的值，或者是任务或actor计算的结果。
5. .remote(): 在Ray集群上运行actor方法或任务，并用于实例化actor。
6. ray.wait(): 返回两个对象引用列表，一个包含已完成的任务，另一个包含未完成的任务。

示例：
通过一个简单的例子来理解Ray是如何工作的。假设有一个数据库，想要从数据库中检索数据。以下是使用Ray进行并行化处理的示例：

import ray
import time

start = time.time()
# 模拟数据库的Python列表
database = ["Learning", "Ray", "Flexible", "Distributed", "Python", "for", "Machine", "Learning"]

# 使用ray.remote装饰器将函数转换为Ray任务
@ray.remote
def retrieve_task(item):
    time.sleep(item / 10.)
    return item, database[item]

# 启动Ray任务并收集它们的引用
object_references = [retrieve_task.remote(item) for item in range(5)]

# 使用ray.get()获取任务结果
data = ray.get(object_references)

# 打印获取数据所需的时间
print(f'Runtime: {time.time() - start:.2f} seconds, data:')
print(*data, sep="\n")