[[420515]]

面對計算密集型的任務(wù)，除了多進程，就是分布式計算，如何用 Python 實現(xiàn)分布式計算呢?今天分享一個很簡單的方法，那就是借助于 Ray。

什么是 Ray

Ray 是基于 Python 的分布式計算框架，采用動態(tài)圖計算模型，提供簡單、通用的 API 來創(chuàng)建分布式應(yīng)用。使用起來很方便，你可以通過裝飾器的方式，僅需修改極少的的代碼，讓原本運行在單機的 Python 代碼輕松實現(xiàn)分布式計算，目前多用于機器學(xué)習(xí)。

Ray 的特色：

1、提供用于構(gòu)建和運行分布式應(yīng)用程序的簡單原語。

2、使用戶能夠并行化單機代碼，代碼更改很少甚至為零。

3、Ray Core 包括一個由應(yīng)用程序、庫和工具組成的大型生態(tài)系統(tǒng)，以支持復(fù)雜的應(yīng)用程序。比如 Tune、RLlib、RaySGD、Serve、Datasets、Workflows。

安裝 Ray

最簡單的安裝官方版本的方式：

pip install -U ray 
pip install 'ray[default]' 

如果是 Windows 系統(tǒng)，要求必須安裝 Visual C++ runtime

其他安裝方式見官方文檔。

使用 Ray

一個裝飾器就搞定分布式計算：

import ray 
ray.init() 
 
@ray.remote 
def f(x): 
    return x * x 
 
futures = [f.remote(i) for i in range(4)] 
print(ray.get(futures)) # [0, 1, 4, 9] 

先執(zhí)行 ray.init()，然后在要執(zhí)行分布式任務(wù)的函數(shù)前加一個裝飾器 @ray.remote 就實現(xiàn)了分布式計算。裝飾器 @ray.remote 也可以裝飾一個類：

import ray 
ray.init() 
 
@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)] 
tmp1 = [c.increment.remote() for c in counters] 
tmp2 = [c.increment.remote() for c in counters] 
tmp3 = [c.increment.remote() for c in counters] 
futures = [c.read.remote() for c in counters] 
print(ray.get(futures)) # [3, 3, 3, 3] 

當(dāng)然了，上述的分布式計算依然是在自己的電腦上進行的，只不過是以分布式的形式。程序執(zhí)行的過程中，你可以輸入 http://127.0.0.1:8265/#/ 查看分布式任務(wù)的執(zhí)行情況：

那么如何實現(xiàn) Ray 集群計算呢?接著往下看。

使用 Ray 集群

Ray 的優(yōu)勢之一是能夠在同一程序中利用多臺機器。當(dāng)然，Ray 可以在一臺機器上運行，因為通常情況下，你只有一臺機器。但真正的力量是在一組機器上使用 Ray。

Ray 集群由一個頭節(jié)點和一組工作節(jié)點組成。需要先啟動頭節(jié)點，給 worker 節(jié)點賦予頭節(jié)點地址，組成集群：

你可以使用 Ray Cluster Launcher 來配置機器并啟動多節(jié)點 Ray 集群。你可以在 AWS、GCP、Azure、Kubernetes、阿里云、內(nèi)部部署和 Staroid 上甚至在你的自定義節(jié)點提供商上使用集群啟動器。

Ray 集群還可以利用 Ray Autoscaler，它允許 Ray 與云提供商交互，以根據(jù)規(guī)范和應(yīng)用程序工作負載請求或發(fā)布實例。

現(xiàn)在，我們來快速演示下 Ray 集群的功能，這里是用 Docker 來啟動兩個 Ubuntu 容器來模擬集群：

• 環(huán)境 1: 172.17.0.2 作為 head 節(jié)點
• 環(huán)境 2: 172.17.0.3 作為 worker 節(jié)點，可以有多個 worker 節(jié)點

具體步驟：

1. 下載 ubuntu 鏡像

docker pull ubuntu 

2. 啟動 ubuntu 容器，安裝依賴

啟動第一個

docker run -it --name ubuntu-01 ubuntu bash 

啟動第二個

docker run -it --name ubuntu-02 ubuntu bash 

檢查下它們的 IP 地址：

$ docker inspect -f "{{ .NetworkSettings.IPAddress }}" ubuntu-01 
172.17.0.2 
$ docker inspect -f "{{ .NetworkSettings.IPAddress }}" ubuntu-02 
172.17.0.3 

然后分別在容器內(nèi)部安裝 python、pip、ray

apt update && apt install python3  
apt install python3-pip 
pip3 install ray 

3. 啟動 head 節(jié)點和 worker 節(jié)點

選擇在其中一個容器作為 head 節(jié)點，這里選擇 172.17.0.2，執(zhí)行：

ray start --head --node-ip-address 172.17.0.2 

默認端口是 6379，你可以使用 --port 參數(shù)來修改默認端口，啟動后的結(jié)果如下：

忽略掉警告，可以看到給出了一個提示，如果要把其他節(jié)點綁定到該 head，可以這樣：

ray start --address='172.17.0.2:6379' --redis-password='5241590000000000' 

在另一個節(jié)點執(zhí)行上述命令，即可啟動 worker 節(jié)點：

如果要關(guān)閉，執(zhí)行：

ray stop 

4、執(zhí)行任務(wù)

隨便選擇一個節(jié)點，執(zhí)行下面的腳本，修改下 ray.init() 函數(shù)的參數(shù)：

from collections import Counter 
import socket 
import time 
 
import ray 
 
ray.init(address='172.17.0.2:6379', _redis_password='5241590000000000') 
 
print('''This cluster consists o    f 
    {} nodes in total 
    {} CPU resources in total 
'''.format(len(ray.nodes()), ray.cluster_resources()['CPU'])) 
 
@ray.remote 
def f(): 
    time.sleep(0.001) 
    # Return IP address. 
    return socket.gethostbyname(socket.gethostname()) 
 
object_ids = [f.remote() for _ in range(10000)] 
ip_addresses = ray.get(object_ids) 
 
print('Tasks executed') 
for ip_address, num_tasks in Counter(ip_addresses).items(): 
    print('    {} tasks on {}'.format(num_tasks, ip_address)) 

執(zhí)行結(jié)果如下：

可以看到 172.17.0.2 執(zhí)行了 4751 個任務(wù)，172.17.0.3 執(zhí)行了 5249 個任務(wù)，實現(xiàn)了分布式計算的效果。

最后的話

有了 Ray，你可以不使用 Python 的多進程就可以實現(xiàn)并行計算。今天的機器學(xué)習(xí)主要就是計算密集型任務(wù)，不借助分布式計算速度會非常慢，Ray 提供了簡單實現(xiàn)分布式計算的解決方案。官方文檔提供了很詳細的教程和樣例，感興趣的可以去了解下。

如果有幫助，不妨隨手一個關(guān)注，每天學(xué)點 Python 技術(shù)。