python - 需要在pyspark中通过类似于scipy.linalg.eig的特征值分解来找到非对称方阵的特征向量

Question

本人是个初学者，如有错误请指正。

我有一个大小为 100 万 x 100 万的方阵。 我想在 pyspark 中找到它的特征向量。我知道computeSVD 给了我特征向量，但这些是通过SVD 得到的，结果是一个密集矩阵，它是一个本地数据结构。我想要 scipy.linalg.eig 给出的结果。

我看到有一个函数 EigenValueDecomposition 在 java 和 Spark 的 scala api 中使用 ARPACK。它会给出与 scipy 中的 eig 相同的特征向量吗？如果是的话，有什么办法可以在 pyspark 中使用它吗？或者是否有任何替代解决方案。我可以以某种方式直接在我的代码中使用 ARPACK 还是我必须自己编写 Arnoldi 迭代(例如)？感谢您的帮助。

Answer 1

我开发了一个Python代码来获取一个scipy稀疏矩阵并创建一个RowMatrix作为computeSVD的输入。这是将 csr_matrix 转换为 SparseVectors 列表所需的部分。我使用并行版本，因为顺序版本要慢得多并且很容易使其并行。

from pyspark.ml.linalg import SparseVector
from pyspark.mllib.linalg.distributed import RowMatrix
from multiprocessing.dummy import Pool as ThreadPool 
from functools import reduce
from pyspark.sql import DataFrame


num_row, num_col = fullMatrix.shape
lst_total = [None] * num_row
selected_indices = [i for i in range(num_row)]

def addMllibSparseVector(idx):
    curr = fullMatrix.getrow(idx)
    arr_ind = np.argsort(curr.indices)
    lst_total[idx] = (idx, SparseVector(num_col\
                 , curr.indices[arr_ind], curr.data[arr_ind]),)
pool = ThreadPool() 
pool.map(addMllibSparseVector, selected_indices)
pool.close()
pool.join()

然后我使用下面的代码创建数据框。

import math
lst_dfs = []
batch_size = 5000
num_range = math.ceil(num_row / batch_size)

lst_dfs = [None] * num_range
selected_dataframes = [i for i in range(num_range)]

def makeDataframes(idx):
    start = idx * batch_size
    end = min(start + batch_size, num_row)
    lst_dfs[idx] = sqlContext.createDataFrame(lst_total[start:end]\
        , ["id", "features"])            
pool = ThreadPool() 
pool.map(makeDataframes, selected_dataframes)
pool.close()
pool.join()

然后我将它们减少到 1 个数据帧并创建 RowMatrix。

raw_df = reduce(DataFrame.unionAll,*lst_dfs)
raw_rdd = raw_df.select('features').rdd.map(list)
raw_rdd.cache() 
mat = RowMatrix(raw_rdd)
svd = mat.computeSVD(100, computeU=True)

我简化了代码，还没有完全测试它。如果有问题请随时发表评论。