pyspark计算稀疏向量的距离矩阵

Question

我正在尝试构建一种通用方法来计算许多稀疏向量（100k 向量，长度为 250k）的距离矩阵。在我的示例中，数据以 scipy csr 矩阵表示。这就是我正在做的：

首先，我定义了一种将 csr 行转换为 pyspark SparseVectors 的方法：

def csr_to_sparse_vector(row):
    return SparseVector(row.shape[1], sorted(row.indices), row.data)

现在我将行转换为向量并将它们保存到一个列表，然后将其提供给 SparkContext：

sparse_vectors = [csr_to_sparse_vector(row) for row in refs_sample]
rdd = sc.parallelize(sparse_vectors)

在下一步中，我使用笛卡尔函数构建所有对（类似于此 post：）

在这个实验中，我想使用相应定义的 tje Jaccard 相似度：

def jacc_sim(pair):
    dot_product = pair[0].dot(pair[1])
    try:
        sim = dot_product / (pair[0].numNonzeros() + pair[1].numNonzeros())
    except ZeroDivisionError:
        return 0.0
    return sim

现在我应该只映射函数并收集结果：

distance_matrix = rdd2.map(lambda x: jacc_sim(x)).collect()

我运行此代码是在一个小样本上编写的，在本地计算机和具有 180 个节点的集群上都只有 100 个文档。任务耗时很长，最后崩溃了：https://pastebin.com/UwLUXvUZ

有什么错误的建议吗？

此外，如果距离度量是对称的 sim(x,y) == sim(y,x) 我们只需要矩阵的上三角。我找到了一个post，它通过过滤解决了这个问题():

rdd2 = rdd.cartesian(rdd).filter(lambda x: x[0] < x[1])

但这不适用于 SparseVectors 列表。

Answer 1

是列表有问题，还是列表由 SparseVectors 组成？一种想法是尝试将 SparseVectors 转换为 DenseVectors，这是我在此处找到的建议 ()。计算结果没什么不同，只是Spark如何处理而已。

Answer 2

问题是配置错误导致我的数据被分割成 1000 个分区。解决方案是简单地明确告诉 spark 他应该创建多少个分区（例如 10）：

rdd = sc.parallelize(sparse_vectors, 10)

此外，我用枚举扩展了稀疏向量列表，这样我就可以过滤掉不属于上三角矩阵的对：

sparse_vectors = [(i, csr_to_sparse_vector(row)) for i, row in enumerate(authors)]
rdd = sc.parallelize(sparse_vectors, 10)
rdd2 = rdd.cartesian(rdd).filter(lambda x: x[0][0] < x[1][0])
rdd2.map(lambda x: jacc_sim(x)).filter(lambda x: x is not None).saveAsTextFile('hdfs:///user/username/similarities')

归属相似度函数如下所示：

def jacc_sim(pair):
    id_0 = pair[0][0]
    vec_0 = pair[0][1]
    id_1 = pair[1][0]
    vec_1 = pair[1][1]
    dot_product = vec_0.dot(vec_1)
    try:
        sim = dot_product / (vec_0.numNonzeros() + vec_1.numNonzeros())
        if sim > 0:
            return (id_0, id_1, sim)
    except ZeroDivisionError:
        pass
    return None

这对我来说非常有用，我希望其他人也会觉得它有用！

pyspark计算稀疏向量的距离矩阵

pyspark calculate distance matrix of sparse vectors

python

scipy

sparse-matrix

apache-spark

pyspark