簡單的方式創建分布式應用程序
面對計算密集型的任務，除了多進程，就是分布式計算，如何用 Python 實現分布式計算呢?今天分享一個很簡單的方法，那就是借助于 Ray。
什么是 Ray
Ray 是基于 Python 的分布式計算框架，采用動態圖計算模型，提供簡單、通用的 API 來創建分布式應用。使用起來很方便，你可以通過裝飾器的方式，僅需修改極少的的代碼，讓原本運行在單機的 Python 代碼輕松實現分布式計算，目前多用于機器學習。
Ray 的特色：
1、提供用于構建和運行分布式應用程序的簡單原語。
2、使用戶能夠并行化單機代碼，代碼更改很少甚至為零。
3、Ray Core 包括一個由應用程序、庫和工具組成的大型生態系統，以支持復雜的應用程序。比如 Tune、RLlib、RaySGD、Serve、Datasets、Workflows。
安裝 Ray
最簡單的安裝官方版本的方式：
- pip install -U ray
- pip install 'ray[default]'
如果是 Windows 系統，要求必須安裝 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]
先執行 ray.init()，然后在要執行分布式任務的函數前加一個裝飾器 @ray.remote 就實現了分布式計算。裝飾器 @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]
當然了，上述的分布式計算依然是在自己的電腦上進行的，只不過是以分布式的形式。程序執行的過程中，你可以輸入 http://127.0.0.1:8265/#/ 查看分布式任務的執行情況：
那么如何實現 Ray 集群計算呢?接著往下看。
使用 Ray 集群
Ray 的優勢之一是能夠在同一程序中利用多臺機器。當然，Ray 可以在一臺機器上運行，因為通常情況下，你只有一臺機器。但真正的力量是在一組機器上使用 Ray。
Ray 集群由一個頭節點和一組工作節點組成。需要先啟動頭節點，給 worker 節點賦予頭節點地址，組成集群：
你可以使用 Ray Cluster Launcher 來配置機器并啟動多節點 Ray 集群。你可以在 AWS、GCP、Azure、Kubernetes、阿里云、內部部署和 Staroid 上甚至在你的自定義節點提供商上使用集群啟動器。
Ray 集群還可以利用 Ray Autoscaler，它允許 Ray 與云提供商交互，以根據規范和應用程序工作負載請求或發布實例。
現在，我們來快速演示下 Ray 集群的功能，這里是用 Docker 來啟動兩個 Ubuntu 容器來模擬集群：
- 環境 1: 172.17.0.2 作為 head 節點
- 環境 2: 172.17.0.3 作為 worker 節點，可以有多個 worker 節點
具體步驟：
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
然后分別在容器內部安裝 python、pip、ray
- apt update && apt install python3
- apt install python3-pip
- pip3 install ray
3. 啟動 head 節點和 worker 節點
選擇在其中一個容器作為 head 節點，這里選擇 172.17.0.2，執行：
- ray start --head --node-ip-address 172.17.0.2
默認端口是 6379，你可以使用 --port 參數來修改默認端口，啟動后的結果如下：
忽略掉警告，可以看到給出了一個提示，如果要把其他節點綁定到該 head，可以這樣：
- ray start --address='172.17.0.2:6379' --redis-password='5241590000000000'
在另一個節點執行上述命令，即可啟動 worker 節點：
如果要關閉，執行：
- ray stop
4、執行任務
隨便選擇一個節點，執行下面的腳本，修改下 ray.init() 函數的參數：
- 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))
執行結果如下：
可以看到 172.17.0.2 執行了 4751 個任務，172.17.0.3 執行了 5249 個任務，實現了分布式計算的效果。
最后的話
有了 Ray，你可以不使用 Python 的多進程就可以實現并行計算。今天的機器學習主要就是計算密集型任務，不借助分布式計算速度會非常慢，Ray 提供了簡單實現分布式計算的解決方案。官方文檔提供了很詳細的教程和樣例，感興趣的可以去了解下。
如果有幫助，不妨隨手一個關注，每天學點 Python 技術。
