即使是具有多個(gè)CPU核心的單處理器計(jì)算機(jī)（處理器具有一個(gè)或多個(gè)核心，計(jì)算機(jī)具有一個(gè)或多個(gè)處理器），也會(huì)給人一種能夠同時(shí)運(yùn)行多個(gè)任務(wù)的錯(cuò)覺。當(dāng)我們擁有多個(gè)處理器時(shí)，就可以真正以并行的方式執(zhí)行計(jì)算。V8X28資訊網(wǎng)——每日最新資訊28at.com

一、并行計(jì)算與分布式計(jì)算的區(qū)別

并行計(jì)算在現(xiàn)代計(jì)算中非常有用，幾乎是必需的，目的是實(shí)現(xiàn)最大性能。開發(fā)者將運(yùn)行時(shí)間較長的計(jì)算任務(wù)分成較小的塊，并將其分配給不同的處理器。這種策略使開發(fā)者能夠在相同的時(shí)間內(nèi)進(jìn)行更多的計(jì)算。對(duì)于構(gòu)建基于GUI的應(yīng)用程序，總是需要對(duì)系統(tǒng)進(jìn)行并行設(shè)計(jì)，以便一個(gè)線程可以保持可用狀態(tài)以更新GUI并響應(yīng)用戶輸入。V8X28資訊網(wǎng)——每日最新資訊28at.com

并行計(jì)算和分布式計(jì)算的區(qū)別在于，對(duì)于并行計(jì)算，多個(gè)處理器位于同一主板上。分布式計(jì)算則使用多臺(tái)計(jì)算機(jī)同時(shí)解決問題?，F(xiàn)代分布式系統(tǒng)能夠在網(wǎng)絡(luò)（局域網(wǎng)/廣域網(wǎng)）上進(jìn)行通信。分布式計(jì)算的優(yōu)點(diǎn)在于其價(jià)格和可擴(kuò)展性。如果開發(fā)者需要更多的計(jì)算能力，那么可以很輕松地添加更多的計(jì)算機(jī)。V8X28資訊網(wǎng)——每日最新資訊28at.com

從根本上講，并行計(jì)算和分布式計(jì)算的架構(gòu)非常相似。主要區(qū)別在于分布式計(jì)算使用的是分布式內(nèi)存空間，而不是共享內(nèi)存空間。它具有能夠?yàn)殚_發(fā)者的應(yīng)用程序提供統(tǒng)一邏輯（而不是物理）內(nèi)存空間的軟件層，可以幫助開發(fā)者將為并行計(jì)算編寫的代碼應(yīng)用于分布式計(jì)算。V8X28資訊網(wǎng)——每日最新資訊28at.com

在本文中將介紹如何使用開源Python庫Ray來幫助開發(fā)者進(jìn)行并行和分布式計(jì)算，Ray將Pythonic函數(shù)和類轉(zhuǎn)換為分布式設(shè)置中的任務(wù)和角色。本文將只介紹函數(shù)的示例，但是類的概念非常相似。V8X28資訊網(wǎng)——每日最新資訊28at.com

二、使用pip安裝Ray

這將安裝支持儀表板+集群?jiǎn)?dòng)器的Ray。V8X28資訊網(wǎng)——每日最新資訊28at.com

pip install 'ray[default]'

如果只想進(jìn)行最小化安裝：V8X28資訊網(wǎng)——每日最新資訊28at.com

pip install -U ray

三、使用Ray進(jìn)行并行計(jì)算任務(wù)

接下來執(zhí)行一個(gè)示例，該示例使用concurrent.futures，并將其與使用ray執(zhí)行相同任務(wù)的運(yùn)行進(jìn)行比較。V8X28資訊網(wǎng)——每日最新資訊28at.com

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

這將返回：V8X28資訊網(wǎng)——每日最新資訊28at.com

$ python test_ray.py Total time of sleep: 9.9 for 16 tasks Finished in 1.65

這項(xiàng)工作在順序執(zhí)行時(shí)需要9.9秒完成。由于本文執(zhí)行的是并行執(zhí)行，因此在示例中只用了1.65秒就完成了這項(xiàng)工作。請(qǐng)注意，這個(gè)時(shí)間可能因?yàn)椴煌挠?jì)算機(jī)而有所不同。V8X28資訊網(wǎng)——每日最新資訊28at.com

現(xiàn)在使用Ray來完成同樣的工作。本文首先使用ray.init()初始化Ray。然后，裝飾器ray.remote將Python函數(shù)轉(zhuǎn)換為可以異步遠(yuǎn)程執(zhí)行的函數(shù)。它會(huì)立即返回N個(gè)可以并行執(zhí)行的函數(shù)副本。V8X28資訊網(wǎng)——每日最新資訊28at.com

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

這將返回：V8X28資訊網(wǎng)——每日最新資訊28at.com

Total time of sleep: 9.9 for 16 tasks Finished in 3.18

由于存在一些開銷，會(huì)有一些延遲，但對(duì)于大型計(jì)算來說，這種延遲可以忽略不計(jì)。V8X28資訊網(wǎng)——每日最新資訊28at.com

四、大規(guī)模計(jì)算的聚合值

Ray可以輕松地用于聚合多個(gè)值，這對(duì)于構(gòu)建需要跨多臺(tái)機(jī)器進(jìn)行計(jì)算的大型應(yīng)用程序非常重要。對(duì)于大規(guī)模計(jì)算，Ray可以將聚合的運(yùn)行時(shí)間從線性改為對(duì)數(shù)。V8X28資訊網(wǎng)——每日最新資訊28at.com

接下來看一個(gè)示例：V8X28資訊網(wǎng)——每日最新資訊28at.com

import timeimport rayimport numpy as npStime = time.perf_counter()@ray.remotedef create_matrix(size):    return np.random.normal(size=size)@ray.remotedef multiply_matrices(x, y):    return np.dot(x, y)@ray.remotedef sum_matrices(x, y):    return np.add(x, y)m1 = create_matrix.remote([1000, 1000])m2 = create_matrix.remote([1000, 1000])m3 = create_matrix.remote([1000, 1000])m4 = create_matrix.remote([1000, 1000])m12 = multiply_matrices.remote(m1, m2)m34 = multiply_matrices.remote(m3, m4)a12_34 =  sum_matrices.remote(m12, m34)## 結(jié)果MM = ray.get(a12_34)print(f"Finished in {time.perf_counter()-Stime:.2f}")

在上面的示例中，本文首先創(chuàng)建了四個(gè)矩陣，將它們分為兩組，對(duì)每組中的矩陣進(jìn)行乘法運(yùn)算，然后對(duì)每組的乘法結(jié)果進(jìn)行求和。在這里，乘法運(yùn)算是并行進(jìn)行的，然后將結(jié)果聚合以獲得求和結(jié)果。V8X28資訊網(wǎng)——每日最新資訊28at.com

參考資料

1. 【安裝Ray】：https://docs.ray.io/en/latest/ray-overview/installation.html
2. Pierfederici, F. (2016). 《Distributed Computing with Python》. Journal of Physics A: Mathematical and Theoretical (Vol. 44, Issue 8). Packt Publishing Ltd.

本文鏈接：http://www.tebozhan.com/showinfo-26-16533-0.html使用Ray輕松進(jìn)行Python分布式計(jì)算

聲明：本網(wǎng)頁內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com

上一篇： 彩虹橋架構(gòu)演進(jìn)之路-性能篇

下一篇： 七個(gè)殺手級(jí)IntelliJ IDEA插件