Одновременное чтение массивов numpy параллельно

Рассмотрим следующее:

fine = np.random.uniform(0,100,10)
fine[fine<20] = 0 # introduce some intermittency
coarse = np.sum(fine.reshape(-1,2),axis=1)

fine — это временной ряд величин (например, количество осадков). coarse — это тот же временной ряд, но с уменьшенным вдвое разрешением, поэтому каждые 2 временных шага в fine объединяются в одно значение в coarse.

Затем меня интересует взвешивание, которое определяет пропорции величины coarse, которые соответствуют каждому временному шагу в fine для случаев, когда значение coarse выше нуля.

def w_xx(fine, coarse):
    weights = [] 
    for i, val in enumerate(coarse):
        if val > 0:
            w = fine[i*2:i*2+2]/val # returns both w1 and w2, w1 is 1st element, w2 = 1-w1 is second
            weights.append(w)
    return np.asarray(weights)

Таким образом, w_xx(fine,coarse) вернет массив формы 5,2, где элементы axis=1 являются весами fine для значения coarse.

Все это хорошо для небольших временных рядов, но я запускаю этот анализ на массивах размером ~ 60 000 из fine плюс в цикле из 300+ итераций.

Я пытался сделать это параллельно, используя библиотеку multiprocessing в Python2.7, но мне не удалось продвинуться далеко. Мне нужно одновременно читать оба временных ряда, чтобы получить соответствующие значения fine для каждого значения в coarse, а также работать только со значениями выше 0, чего требует мой анализ.

Я был бы признателен за предложения по лучшему способу сделать это. Я полагаю, что если я смогу определить функцию сопоставления для использования с Pool.map в multiprocessing, я смогу распараллелить это? Я только начал с multiprocessing, поэтому не знаю, есть ли другой способ?

Спасибо.


python numpy arrays parallel-processing python-multiprocessing
person areuexperienced    schedule 07.09.2015    source источник

Ответы (3)


arrow_upward
2
arrow_downward

Вы можете добиться того же результата в векторизованной форме, просто выполнив:

>>> (fine / np.repeat(coarse, 2)).reshape(-1, 2)

затем вы можете отфильтровать строки, в которых coarse равно нулю, используя np.isfinite, поскольку, если coarse равно нулю, вывод будет либо inf, либо nan.

person behzad.nouri    schedule 07.09.2015

arrow_upward
1
arrow_downward

В дополнение к выражению NumPy, предложенному @behzad.nouri, вы можете использовать компилятор Pythran, чтобы получить дополнительное ускорение:

$ cat w_xx.py
#pythran export w_xx(float[], float[])
import numpy as np

def w_xx(fine, coarse):
    w = (fine / np.repeat(coarse, 2))
    return w[np.isfinite(w)].reshape(-1, 2)
$ python -m timeit -s 'import numpy as np; fine = np.random.uniform(0, 100, 100000); fine[fine<20] = 0; coarse = np.sum(fine.reshape(-1, 2), axis=1); from w_xx import w_xx' 'w_xx(fine, coarse)'
1000 loops, best of 3: 1.5 msec per loop
$ pythran w_xx.py -fopenmp -march=native # yes, this generates parallel code
$ python -m timeit -s 'import numpy as np; fine = np.random.uniform(0, 100, 100000); fine[fine<20] = 0; coarse = np.sum(fine.reshape(-1, 2), axis=1); from w_xx import w_xx' 'w_xx(fine, coarse)'
1000 loops, best of 3: 867 usec per loop

Отказ от ответственности: я разработчик Pythran.

person serge-sans-paille    schedule 07.09.2015

arrow_upward
0
arrow_downward

Превосходно! Я не знал о np.repeat, большое спасибо.

Чтобы ответить на мой первоначальный вопрос в том виде, в каком он был представлен, мне также удалось выполнить эту работу с multiprocessing:

import numpy as np    
from multiprocessing import Pool

fine = np.random.uniform(0,100,100000)
fine[fine<20] = 0
coarse = np.sum(fine.reshape(-1,2),axis=1)

def wfunc(zipped):
   return zipped[0]/zipped[1]

def wpar(zipped, processes):
    p = Pool(processes)
    calc = np.asarray(p.map(wfunc, zip(fine,np.repeat(coarse,2))))

    p.close()
    p.join()

    return calc[np.isfinite(calc)].reshape(-1,2)

Однако предложение @behzad.nouri явно лучше:

def w_opt(fine, coarse):
    w = (fine / np.repeat(coarse, 2))
    return w[np.isfinite(w)].reshape(-1,2)    

#using some iPython magic
%timeit w_opt(fine,coarse)
1000 loops, best of 3: 1.88 ms per loop

%timeit w_xx(fine,coarse)
1 loops, best of 3: 342 ms per loop

%timeit wpar(zip(fine,np.repeat(coarse,2)),6) #I've 6 cores at my disposal
1 loops, best of 3: 1.76 s per loop

Спасибо еще раз!

person areuexperienced    schedule 07.09.2015