Блог

У меня есть куча операций умножения матриц, которые выполняются только по строкам. Мне было интересно, как ускорить вычисления путем распараллеливания:

 data = np.random.randint(1, 100, (100000, 800))
indices_1 = np.equal(data, 1)
A = np.zeros((100000, 100))
B = np.random.randn(800, 100)

for i in range(100000):
   ones = indices_1[i]
   not_ones = ~indices_1[i]
   B_ones = B[ones]
   B_not_ones = B[not_ones]
   A[i] = (data[i][not_ones] @ B_not_ones) @ np.linalg.inv(B_not_ones.T @ B_not_ones)  
   data[i][ones] = A[i] @ B_ones.T
    
 

Я пробовал многопроцессорный, но по какой-то причине он работал не лучше, чем последовательный. Вот моя многопроцессорная реализация:

 from multiprocessing.pool import ThreadPool, Pool
pool = ThreadPool() # can also use Pool

def f(i):
   ones = indices_1[i]
   not_ones = ~indices_1[i]
   B_ones = B[ones]
   B_not_ones = B[not_ones]
   A[i] = (data[i][not_ones] @ B_not_ones) @ np.linalg.inv(B_not_ones.T @ B_not_ones)  
   data[i][ones] = A[i] @ B_ones.T


pool.map(f, range(100000))
 

Оба дали одинаковое время выполнения (около 32 секунд). Другой метод распараллеливания, например concurrent.futures , не улучшил время выполнения (используется, как показано ниже):

 with concurrent.futures.ThreadPoolExecutor() as executor:
     result = executor.map(f, range(100000))
 

Я также пытался применить dask , но не смог заставить их фреймворк работать в моем случае. Любая помощь будет высоко оценена! Спасибо!

 import numpy as np
import multiprocessing as mp


data = list(np.random.randint(1, 100, (100000, 800)))
indices_1 = np.equal(data, 1)
A = list(np.zeros((100000, 100)))
B = np.random.randn(800, 100)


def f(data, A, i):
    ones = indices_1[i]
    not_ones = ~indices_1[i]
    B_ones = B[ones]
    B_not_ones = B[not_ones]
    A[i] = (data[i][not_ones] @ B_not_ones) @ np.linalg.inv(B_not_ones.T @ B_not_ones)
    data[i][ones] = A[i] @ B_ones.T

with mp.Manager() as manager:
    data_global = manager.list(data)
    A_global = manager.list(A)

    with mp.Pool() as p:
        results = [ p.apply_async(f, (data_global, A_global, i,)) for i in range(100000) ]
        for i in results:
            i.wait()

    data_global = list(data_global)
    A_global = list(A_global)