python - 计算成对距离矩阵 : is a scalable, Python 中可用的大数据就绪方法？

Question

我有一个包含项目特征值的 CSV 文件:每一行都是一个三元组(id_item、id_feature、值)，表示特定项目的特定特征值。数据非常稀疏。

我需要计算两个项目距离矩阵，一个使用 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 ) 但据我所知它支持同一台机器上的多个内核而不是集群计算。 这是一个相关问题Easy way to use parallel options of scikit-learn functions on HPC但我没有得到在我的具体情况下最好的解决方案是什么，如果 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，以map reduce方式实现计算，通过mrjob运行job

3) 将数据存储在HDFS中，用类似SQL的方式实现计算(不知道是否简单可行，得好好想想)，用PyHive运行

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

Answer 1

原型(prototype):

我建议你使用Pyro4并通过 分而治之 范式、一个主节点和多个从节点来实现它。

如果你有 n 个项目你有 n(n-1)/2 对，你使用 sklearn 成对距离和每个节点上的最大作业(n_jobs 参数) .

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

用于生产:

我建议你PySpark 2.1.1 . Map Reduce 已弃用。