怎么在python中使用dispy實現分布式計算

怎么在python中使用dispy實現分布式計算？相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。

dispy，是用asyncoro實現的分布式并行計算框架。

框架也是非常精簡，只有4個組件，在其源碼文件夾下可以找到：

dispy.py (client) provides two ways of creating “clusters”: JobCluster when only one instance of dispy may run and SharedJobCluster when multiple instances may run (in separate processes). If JobCluster is used, the scheduler contained within dispy.py will distribute jobs on the server nodes; if SharedJobCluster is used, a separate scheduler (dispyscheduler) must be running.

dispynode.py executes jobs on behalf of dispy. dispynode must be running on each of the (server) nodes that form the cluster.

dispyscheduler.py is needed only when SharedJobCluster is used; this provides a scheduler that can be shared by multiple dispy users.

dispynetrelay.py is needed when nodes are located across different networks; this relays information about nodes on a network to the scheduler. If all the nodes are on same network, there is no need for dispynetrelay - the scheduler and nodes automatically discover each other.

一般情況下，使用dispy和dispynode就已經足夠解決問題了。

簡單使用：

服務器端：

在服務器端啟動dispy，監聽并接收所有發來的計算任務，完成計算后將結果返回給客戶端。

打開python_home/Scripts文件夾，在安裝dispy后會有上面說到的4個dispy組件，以py文件形式存在。當然你也可以在dispy的源碼文件夾里面找到對于的dispynode.py文件，然后執行

python dispynode.py -c 2 -i 192.168.138.128 -p 51348 -s secret --clean

python dispynode.py -c 2 -i 192.168.8.143 -p 51348 -s secret --clean

這里192.168.138.128和192.168.8.143是執行計算節點的ip（對服務器來說相當于localhost），這里我啟用了兩個節點，每個節點使用2個cpu資源，其中有一個節點是在虛擬機，一個是本地機器。

-s secret是通信密碼，客戶端和服務器連接需要密碼，密碼隨意。

--clean表示每次啟動服務都刪除上次的啟動信息，如果不刪除，可能會出現pid占用的錯誤。

客戶端：

在客戶端需要注意的是，發送到計算節點函數所引用的模塊，不能在py文件的頂層導入，而需要在函數內導入。

對于需要導入自定義模塊，比較麻煩一點，需要先實例化函數，才能在計算節點的函數中使用。

# 這些在頂層導入的模塊只能是這個py文件用
import time
import socket
import numpy
import datetime

# 這個是自定義函數，要在本模塊中先實例化才能在計算節點函數中調用使用，
# 而本模塊的其他地方可以直接調用使用
from my_package.my_model import get_time 

# 實例化自定義的函數，注意后面是沒有括號的，否則就是直接調用得到返回值了
now = get_time.now

# 計算函數，dispy將這個函數和參數一并發送到服務器節點
# 如果函數有多個參數，需要包裝程tuple格式
def compute(args):
 n,array=args # 如果函數有多個參數，需要包裝程tuple格式
 # 看到沒，計算需要的模塊是在函數內導入的
 import time, socket
 time.sleep(3)
 host = socket.gethostname()
 # 這個py文件中自定義函數，可以直接引用
 total= my_sum(array)
 # 這個now是在其他模塊中自定義的函數，需要在頂層先實例化才能引用
 now_time=now()
 return (host, n, total,now_time)

def sum(array):
 # 自定義函數，需要的模塊同樣需要在函數內導入
 import numpy as np
 return np.sum(array)

def loadData():
 # 自定義函數，生成測試數據
 import numpy as np
 data = np.random.rand(20,20)
 data = [line for line in data]
 return data



if __name__ == '__main__':
 import dispy, random
 # 定義兩個計算節點
 nodes = ['192.168.8.143', '192.168.138.128']
 # 啟動計算集群，和服務器通信，通信密鑰是'secret'
 # depends 為依賴函數
 cluster = dispy.JobCluster(compute,nodes=nodes,
      secret='secret',depends=[sum，now])
 jobs = []

 datas = loadData()
 for n in range(len(datas)):
  # 提交任務
  job = cluster.submit((n,datas[n]))
  job.id = n
  jobs.append(job)
 # print(datetime.datetime.now())
 # cluster.wait() # 等待所有任務完成后才接著往下執行
 # print(datetime.datetime.now())
 for job in jobs:
  host, n, total,t = job()
  print('%s executed job %s at %s with %s total=%.2f t=%s' 
    % (host, job.id, job.start_time, n,total,t))
  # other fields of 'job' that may be useful:
  # print job.stdout, job.stderr, job.exception, 
  # job.ip_addr, job.start_time, job.end_time
 # 顯示集群計算狀態
 cluster.stats()

看完上述內容，你們掌握怎么在python中使用dispy實現分布式計算的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！