机器学习分布式框架Ray

简介: Ray是UC Berkeley RISELab推出的一个高性能分布式执行框架,它比Spark更具计算优势,部署简单,支持机器学习和深度学习的分布式训练。Ray包括节点(head和worker)、本地调度器、object store、全局调度器(GCS),用于处理各种分布式计算任务。它支持超参数调优(Ray Tune)、梯度下降(Ray SGD)、推理服务(Ray SERVE)等。安装简单,可通过`pip install ray`。使用时,利用`@ray.remote`装饰器将函数转换为分布式任务,通过`.remote`提交并用`ray.get`获取结果。5月更文挑战第15天

机器学习分布式框架Ray

1.什么是Ray

分布式计算框架大家一定都耳熟能详,诸如离线计算的Hadoop(map-reduce),spark, 流式计算的strom,Flink等。相对而言,这些计算框架都依赖于其他大数据组件,安装部署也相对复杂。

在python中,之前有分享过的Celery可以提供分布式的计算。今天和大家分享另外一个开源的分布式计算框架Ray。Ray是UC Berkeley RISELab新推出的高性能分布式执行框架,具有比Spark更优异的计算性能,而且部署和改造更简单,同时支持机器学习和深度学习的分布式训练,支持主流的深度学习框架(pytorch,tensorflow,keras等)

md-2021-09-07-17-46-27.png

2. Ray架构

Ray的架构参见最早发布的论文Ray: A Distributed Framework for Emerging AI Applications

md-2021-09-07-18-16-08.png

由上图可以Ray主要包括:

  • Node: 节点,主要是head和worker, head可以认为是Master,worker是执行任务的单元
    • 每个节点都有自己的本地调度器local scheduler
    • object store:一个内存对象存储,允许Node之间进行通信
  • scheduler: 有两个调度器,每个节点都有本地的调度器, 在提交任务时,Local Scheduler会判断是否需要提交给Global Scheduler分发给其他worker来执行。
  • GCS:全局状态控制记录了Ray中各种对象的状态信息,可以认为是meta数据,是Ray容错的保证

Ray适用于任何分布式计算的任务,包括分布式训练。笔者最近是用在大量的时间序列预测模型训练和在线预测上。

Ray目前库支持超参数调优Ray tune, 梯度下降Ray SGD,推理服务RaySERVE, 分布式数据Dataset以及分布式增强学习RLlib。还有其他第三方库,如下所示:

md-2021-09-07-19-12-34.png

3. 简单使用

3.1 安装部署

pip install --upgrade pip
# pip install ray
pip install ray == 1.6.0

# ImportError: cannot import name 'deep_mapping' from 'attr.validators'
# pip install attr == 19.1.0

3.2 单机使用

  • 简单例子
    Ray 通过@ray.remote装饰器使得函数变成可分布式调用的任务。通过函数名.remote方式进行提交任务,通过ray.get方式来获取任务返回值。单击情况下和多线程异步执行的方式类似。

      import time
      import ray
      ray.init(num_cpus = 4) # Specify this system has 4 CPUs.
    
      @ray.remote
      def do_some_work(x):
          time.sleep(1) # Replace this is with work you need to do.
          return x
    
      start = time.time()
      results = ray.get([do_some_work.remote(x) for x in range(4)])
      print("duration =", time.time() - start)
      print("results = ", results)
    
      # duration = 1.0107324123382568
      # results =  [0, 1, 2, 3]
    

    remote返回的对象的id 如ObjectRef(7f10737098927148ffffffff0100000001000000)。需要通过ray.get来获取实际的值, 需要注意的是ray.get是阻塞式的调用,不能[ray.get(do_some_work.remote(x)) for x in range(4)]

  • 注意小任务使用情况
    需要注意的是ray分布式计算在调度的时候需要发费额外的时间,如调度,进程间通信以及任务状态的更新等等,所以避免过小的任务。可以把小任务进行合并

      @ray.remote
      def tiny_work(x):
          time.sleep(0.0001) # Replace this is with work you need to do.
          return x
    
      start = time.time()
      result_ids = [tiny_work.remote(x) for x in range(100000)]
      results = ray.get(result_ids)
      print("duration =", time.time() - start)
    
  • ray.put
    ray.put() 把一个对象放到对象存储上,返回一个object id, 这个id可以在分布式机器上都可以调用,该操作为异步的。通过ray.get()可以是获取。

      num = ray.put(10)
      ray.get(num)
    
  • ray.wait
    如果任务返回多个结果,ray.get()会等所有结果都完成之后才会执行后续的操作。如果多个结果执行的耗时不同,此时短板在于最长的那个任务。

    这个时候可以采用ray.wait()方法,ray.wait()返回执行完毕的和未执行完毕的任务结果,执行完成的结果可以继续后续的操作

      import random
      @ray.remote
      def do_some_work(x):
          time.sleep(random.uniform(0, 4)) # Replace this is with work you need to do.
          return x
    
      def process_incremental(sum, result):
          time.sleep(1) # Replace this with some processing code.
          return sum + result
    
      start = time.time()
      result_ids = [do_some_work.remote(x) for x in range(4)]
      sum = 0
      while len(result_ids):
          done_id, result_ids = ray.wait(result_ids)
          sum = process_incremental(sum, ray.get(done_id[0]))
      print("duration =", time.time() - start, "\nresult = ", sum)
    
      # duration = 5.270821809768677 
      # result =  6
    

    md-2021-09-07-10-24-14.png

2.3 集群部署

Ray的架构遵循master-slave的模式。Head Node 可以认为是Master,其他的Node为worker。在集群部署时,Head Node需要首先启动ray start --head, 其他机器依次启动worker,注意需要指定head Node的地址确定关系,ray start --address 10.8.xx.3:6379

关闭服务,需要每一台机器执行 ray.stop

md-2021-09-07-11-17-49.png

# To start a head node.
#ray start --head --num-cpus=<NUM_CPUS> --num-gpus=<NUM_GPUS>
ray start --head --node-ip-address 10.8.xx.3 --port=6379


# To start a non-head node.
# ray start --address=<address> --num-cpus=<NUM_CPUS> --num-gpus=<NUM_GPUS>
ray start --address 10.8.xx.3:6379 --node-ip-address 10.8.xx.3 --num-cpus 10 --temp-dir={your temp path}

md-2021-09-07-13-37-40.png

  • 提交任务
    任何一台worker机器都可以提交任务, 先通过init连接Head Node就可以remote起来了。
      import ray
      ray.init(10.8.xx.3:6379)
    

3. 不同任务的例子

  • 任务依赖
    任务之间存在依赖关系,Ray和Spark一样也是通过生成DAG图的方式来确定依赖关系,确定可以并行跑的任务。如下图所示zeros是可以并行跑的。

    md-2021-09-07-10-23-42.png

``` python
import numpy as np
# Define two remote functions. Invocations of these functions create tasks
# that are executed remotely.

@ray.remote
def multiply(x, y):
    return np.dot(x, y)

@ray.remote
def zeros(size):
    return np.zeros(size)

# Start two tasks in parallel. These immediately return futures and the
# tasks are executed in the background.
x_id = zeros.remote((100, 100))
y_id = zeros.remote((100, 100))

# Start a third task. This will not be scheduled until the first two
# tasks have completed.
z_id = multiply.remote(x_id, y_id)

# Get the result. This will block until the third task completes.
z = ray.get(z_id)
print(z)
```
  • 有状态任务
    上面提到的任务都是无状态的(除依赖外),即任务之间都是无关系的。Ray也是支持有状态的任务成为Actor。常是在python class上加@ray.remote,ray会跟踪每个class内部状态的不同状态。

      @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)]
    
      # 不断的执行可以每个counter计数不断增加
      [c.increment.remote() for c in counters]
      futures = [c.read.remote() for c in counters]
      print(ray.get(futures))
      # [1, 1, 1, 1]
      # [11, 11, 11, 11]
    
  • map-reduce 任务
    map-reduce任务其实可以其他分布式任务是一样的。主要是各种聚合操作。Map-Reduce常规操作如下
    md-2021-09-07-13-55-07.png

  - word count例子见:https://github.com/ray-project/ray/blob/master/doc/examples/streaming/streaming.py

这里举一个简单的例子:
``` python
@ray.remote
def map(obj, f):
    return f(obj)
@ray.remote
def sum_results(*elements):
    return np.sum(elements)

items = list(range(100))
map_func = lambda i : i*2
remote_elements = [map.remote(i, map_func) for i in items]

# simple reduce
remote_final_sum = sum_results.remote(*remote_elements)
result = ray.get(remote_final_sum)

# tree reduce
intermediate_results = [sum_results.remote(
    *remote_elements[i * 20: (i + 1) * 20]) for i in range(5)]
remote_final_sum = sum_results.remote(*intermediate_results)
result = ray.get(remote_final_sum)

```
![md-2021-09-07-19-34-45.png](https://ucc.alicdn.com/pic/developer-ecology/abb7gqinvjggw_0b8ad35cdc774ab78d681c2c782ff884.png)
  • 训练模型如pytorch
    官网提供了Best Practices: Ray with PyTorch, 主要是下载训练/测试数据和训练多个模型(感觉不是很实用)。训练多个模型,可以进行参数融合。

    参见 https://docs.ray.io/en/latest/using-ray-with-pytorch.html

4. 总结

本文分享了高效的Python分布式计算框架Ray,希望对你有帮助。总结如下:

  • Ray是UC Berkeley RISELab新推出的高性能分布式执行框架, Spark也是伯克利出品的
  • Ray架构关键:两个调度器, Head和worker节点,GCS全局状态控制保证计算容错
  • Ray应用简单:@ray.remote把任务变成分布式任务, x.remote提交任务, get/wait获取结果
  • 集群不是:ray start
  • Ray支持多种任务:有依赖DAG,有状态Actor以及深度学习支持
  • 不断丰富的库:RaySERVE, RaySGD, RayTune, Ray data,rllib
相关实践学习
基于阿里云DeepGPU实例,用AI画唯美国风少女
本实验基于阿里云DeepGPU实例,使用aiacctorch加速stable-diffusion-webui,用AI画唯美国风少女,可提升性能至高至原性能的2.6倍。
目录
相关文章
|
6天前
|
机器学习/深度学习 分布式计算 算法
联邦学习是保障数据隐私的分布式机器学习方法
【6月更文挑战第13天】联邦学习是保障数据隐私的分布式机器学习方法,它在不暴露数据的情况下,通过在各设备上本地训练并由中心服务器协调,实现全局模型构建。联邦学习的优势在于保护隐私、提高训练效率和增强模型泛化。已应用于医疗、金融和物联网等领域。未来趋势包括更高效的数据隐私保护、提升可解释性和可靠性,以及与其他技术融合,有望在更多场景发挥潜力,推动机器学习发展。
17 4
|
8天前
|
负载均衡 监控 Go
使用Golang框架构建分布式系统
本文探讨了使用Golang构建分布式系统的方法。Golang因其高效、简洁的语法和并发支持成为理想的开发语言。文中列举了几个常用的Golang框架,如Echo、Gin、gRPC和NATS等,并强调了服务拆分、通信机制、负载均衡等构建分布式系统的关键要素。通过选择合适的框架,遵循需求分析、技术选型、服务设计等步骤,开发者可以构建出高性能、高可用和可扩展的系统。此外,文中还提供了一个使用gRPC和etcd的简单代码案例来说明实现过程。
26 4
|
8天前
|
分布式计算 负载均衡 并行计算
Python 分布式计算框架 PP (Parallel Python):集群模式下的实践探索
该文介绍了使用Parallel Python (PP) 在两台物理机上构建分布式计算集群的经验。PP是一个轻量级框架,旨在简化Python代码在多处理器系统和集群中的并行执行。文中通过设置子节点的IP、端口和密钥启动PP服务器,并在主节点创建PP实例进行负载均衡。实验使用官方的质数和计算示例,显示PP在集群模式下能有效利用多台机器的多核CPU,实现计算效率的显著提升。未来,作者计划进一步研究PP在更复杂任务和大规模集群中的应用潜力。
|
15天前
|
机器学习/深度学习 人工智能 分布式计算
PAI底层支持多种计算框架
PAI底层支持多种计算框架:
30 0
|
19天前
|
分布式计算 Hadoop 大数据
分布式计算框架比较:Hadoop、Spark 与 Flink
【5月更文挑战第31天】Hadoop是大数据处理的开创性框架,专注于大规模批量数据处理,具有高扩展性和容错性。然而,它在实时任务上表现不足。以下是一个简单的Hadoop MapReduce的WordCount程序示例,展示如何统计文本中单词出现次数。
69 0
|
22天前
|
分布式计算 网络协议 大数据
基于C++的分布式计算框架设计与实现
基于C++的分布式计算框架设计与实现
57 2
|
23天前
|
机器学习/深度学习 敏捷开发 测试技术
深入理解自动化测试:框架选择与实践挑战利用机器学习技术优化数据中心冷却系统
【5月更文挑战第27天】 在现代软件开发周期中,自动化测试已成为确保产品质量和加快市场投放的关键步骤。本文深入探讨了自动化测试的框架选择问题,并剖析了实施过程中面临的挑战及其解决方案。通过比较不同测试框架的特点,我们旨在为读者提供一套明确的指导原则,帮助他们根据项目需求做出恰当的技术决策。同时,文中还分享了实际案例和最佳实践,以期帮助开发团队克服实施自动化测试时可能遇到的障碍。
|
23天前
|
人工智能 监控 开发者
阿里云PAI发布DeepRec Extension,打造稳定高效的分布式训练,并宣布开源!
阿里云人工智能平台PAI正式发布自研的 DeepRec Extension(即 DeepRec 扩展),旨在以更低成本,更高效率进行稀疏模型的分布式训练。
|
30天前
|
SQL 分布式计算 Hadoop
Spark分布式内存计算框架
Spark分布式内存计算框架
53 0
|
1月前
|
NoSQL Java 关系型数据库
【Redis系列笔记】分布式锁
分布式锁:满足分布式系统或集群模式下多进程可见并且互斥的锁。 分布式锁的核心思想就是让大家都使用同一把锁,只要大家使用的是同一把锁,那么我们就能锁住线程,不让线程进行,让程序串行执行,这就是分布式锁的核心思路
420 2

热门文章

最新文章