按键减少时会导致内存不足

Question

我正在研究一种需要对大矩阵进行数学运算的算法。基本上，该算法包括以下步骤：

输入：大小为n的两个向量u和v

>

对于两个矩阵中的每个条目，应用一个函数f。返回两个矩阵M_u，M_v

求M_的本征值和本征向量。对于i=0，返回e_i，ev_i，。。。，n-1

计算每个特征向量的外积。返回一个矩阵O_i=e_i*转置（e_i），i=0，。。。，n-1

用e_i=e_i delta_i调整每个特征值，其中delta_i=所有元素之和（O_i和M_v的元素乘积）/2*mu，其中mu是一个参数

最终返回一个矩阵a=i=0，。。。，n-1

我面临的问题主要是当n很大（15000或更多）时的内存，因为这里所有的矩阵都是稠密矩阵。我目前实现这一点的方法可能不是最好的，而且部分有效。

我用行矩阵表示mu，用奇异值分解得到特征分解。

SVD的结果U因子是一个列为ev_i的行矩阵，因此我必须手动转置它，使其行成为ev_i。结果e向量是特征值e_i。

由于内存溢出，之前将每行ev_i直接映射到O_i的尝试失败了

R = U.map{
    case(i,ev_i) => {
      (i, ev_i.toArray.zipWithIndex)
    }
  }//add index for each element in a vector
  .flatMapValues(x=>x)}
  .join(U)//eigen vectors column is appended
  .map{case(eigenVecId, ((vecElement,elementId), eigenVec))=>(elementId, (eigenVecId, vecElement*eigenVec))}

为了在上面的步骤5中计算调整后的e_i，M_v被存储为元组（i，denseVector）的rdd。然后

deltaRdd = R.join(M_v)
  .map{
    case(j,((i,row_j_of_O_i),row_j_of_M_v))=>
    (i,row_j_of_O_i.t*DenseVector(row_j_of_M_v.toArray)/(2*mu))
  }.reduceByKey(_+_)

最后，为了计算A，同样由于内存问题，我必须首先连接来自不同RDD的行，然后按键减少。明确地

R_rearranged = R.map{case(j, (i, row_j_of_O_i))=>(i,(j,row_j_of_O_i))}
termsForA = R_rearranged.join(deltaRdd)
A = termsForA.map{
  case(i,(j,row_j_of_O_i), delta_i)) => (j, (delta_i + e(i))*row_j_of_O_i)
}
.reduceByKey(_+_)

上述实现适用于termsForA的步骤，这意味着如果我对termsForA执行像termsForA. get（1）. foreach（println）这样的操作，它就成功了。但是如果我在A上执行操作，比如A. count（），驱动程序上会出现OOM错误。

我试图调整sparks配置以增加驱动程序内存和并行度，但都失败了。

Answer 1

我猜Spark在某个时候决定不需要对执行器执行操作，而需要对驱动程序执行所有操作。事实上，termsForA也会像count一样失败。不知何故，我通过广播德尔塔德和e。

Answer 2

使用IndexedRowMatrix而不是RowMatrix，这将有助于转换和转置。假设您的IndexedRowMatrix是Irm

svd = Irm.computeSVD(k, True)
U = svd.U
U =  U.toCoordinateMatrix().transpose().toIndexedRowMatrix()

您可以将Irm转换为BlockMatrix，以便与另一个分布式BlockMatrix相乘。