如何用 Python 實(shí)現(xiàn)分布式計(jì)算？

劇照：云南蟲(chóng)谷

作者：somenzz

來(lái)源：Python七號(hào)

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

什么是 Ray

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

Ray 的特色：

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

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

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

安裝 Ray

最簡(jiǎn)單的安裝官方版本的方式：

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

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

其他安裝方式見(jiàn)官方文檔。

使用 Ray

一個(gè)裝飾器就搞定分布式計(jì)算：

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ù)前加一個(gè)裝飾器 @ray.remote 就實(shí)現(xiàn)了分布式計(jì)算。裝飾器 @ray.remote 也可以裝飾一個(gè)類：

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)然了，上述的分布式計(jì)算依然是在自己的電腦上進(jìn)行的，只不過(guò)是以分布式的形式。程序執(zhí)行的過(guò)程中，你可以輸入 http://127.0.0.1:8265/#/ 查看分布式任務(wù)的執(zhí)行情況：

那么如何實(shí)現(xiàn) Ray 集群計(jì)算呢？接著往下看。

使用 Ray 集群

Ray 的優(yōu)勢(shì)之一是能夠在同一程序中利用多臺(tái)機(jī)器。當(dāng)然，Ray 可以在一臺(tái)機(jī)器上運(yùn)行，因?yàn)橥ǔＧ闆r下，你只有一臺(tái)機(jī)器。但真正的力量是在一組機(jī)器上使用 Ray。

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

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

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

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

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

具體步驟：

1. 下載 ubuntu 鏡像

docker pull ubuntu

2. 啟動(dòng) ubuntu 容器，安裝依賴

啟動(dòng)第一個(gè)

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

啟動(dòng)第二個(gè)

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. 啟動(dòng) head 節(jié)點(diǎn)和 worker 節(jié)點(diǎn)

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

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

默認(rèn)端口是 6379，你可以使用 --port 參數(shù)來(lái)修改默認(rèn)端口，啟動(dòng)后的結(jié)果如下：

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

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

在另一個(gè)節(jié)點(diǎn)執(zhí)行上述命令，即可啟動(dòng) worker 節(jié)點(diǎn)：

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

ray stop

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

隨便選擇一個(gè)節(jié)點(diǎn)，執(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 個(gè)任務(wù)，172.17.0.3 執(zhí)行了 5249 個(gè)任務(wù)，實(shí)現(xiàn)了分布式計(jì)算的效果。

最后的話

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

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

Python貓技術(shù)交流群開(kāi)放啦！群里既有國(guó)內(nèi)一二線大廠在職員工，也有國(guó)內(nèi)外高校在讀學(xué)生，既有十多年碼齡的編程老鳥(niǎo)，也有中小學(xué)剛剛?cè)腴T的新人，學(xué)習(xí)氛圍良好！想入群的同學(xué)，請(qǐng)?jiān)诠?hào)內(nèi)回復(fù)『交流群』，獲取貓哥的微信（謝絕廣告黨，非誠(chéng)勿擾！）~