Calcul d'une matrice de distance par paire: est-ce une approche évolutive, prête pour les données volumineuses, disponible en Python?

Question

J'ai un fichier CSV contenant des valeurs de caractéristiques pour les éléments: chaque ligne est un triple (id_item, id_feature, valeur) représentant la valeur d'une caractéristique spécifique pour un élément spécifique. Les données sont très clairsemées.

Je dois calculer deux matrices de distance d'items, l'une utilisant la corrélation de Pearson comme métrique et l'autre utilisant l'index de Jaccard.

Au moment où je mis en œuvre une solution en mémoire et je fais quelque chose comme ceci:

import numpy as np 
from numpy import genfromtxt 
from scipy.sparse import coo_matrix 
from scipy.sparse import csr_matrix 
from scipy.stats.stats import pearsonr 
import sklearn.metrics.pairwise 
import scipy.spatial.distance as ds 
import scipy.sparse as sp 

# read the data 
my_data = genfromtxt('file.csv', delimiter=',') 
i,j,value=my_data.T 

# create a sparse matrix 
m=coo_matrix((value,(i,j))) 

# convert in a numpy array 
m = np.array(m.todense()) 

# create the distance matrix using pdist 
d = ds.pdist(m.T, 'correlation') 

d= ds.squareform(d) 

il fonctionne bien et il est assez rapide mais il est pas extensible horizontalement. Je voudrais pouvoir augmenter les performances simplement en ajoutant des nœuds à un cluster et que tout fonctionne même dans un gros scénario de données, encore une fois en ajoutant des nœuds. Je me fiche que le processus prenne des heures; les distances doivent être mises à jour une fois par jour.

Quelle est la meilleure approche? 1) Sklearn pairwise_distances possède un paramètre n_jobs qui permet de tirer parti du calcul parallèle (http://scikit-learn.org/stable/modules/generated/sklearn.metrics.pairwise.pairwise_distances.html) mais pour autant que je sache, il prend en charge plusieurs cœurs sur la même machine et non en cluster. Ceci est une question connexe Easy way to use parallel options of scikit-learn functions on HPC mais je n'ai pas compris quelle est la meilleure solution dans mon cas spécifique et si Joblib a effectivement des problèmes.

En outre, la partie qui se lit dans la mémoire du CSV serait toujours un goulot d'étranglement: Je peux stocker le CSV dans HDFS et lisez faire quelque chose comme:

import subprocess 
cat = subprocess.Popen(["hadoop", "fs", "-cat", "data.csv"], stdout=subprocess.PIPE) 

puis boucle à travers cat.stdout:

for line in cat.stdout: 
    .... 

mais je ne suis pas sûr que ce soit une bonne solution.

2) Stocker les données dans HDFS, mettre en œuvre le calcul sur une carte réduire la mode et exécuter le travail via mrjob

3) Stocker les données dans HDFS, mettre en œuvre le calcul d'une manière type SQL (je ne sais pas si c'est facile et faisable, je dois y penser) et le lancer en utilisant PyHive

Bien sûr je voudrais garder le plus possible le code actuel, donc une variante de la solution 1) est la meilleure pour moi.

Answer 1

Prototype:

Je vous suggère d'utiliser Pyro4 et de mettre en œuvre ce paradigme avec divide and conquer, un noeud maître et plusieurs noeuds esclaves.

Si vous avez n éléments vous avez n(n-1)/2 paires, vous utilisez sklearn distances par paires avec le maximum de travaux (paramètre n_jobs) sur chaque nœud.

Vous divisez votre ensemble de paires en tâches a et exécutez-le sur b nœuds et regroupez les résultats sur votre nœud maître.

Pour la production:

conseils que je vous PySpark 2.1.1.La réduction de la carte devient obsolète.