本筆記使用方大佬的降水計算指標，鏈接如下

降水各類常用指標csdn地址

向量化計算

使用np的向量化函數將一次只能返回單個標量的函數，向量化成能接受定制shape數組且可以指定類型的返回。

首先觀察函數如：

def HSS_one_threshold(obs, pre, threshold=0.1):

'''

HSS - Heidke skill score

Args:

obs (numpy.ndarray): observations

pre (numpy.ndarray): pre

threshold (float) : threshold for rainfall values binaryzation

(rain/no rain)

Returns:

float: HSS value

'''

hits, misses, falsealarms, correctnegatives = prep_clf(obs=obs, pre = pre,

threshold=threshold)

HSS_num = 2 * (hits * correctnegatives - misses * falsealarms)

HSS_den = (misses**2 + falsealarms**2 + 2*hits*correctnegatives +

(misses + falsealarms)*(hits + correctnegatives))

return HSS_num / HSS_den

這個HSS函數只能計算單個時次的，如果我想向量化計算不同時刻的hss評分，就只能使用for循環，但顯然這不是我的風格。而向量化函數很簡單只需要一句：

hss = np.vectorize(HSS_one_threshold,excluded=['threshold'],signature='(m,n),(m,n)->()')

np.vectorize這個函數很簡單，但定制自己的向量化函數核心卻是

signature=’(m,n),(m,n)->()’

，signatrue標簽，這個異常晦澀難懂且很少有人講清楚這個東西是干什么的，具體請參考文檔，我假設你詳細閱讀完文檔后還是不懂，那么接下來請看我的表演，首先signature中有兩個核心定義即numpy數組的迭代維和核心計算維，我們假設有一個三維數組(time,m,n)迭代維默認從數組最左邊的維開始，核心計算維從右邊開始，如果我想通過向量化的hss函數計算time個時間步的hss得分并返回到一個列表中，那么就如我給出的函數所示，計算維為(m,n)那么向量化后的函數接受的參數就是三維數組中的后兩維，然后在迭代維time上并行計算，因為numpy底層是用c計算，所以這個簽名相當于一個給c接口的參數。而

->()

表示計算返回的格式()代表標量當然這個還能根據任務自定義維度，甚至可以跳過中間某個維或者自定義廣播形式非常復雜，但普通需要for循環的計算均可以用signture消滅，非常暢快。比如計算某個場與另一個場的相關系數，如果分辨率很小據我觀察大多數python計算程序都是用for來遍歷這樣直接爆炸，但使用signuture自定義可以非常完美的解決這個問題。

利用多核進行多進程計算

顯然根據經驗io密集型使用多線程，計算密集使用多進程，我曾嘗試過多線程開文件套多進程計算但碰到很多問題，但針對本問題使用多進程即可，或者多進程開文件套多進程計算，這需要修改linux默認限制打開文件數。如果我想同時計算不同的如csi，pod的得分就可以使用如下程序：

dbz = ['10dbz','20dbz','30dbz','40dbz','50dbz']

thdlst = [10,20,30,40,50]

for thd_number,key in enumerate(dbz):

"""

多進程實現并行運算

"""

func_lst = [hss,csi,far,pod]

#進程級的字典

share_dict = multiprocessing.Manager().dict()

#進程級的鎖

share_lock = multiprocessing.Manager().Lock()

def multiprocess(func,thd):

func = func

print(func._name_)

#print(dir(func))

thd = thd

#全局變量

global unet,rover,tagan

temp_unet = func(unet[:,:,:480],unet[:,:,480:],threshold=thd)

temp_rover = func(rover[:,:,:480],rover[:,:,480:],threshold=thd)

temp_tagan = func(tagan[:,:,:480],tagan[:,:,480:],threshold=thd)

data30 = [temp_unet[4],temp_rover[4],temp_tagan[4]]

data60 = [temp_unet[9],temp_rover[9],temp_tagan[9]]

data30 = pd.Series(data30,['uent','rover','tagan'])

data60 = pd.Series(data60,['uent','rover','tagan'])

#使用當前函數名作為進程字典的key

key = func._name_

share_lock.acquire()

share_dict[key] = (data30,data60)

share_lock.release()

print('the process of %s is finisthed '%(key))

thd = thdlst[thd_number]

process_lst =[]

p = Pool(4)

for func in func_lst :

#p.apply_async(multiprocess, args=(multiprocess,thd))

process = multiprocessing.Process(target=multiprocess,args=(func,thd))

process_lst.append(process)

for pss in process_lst:

pss.start()

for pss in process_lst:

pss.join()

首先給一個進程級別的共享列表或者字典用來儲存所有返回的得分值，在計算過程中使用相同的數據不同的技巧評分所以將數據在函數中定義為全局變量，不同進程只訪問內存不修改，但在寫入字典的時候給一個鎖防止沖突。這樣如果計算指標在幾十個以上時數據量在工業的TB或者PB級時，就可以利用服務器的多核把cpu拉滿。

需要注意的是將函數向量化后新的對象就沒有了__name__屬性，需要在向量化函數時setattr一個name屬性

總結

以上是生活随笔為你收集整理的python深入与提高_关于提高Python计算性能的说明摘要,深入,提升,python,笔记,小结,不,定时,更新...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。