江苏龙网

  1. 主页 > 生活百科 > >

使用Ray轻松进行Python分布式计算

Python

即使是具有多个CPU核心的单处理器计算机（处理器具有一个或多个核心，计算机具有一个或多个处理器） ， 也会给人一种能够同时运行多个任务的错觉 。当我们拥有多个处理器时，就可以真正以并行的方式执行计算 。
一、并行计算与分布式计算的区别并行计算在现代计算中非常有用，几乎是必需的，目的是实现最大性能 。开发者将运行时间较长的计算任务分成较小的块，并将其分配给不同的处理器 。这种策略使开发者能够在相同的时间内进行更多的计算 。对于构建基于GUI的应用程序，总是需要对系统进行并行设计 ， 以便一个线程可以保持可用状态以更新GUI并响应用户输入 。
并行计算和分布式计算的区别在于，对于并行计算，多个处理器位于同一主板上 。分布式计算则使用多台计算机同时解决问题 。现代分布式系统能够在网络（局域网/广域网）上进行通信 。分布式计算的优点在于其价格和可扩展性 。如果开发者需要更多的计算能力，那么可以很轻松地添加更多的计算机 。
从根本上讲，并行计算和分布式计算的架构非常相似 。主要区别在于分布式计算使用的是分布式内存空间 ， 而不是共享内存空间 。它具有能够为开发者的应用程序提供统一逻辑（而不是物理）内存空间的软件层，可以帮助开发者将为并行计算编写的代码应用于分布式计算 。
在本文中将介绍如何使用开源Python/ target=_blank class=infotextkey>Python库Ray来帮助开发者进行并行和分布式计算，Ray将Pythonic函数和类转换为分布式设置中的任务和角色 。本文将只介绍函数的示例 ， 但是类的概念非常相似 。
二、使用pip安装Ray

使用Ray轻松进行Python分布式计算

文章插图
这将安装支持仪表板+集群启动器的Ray 。
pip install 'ray[default]'如果只想进行最小化安装：
pip install -U ray三、使用Ray进行并行计算任务接下来执行一个示例，该示例使用concurrent.futures，并将其与使用ray执行相同任务的运行进行比较 。
import timeimport concurrent.futuresStime = time.perf_counter()tasks = []sleepTimes = [0.1, 0.2, 0.1, 0.5, 0.7, 0.9, 0.5,              0.4, 1.5, 1.3, 1.0, 0.3, 0.7, 0.6, 0.3, 0.8]print(f"Total time of sleep: {sum(sleepTimes)} for {len(sleepTimes)} tasks")def my_awesome_function(sleepTime=0.1):    time.sleep(sleepTime)    return f"Sleep time {sleepTime}"all_results = []with concurrent.futures.ProcessPoolExecutor() as executor:    tasks = [executor.submit(my_awesome_function, sleep)             for sleep in sleepTimes]    for ff in concurrent.futures.as_completed(tasks):        all_results.Append(ff.result())print(f"Finished in {time.perf_counter()-Stime:.2f}")这将返回：
$ python test_ray.py Total time of sleep: 9.9 for 16 tasks Finished in 1.65这项工作在顺序执行时需要9.9秒完成 。由于本文执行的是并行执行，因此在示例中只用了1.65秒就完成了这项工作 。请注意，这个时间可能因为不同的计算机而有所不同 。
现在使用Ray来完成同样的工作 。本文首先使用ray.init()初始化Ray 。然后，装饰器ray.remote将Python函数转换为可以异步远程执行的函数 。它会立即返回N个可以并行执行的函数副本 。
import timeimport rayimport concurrent.futuresStime = time.perf_counter()tasks = []sleepTimes = [0.1, 0.2, 0.1, 0.5, 0.7, 0.9, 0.5,              0.4, 1.5, 1.3, 1.0, 0.3, 0.7, 0.6, 0.3, 0.8]print(f"Total time of sleep: {sum(sleepTimes)} for {len(sleepTimes)} tasks")# 初始化Ray 。ray.init()@ray.remote #convert to a function that can be executed remotely and asynchronouslydef my_awesome_function(sleepTime=0.1):    time.sleep(sleepTime)    return f"Sleep time {sleepTime}"tasks = []for sleep in sleepTimes:    tasks.append(my_awesome_function.remote(sleep))all_results = ray.get(tasks)print(f"Finished in {time.perf_counter()-Stime:.2f}")


推荐阅读


上一篇：Next.js的崛起：为什么它是现代网站的首选全栈框架

下一篇：一文搞明白Hive与数据库区别