dispy，是用asyncoro實現(xiàn)的分布式并行計算框架。

框架也是非常精簡，只有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就已經(jīng)足夠解決問題了。

簡單使用：

服務(wù)器端：

在服務(wù)器端啟動dispy，監(jiān)聽并接收所有發(fā)來的計算任務(wù)，完成計算后將結(jié)果返回給客戶端。

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

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是執(zhí)行計算節(jié)點的ip（對服務(wù)器來說相當(dāng)于localhost），這里我啟用了兩個節(jié)點，每個節(jié)點使用2個cpu資源，其中有一個節(jié)點是在虛擬機(jī)，一個是本地機(jī)器。

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

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

客戶端：

在客戶端需要注意的是，發(fā)送到計算節(jié)點函數(shù)所引用的模塊，不能在py文件的頂層導(dǎo)入，而需要在函數(shù)內(nèi)導(dǎo)入。

對于需要導(dǎo)入自定義模塊，比較麻煩一點，需要先實例化函數(shù)，才能在計算節(jié)點的函數(shù)中使用。

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

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

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

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

def sum(array):
 # 自定義函數(shù)，需要的模塊同樣需要在函數(shù)內(nèi)導(dǎo)入
 import numpy as np
 return np.sum(array)

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



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

 datas = loadData()
 for n in range(len(datas)):
  # 提交任務(wù)
  job = cluster.submit((n,datas[n]))
  job.id = n
  jobs.append(job)
 # print(datetime.datetime.now())
 # cluster.wait() # 等待所有任務(wù)完成后才接著往下執(zhí)行
 # 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
 # 顯示集群計算狀態(tài)
 cluster.stats()

以上這篇python分布式計算dispy的使用詳解就是小編分享給大家的全部內(nèi)容了，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論