通過多處理共享和操作多個numpy陣列-有解無憂

我正在嘗試利用多處理來加快基于陣列的計算。一般作業流程如下：

我有三個陣列：
- id_array持有屬于一起的陣列的 ID
- class_array是一個分類陣列（只是表示影像分類中的類值的整數）
- prob_array有這些類的概率
基于我想要的部分：
- 在每個段中找到類多數
- 平均段內的概率，但僅適用于類別多數的“像素”

這是我的非并行示例，效果很好：

import numpy as np

id_array = np.array([[1, 1, 2, 2, 2],
                     [1, 1, 2, 2, 4],
                     [3, 3, 4, 4, 4],
                     [3, 3, 4, 4, 4]])
class_array = np.array([[7, 7, 6, 8, 8],
                        [5, 7, 7, 8, 8],
                        [8, 8, 5, 5, 8],
                        [9, 9, 8, 7, 7]])
prob_array = np.array([[0.7, 0.3, 0.9, 0.5, 0.1],
                       [0.4, 0.6, 0.3, 0.5, 0.9],
                       [0.8, 0.6, 0.2, 0.2, 0.3],
                       [0.4, 0.4, 0.6, 0.3, 0.7]])

all_ids = np.unique(                     )
dst_classes = np.zeros_like(class_array)
dst_probs = np.zeros_like(prob_array)

for my_id in all_ids:
    segment = np.where(id_array == my_id)
    class_data = class_array[segment]
    # get majority of classes within segment
    majority = np.bincount(class_data.flatten()).argmax()
    # get probabilities within segment
    prob_data = prob_array[segment]
    # get probabilities within segment where class equals majority
    majority_probs = prob_data[np.where(class_data == majority)]
    # get median of these probabilities
    median_prob = np.nanmedian(majority_probs)
    # write values
    dst_classes[segment] = majority
    dst_probs[segment] = median_prob

print(dst_classes)
print(dst_probs)

問題是我的真實資料有大約 400 萬個片段，然后需要一周的時間來計算。所以我按照本教程提出了這個：

import numpy as np
import multiprocessing as mp

WORKER_DICT = dict()
NODATA = 0

def shared_array_from_np_array(data_array, init_value=None):
    raw_array = mp.RawArray(np.ctypeslib.as_ctypes_type(data_array.dtype), data_array.size)
    shared_array = np.frombuffer(raw_array, dtype=data_array.dtype).reshape(data_array.shape)
    if init_value:
        np.copyto(shared_array, np.full_like(data_array, init_value))
        return raw_array, shared_array
    else:
        np.copyto(shared_array, data_array)
        return raw_array, shared_array

def init_worker(id_array, class_array, prob_array, class_results, prob_results):
    WORKER_DICT['id_array'] = id_array
    WORKER_DICT['class_array'] = class_array
    WORKER_DICT['prob_array'] = prob_array
    WORKER_DICT['class_results'] = class_results
    WORKER_DICT['prob_results'] = prob_results
    WORKER_DICT['shape'] = id_array.shape
    mp.freeze_support()

def worker(id):
    id_array = WORKER_DICT['id_array']
    class_array = WORKER_DICT['class_array']
    prob_array = WORKER_DICT['prob_array']
    class_result = WORKER_DICT['class_results']
    prob_result = WORKER_DICT['prob_results']
    # array indices for "id"
    segment = np.where(id_array == id)
    # get data at these indices, mask nodata values
    class_data = np.ma.masked_equal(class_array[segment], NODATA)
    # get majority value
    majority_class = np.bincount(class_data.flatten()).argmax()
    # get probabilities
    probs = prob_array[segment]
    majority_probs = probs[np.where(class_array[segment] == majority_class)]
    med_majority_probs = np.nanmedian(majority_probs)
    class_result[segment] = majority_class
    prob_result[segment] = med_majority_probs
    return

if __name__ == '__main__':
    # segment IDs
    id_ra, id_array = shared_array_from_np_array(np.array(
        [[1, 1, 2, 2, 2],
         [1, 1, 2, 2, 4],
         [3, 3, 4, 4, 4],
         [3, 3, 4, 4, 4]]))
    # classification
    cl_ra, class_array = shared_array_from_np_array(np.array(
        [[7, 7, 6, 8, 8],
         [5, 7, 7, 8, 8],
         [8, 8, 5, 5, 8],
         [9, 9, 8, 7, 7]]))
    # probabilities
    pr_ra, prob_array = shared_array_from_np_array(np.array(
        [[0.7, 0.3, 0.9, 0.5, 0.1],
         [0.4, 0.6, 0.3, 0.5, 0.9],
         [0.8, 0.6, 0.2, 0.2, 0.3],
         [0.4, 0.4, 0.6, 0.3, 0.7]]))
    cl_res, class_results = shared_array_from_np_array(class_array, 0)
    pr_res, prob_results = shared_array_from_np_array(prob_array, 0.)
    unique_ids = np.unique(id_array)
    init_args = (id_ra, cl_ra, pr_ra, cl_res, pr_res, id_array.shape)
    with mp.Pool(processes=2, initializer=init_worker, initargs=init_args) as pool:
        pool.map_async(worker, unique_ids)
    print('Majorities:', cl_res)
    print('Probabilities:', pr_res)

但我不明白我現在如何才能得到我的結果以及它們是否正確。我試過了

np.frombuffer(cl_res)
np.frombuffer(pr_res)

但這只給了我 10 個值cl_res（應該有 20 個），它們看起來完全隨機，而pr_res具有與prob_array.

我已經嘗試使用這里的其他示例，例如this，但也無法讓它們作業。這看起來像一個類似的問題，但它已經需要大量的知識，多處理是如何真正作業的，而我沒有（多處理的初學者）。

uj5u.com熱心網友回復：

需要修復的幾件事：

您需要在中創建 numpy 陣列init_worker()，它還應該帶一個shape引數：

def init_worker(id_ra, cl_ra, pr_ra, cl_res, pr_res, shape):
    WORKER_DICT['id_array'] = np.ctypeslib.as_array(id_ra, shape)
    WORKER_DICT['class_array'] = np.ctypeslib.as_array(cl_ra, shape)
    WORKER_DICT['prob_array'] = np.ctypeslib.as_array(pr_ra, shape)
    WORKER_DICT['class_results'] = np.ctypeslib.as_array(cl_res, shape)
    WORKER_DICT['prob_results'] = np.ctypeslib.as_array(pr_res, shape)

您應該檢查 ifinit_value is not None而不是init_valuein shared_array_from_np_array()，因為 0 的計算結果為 False。
mp.freeze_support()if __name__ == '__main__'根據其檔案，只能在之后立即呼叫。
pool.map_async()回傳一個需要等待的 AsyncResult 物件；您可能想要pool.map()，在處理完成之前一直阻塞。

轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/494394.html

標籤：Python 数组 python-3.x 麻木的多处理

上一篇：為什么變數的值會發生變化？

下一篇：來自高吞吐量源的實時資料繪圖