计算成对距离矩阵:是Python中可用的可扩展,大数据就绪方法吗？

Question

我有一个包含项目特征值的CSV文件:每一行都是三元组(id_item,id_feature,value),表示特定项目的特定功能的值.数据非常稀少.

我需要计算两个项目距离矩阵,一个使用Pearson相关作为度量,另一个使用Jaccard索引.

目前我实现了内存解决方案,我做了类似这样的事情:

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)

它工作得很好而且速度非常快但不能水平扩展.我希望能够通过向集群添加节点来提高性能,并且即使在大数据场景中,一切都可以工作,再次只是添加节点.我不在乎这个过程需要几个小时; 距离需要每天更新一次.

什么是最好的方法？

1)Sklearn pairwise_distances有一个n_jobs参数,允许利用并行计算(http://scikit-learn.org/stable/modules/generated/sklearn.metrics.pairwise.pairwise_distances.html)但据我所知支持同一台机器上的多个核心,而不支持群集计算.这是一个相关的问题在HPC上使用scikit-learn函数的并行选项的简单方法,但我没有得到在我的特定情况下最佳解决方案,以及Joblib是否确实存在问题.

此外,在内存中读取CSV的部分仍然是瓶颈:我可以将CSV存储在HDFS中并读取它,例如:

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

然后遍历cat.stdout:

for line in cat.stdout:
    ....

但我不确定这是一个很好的解决方案.

2)将数据存储在HDFS中,在地图中实现计算减少时尚并通过mrjob运行作业

3)将数据存储在HDFS中,以类似SQL的方式实现计算(我不知道它是否简单易行,我必须考虑它)并使用PyHive运行它

当然我想尽可能多地保留当前代码,因此解决方案1)的变体对我来说是最好的.

Answer 1

原型：

我建议您使用Pyro4divide and conquer并通过范例、一个主节点和多个从节点来实现它。

如果您有成对的n项目n(n-1)/2，则可以使用 sklearn 成对距离以及每个节点上的最大作业数（n_jobs 参数）。

您将任务对集拆分a并在节点上执行b，然后在主节点上重新组合结果。

用于生产：

我建议你使用PySpark 2.1.1。MapReduce 已被弃用。