Используйте луч для распараллеливания поиска ближайшего соседа - PullRequest
Новая
       10

Используйте луч для распараллеливания поиска ближайшего соседа

1 голос
/ 03 февраля 2020

Предположим, у меня большая разреженная матрица. Я хочу взять каждый вектор-строку этой матрицы и вычислить косинусное расстояние до ближайшего соседа среди предыдущих window_size строк матрицы.

Поскольку sklearn.neighbors не улучшает время выполнения, когда используя распараллеливание (см. эту проблему на github ), я попытался распараллелить процесс, используя ray. Мой код работает лучше, чем sklearn с многопроцессорной обработкой, но он все еще медленнее, чем просто вычисление последовательного расстояния.

Мой код приведен ниже. Что-то я сделал не так и должен улучшить? 

import scipy.sparse
from sklearn.metrics.pairwise import cosine_distances
from sklearn.neighbors import NearestNeighbors
import numpy as np
import timeit
import ray
import math

n = 4000
m = 100
big_matrix = scipy.sparse.random(n, m).tocsr()
window_size = 1000
retry = 1
n_jobs = 4
ray.init(num_cpus=n_jobs)


def simple_cosine_distance():
    distances = np.zeros(n)
    for i in range(1, n):
        past = big_matrix[max(0, i - window_size):i]
        query = big_matrix[i]
        distances[i] = cosine_distances(query, past).min(axis=1)


def sklearn_nneighbor():
    distances = np.zeros(n)
    for i in range(1, n):
        past = big_matrix[max(0, i - window_size):i]
        nn = NearestNeighbors(metric="cosine", n_neighbors=1, n_jobs=1)
        nn.fit(X=past)
        query = big_matrix[i]
        distance, _ = nn.kneighbors(query)
        distances[i] = distance[0]


def sklearn_nneighbor_parallel():
    distances = np.zeros(n)
    for i in range(1, n):
        past = big_matrix[max(0, i - window_size):i]
        nn = NearestNeighbors(metric="cosine", n_neighbors=1, n_jobs=n_jobs)
        nn.fit(X=past)
        query = big_matrix[i]
        distance, _ = nn.kneighbors(query)
        distances[i] = distance[0]


@ray.remote
def get_chunk_min(data, indices, indptr, shape, slice_i, slice_j, query):
    matrix = scipy.sparse.csr_matrix((data, indices, indptr), shape=shape)
    past = matrix[slice_i:slice_j]
    query = matrix[query]
    chunk_min = cosine_distances(query, past).min(axis=1)
    return chunk_min


def ray_parallel():
    distances = np.zeros(n)
    data = ray.put(big_matrix.data)
    indices = ray.put(big_matrix.indices)
    indptr = ray.put(big_matrix.indptr)
    shape = ray.put(big_matrix.shape)
    for i in range(1, n):
        chunk_size = math.ceil((i - max(0, i - window_size)) / n_jobs)
        chunk_mins = ray.get([
            get_chunk_min.remote(
                data,
                indices,
                indptr,
                shape,
                enum,
                enum + chunk_size,
                i
            ) for enum in range(max(0, i - window_size), i, chunk_size)])
        distances[i] = min(chunk_mins)

for method in ["simple_cosine_distance", "sklearn_nneighbor", "sklearn_nneighbor_parallel", "ray_parallel"]:
    print(method)
    print(timeit.timeit(method + "()", setup="from __main__ import " + method, number=retry))
    print("*"*50)

Вывод:

simple_cosine_distance 3.978868665999471

sklearn_nneighbor 4.265772191996803

sklearn_nneighbor_parallel 28.664759318002325

ray_parallel 17.89882287799992

1 Ответ

0 голосов
/ 08 февраля 2020

https://docs.python.org/3/library/concurrent.futures.html Эта документация является хорошим ресурсом для понимания некоторых ошибок, которые вы можете совершать при распараллеливании рабочей нагрузки с помощью Ray.

Также обратите внимание, что, если ваша удаленная функция достаточно проста, издержки распределенных вычислений (например, время планирования, время сериализации и т. д. c.) могут быть больше, чем время вычислений самой функции.

...