机器学习分布式框架Ray

机器学习分布式框架Ray

作者：wedo实验君 2021-09-09 15:45:17

人工智能

机器学习

分布式 Ray是UC Berkeley RISELab新推出的高性能分布式执行框架，具有比Spark更优异的计算性能，而且部署和改造更简单，同时支持机器学习和深度学习的分布式训练，支持主流的深度学习框架。

创新互联建站专注于企业成都营销网站建设、网站重做改版、鹿城网站定制设计、自适应品牌网站建设、H5高端网站建设、商城网站建设、集团公司官网建设、外贸营销网站建设、高端网站制作、响应式网页设计等建站业务，价格优惠性价比高，为鹿城等各大城市提供网站开发制作服务。

 [[422916]]

Python中文社区 （ID:python-china）

1.什么是Ray

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

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

  •  https://github.com/ray-project/ray

2. Ray架构

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

由上图可知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。还有其他第三方库，如下所示：

3. 简单使用

3.1 安装部署 

  
 
 
 
    
   
  
  
  
  1. pip install --upgrade pip  
    2.    
  
  
  
  2. # pip install ray  
    3.    
  
  
  
  3. pip install ray == 1.6.0  
    4.    
  
  
  
  4. # ImportError: cannot import name 'deep_mapping' from 'attr.validators'  
    5.    
  
  
  
  5. # pip install attr == 19.1.0 
    6.

3.2 单机使用

  •  简单例子 Ray 通过@ray.remote装饰器使得函数变成可分布式调用的任务。通过函数名.remote方式进行提交任务，通过ray.get方式来获取任务返回值。单击情况下和多线程异步执行的方式类似。 
  
 
 
 
    
   
  
  
  
  1.  import time  
    2.    
  
  
  
  2.   import ray  
    3.    
  
  
  
  3.   ray.init(num_cpus = 4) # Specify this system has 4 CPUs.  
    4.    
  
  
  
  4.   @ray.remote  
    5.    
  
  
  
  5.   def do_some_work(x):  
    6.    
  
  
  
  6.       time.sleep(1) # Replace this is with work you need to do.  
    7.    
  
  
  
  7.       return x  
    8.    
  
  
  
  8.   start = time.time()  
    9.    
  
  
  
  9.   results = ray.get([do_some_work.remote(x) for x in range(4)])  
    10.    
  
  
  
  10.   print("duration =", time.time() - start)  
    11.    
  
  
  
  11.   print("results = ", results)   
    12.    
  
  
  
  12.   # duration = 1.0107324123382568  
    13.    
  
  
  
  13.   # results =  [0, 1, 2, 3] 
    14.

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

  •  注意小任务使用情况 需要注意的是ray分布式计算在调度的时候需要发费额外的时间，如调度，进程间通信以及任务状态的更新等等，所以避免过小的任务。可以把小任务进行合并   
  
 
 
 
    
   
  
  
  
  1. @ray.remote  
    2.    
  
  
  
  2.    def tiny_work(x):  
    3.    
  
  
  
  3.        time.sleep(0.0001) # Replace this is with work you need to do.  
    4.    
  
  
  
  4.        return x  
    5.    
  
  
  
  5.    start = time.time()  
    6.    
  
  
  
  6.    result_ids = [tiny_work.remote(x) for x in range(100000)]  
    7.    
  
  
  
  7.    results = ray.get(result_ids)  
    8.    
  
  
  
  8.    print("duration =", time.time() - start) 
    9.
  •  ray.put ray.put() 把一个对象放到对象存储上，返回一个object id， 这个id可以在分布式机器上都可以调用，该操作为异步的。通过ray.get()可以是获取。 
  
 
 
 
    
   
  
  
  
  1. num = ray.put(10)  
    2.    
  
  
  
  2. ray.get(num) 
    3.
  •  ray.wait 如果任务返回多个结果，ray.get()会等所有结果都完成之后才会执行后续的操作。如果多个结果执行的耗时不同，此时短板在于最长的那个任务。

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

  
 
 
 
    
   
  
  
  
  1. import random  
    2.    
  
  
  
  2.    @ray.remote  
    3.    
  
  
  
  3.    def do_some_work(x):  
    4.    
  
  
  
  4.        time.sleep(random.uniform(0, 4)) # Replace this is with work you need to do.  
    5.    
  
  
  
  5.        return   
    6.    
  
  
  
  6.    def process_incremental(sum, result):  
    7.    
  
  
  
  7.        time.sleep(1) # Replace this with some processing code.  
    8.    
  
  
  
  8.        return sum + result  
    9.    
  
  
  
  9.    start = time.time()  
    10.    
  
  
  
  10.    result_ids = [do_some_work.remote(x) for x in range(4)]  
    11.    
  
  
  
  11.    sum = 0  
    12.    
  
  
  
  12.    while len(result_ids):  
    13.    
  
  
  
  13.        done_id, result_ids = ray.wait(result_ids)  
    14.    
  
  
  
  14.        sum = process_incremental(sum, ray.get(done_id[0])) 
    15.    
  
  
  
  15.     print("duration =", time.time() - start, "\nresult = ", sum)  
    16.    
  
  
  
  16.    # duration = 5.270821809768677   
    17.    
  
  
  
  17.    # result =  6    
    18.

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

 

  
 
 
 
    
   
  
  
  
  1. # To start a head node.  
    2.    
  
  
  
  2. #ray start --head --num-cpus= --num-gpus=  
    3.    
  
  
  
  3. ray start --head --node-ip-address 10.8.xx.3 --port=6379  
    4.    
  
  
  
  4. # To start a non-head node.  
    5.    
  
  
  
  5. # ray start --address=
     --num-cpus= --num-gpus=  
    6.    
  
  
  
  6. ray start --address 10.8.xx.3:6379 --node-ip-address 10.8.xx.3 --num-cpus 10 --temp-dir={your temp path} 
    7.

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

3. 不同任务的例子

  •  任务依赖 任务之间存在依赖关系，Ray和Spark一样也是通过生成DAG图的方式来确定依赖关系，确定可以并行跑的任务。如下图所示zeros是可以并行跑的。 
  
 
 
 
    
   
  
  
  
  1. import numpy as np  
    2.    
  
  
  
  2. # Define two remote functions. Invocations of these functions create tasks  
    3.    
  
  
  
  3. # that are executed remotely.  
    4.    
  
  
  
  4. @ray.remote  
    5.    
  
  
  
  5. def multiply(x, y):  
    6.    
  
  
  
  6.     return np.dot(x, y)  
    7.    
  
  
  
  7. @ray.remote  
    8.    
  
  
  
  8. def zeros(size):  
    9.    
  
  
  
  9.     return np.zeros(size)  
    10.    
  
  
  
  10. # Start two tasks in parallel. These immediately return futures and the  
    11.    
  
  
  
  11. # tasks are executed in the background.  
    12.    
  
  
  
  12. x_id = zeros.remote((100, 100))  
    13.    
  
  
  
  13. y_id = zeros.remote((100, 100))  
    14.    
  
  
  
  14. # Start a third task. This will not be scheduled until the first two  
    15.    
  
  
  
  15. # tasks have completed.  
    16.    
  
  
  
  16. z_id = multiply.remote(x_id, y_id)  
    17.    
  
  
  
  17. # Get the result. This will block until the third task completes.  
    18.    
  
  
  
  18. z = ray.get(z_id)  
    19.    
  
  
  
  19. print(z)     
    20.

  •  有状态任务 上面提到的任务都是无状态的(除依赖外)，即任务之间都是无关系的。Ray也是支持有状态的任务成为Actor。常是在python class上加@ray.remote，ray会跟踪每个class内部状态的不同状态。 
  
 
 
 
    
   
  
  
  
  1. @ray.remote  
    2.    
  
  
  
  2.  class Counter(object):  
    3.    
  
  
  
  3.      def __init__(self):  
    4.    
  
  
  
  4.          self.n = 0  
    5.    
  
  
  
  5.      def increment(self):  
    6.    
  
  
  
  6.          self.n += 1  
    7.    
  
  
  
  7.      def read(self):  
    8.    
  
  
  
  8.          return self.n  
    9.    
  
  
  
  9.  counters = [Counter.remote() for i in range(4)]  
    10.    
  
  
  
  10.  # 不断的执行可以每个counter计数不断增加  
    11.    
  
  
  
  11.  [c.increment.remote() for c in counters]  
    12.    
  
  
  
  12.  futures = [c.read.remote() for c in counters]  
    13.    
  
  
  
  13.  print(ray.get(futures))  
    14.    
  
  
  
  14.  # [1, 1, 1, 1]  
    15.    
  
  
  
  15.  # [11, 11, 11, 11] 
    16.
  • map-reduce 任务 map-reduce任务其实可以其他分布式任务是一样的。主要是各种聚合操作。Map-Reduce常规操作如下  

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

         这里举一个简单的例子：   

  
 
 
 
    
   
  
  
  
  1. @ray.remote  
    2.    
  
  
  
  2.    def map(obj, f):  
    3.    
  
  
  
  3.        return f(obj)  
    4.    
  
  
  
  4.    @ray.remote  
    5.    
  
  
  
  5.    def sum_results(*elements):  
    6.    
  
  
  
  6.        return np.sum(elements)  
    7.    
  
  
  
  7.    items = list(range(100))  
    8.    
  
  
  
  8.    map_func = lambda i : i*2  
    9.    
  
  
  
  9.    remote_elements = [map.remote(i, map_func) for i in items]  
    10.    
  
  
  
  10.    # simple reduce  
    11.    
  
  
  
  11.    remote_final_sum = sum_results.remote(*remote_elements)  
    12.    
  
  
  
  12.    result = ray.get(remote_final_sum) 
    13.    
  
  
  
  13.    # tree reduce  
    14.    
  
  
  
  14.    intermediate_results = [sum_results.remote(  
    15.    
  
  
  
  15.        *remote_elements[i * 20: (i + 1) * 20]) for i in range(5)]  
    16.    
  
  
  
  16.    remote_final_sum = sum_results.remote(*intermediate_results)  
    17.    
  
  
  
  17.    result = ray.get(remote_final_sum)     
    18.

  •  训练模型如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 

分享文章：机器学习分布式框架Ray
URL网址：http://www.mswzjz.com/qtweb/news24/168674.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

网站营销知识

分类信息网

云服务器　　 　成都做商城网站　　 　腾讯免备案主机　　 　高防机房　　 　营山网站建设　　 　成都网站设计　　 　腾讯云免备案空间　　 　成都铁科精智测控　　 　成都网络营销公司　　 　成都小程序开发　　 　宣威网站建设　　 　网站制作　　 　雪茄烟　　 　校园网站建设方案　　 　无边发光字　　 　高端品牌网站建设　　 　成都网站建设公司　　 　成都网络营销　　 　品牌网站建设　　 　成都翻译公司