apache-spark - 使用 python 的 Spark 矩阵乘法

Question

我正在尝试使用 Apache Spark 和 Python 进行矩阵乘法。

这是我的数据

from pyspark.mllib.linalg.distributed import RowMatrix

我的向量 RDD

rows_1 = sc.parallelize([[1, 2], [4, 5], [7, 8]])
rows_2 = sc.parallelize([[1, 2], [4, 5]])

我的矩阵

mat1 = RowMatrix(rows_1)
mat2 = RowMatrix(rows_2)

我想做这样的事情:

mat = mat1 * mat2

我写了一个函数来处理矩阵乘法，但我担心处理时间很长。这是我的功能:

def matrix_multiply(df1, df2):
    nb_row = df1.count()    
    mat=[]
    for i in range(0, nb_row):
        row=list(df1.filter(df1['index']==i).take(1)[0])
        row_out = []
        for r in range(0, len(row)):
            r_value = 0
            col = df2.select(df2[list_col[r]]).collect()
            col = [list(c)[0] for c in col]
            for c in range(0, len(col)): 
                r_value += row[c] * col[c]
            row_out.append(r_value)            
        mat.append(row_out)
    return mat 

我的功能做了很多 Spark Action (采取，收集等)。该函数是否会占用大量处理时间？
如果有人有其他想法，这将对我有所帮助。

Answer 1

你不能。自 RowMatrix没有有意义的行索引，不能用于乘法。甚至忽略唯一的分布式矩阵 which supports multiplication with another distributed structure是 BlockMatrix .

from pyspark.mllib.linalg.distributed import *

def as_block_matrix(rdd, rowsPerBlock=1024, colsPerBlock=1024):
    return IndexedRowMatrix(
        rdd.zipWithIndex().map(lambda xi: IndexedRow(xi[1], xi[0]))
    ).toBlockMatrix(rowsPerBlock, colsPerBlock)

as_block_matrix(rows_1).multiply(as_block_matrix(rows_2))