2 개의 희소 행렬 곱하기

Question

텍스트 파일에서 생성 된 두 개의 희소 행렬을 곱하려고합니다. 현재 다음 코드를 가지고 있습니다 :

import org.apache.spark.SparkContext 
import org.apache.spark.SparkConf 


@SerialVersionUID(123L) 
case class M_Matrix (i: Long, j: Long, v: Double) 
extends Serializable {} 

@SerialVersionUID(123L) 
case class N_Matrix (j: Long, k: Long, w: Double) 
extends Serializable {} 

object Multiply { 
def main(args: Array[ String ]){ 
val conf = new SparkConf().setAppName("Multiply") 
val sc = new SparkContext(conf) 

val M_ = sc.textFile(args(0)).map(line => { val a = line.split(",") 
    M_Matrix(a(0).toLong,a(1).toLong,a(2).toDouble) }) 

val N_ = sc.textFile(args(1)).map(line => { val a = line.split(",") 
    N_Matrix(a(0).toLong, a(1).toLong, a(2).toDouble) }) 
val res = M_.map(M_ => (M_.j,M_)) 
    .join(N_.map(N_ => (N_.j, N_))) 
    .map({case (j, ((i,v),(k,w))) => ((i,k), v * w)}) 
    .reduceByKey(_ + _) 
    .map({ case ((i,k), sum) => (i, k, sum)}) 
res.saveAsTextFile(args(2)) 
sc.stop() 
} 
} 

* v * w에서 유효한 연산자가 아닙니다.

내가 뭘 잘못하고 있니?

Answer 1

코드가 거의 정확하며 약간의 변경이 필요합니다. 특히, map에 join 작업을하기 전에 행렬에서 수행 :

튜플은 튜플이 예상되는 것입니다 다음 문에 있기 때문에 여기에 값으로 사용한다

M_.map(M_ => (M_.j, (M_.i, M_.v))).join(N_.map(N_ => (N_.j, (N_.k, N_.w)))) 

.map{ case(j, ((i,v),(k,w))) => ((i,k), v * w)} 

---

Spark에서 내장 기능을 사용하는 것이 좋습니다. 효율성이 높아야합니다. 행렬은 RDD[MatrixEntry]에서 만들 수 있습니다. 먼저이 행렬을 만듭니다

import org.apache.spark.mllib.linalg.distributed.{BlockMatrix, CoordinateMatrix, MatrixEntry} 

val M_ = sc.textFile(args(0)).map{ line => 
    val a = line.split(",") 
    MatrixEntry(a(0).toLong, a(1).toLong, a(2).toDouble) 
} 

val N_ = sc.textFile(args(1)).map{ line => 
    val a = line.split(",") 
    MatrixEntry(a(0).toLong, a(1).toLong, a(2).toDouble) 
} 

을 다음 BlockMatrix로 모두 변환 :

val matrix_M = new CoordinateMatrix(M_).toBlockMatrix() 
val matrix_N = new CoordinateMatrix(N_).toBlockMatrix() 

을 그 후, 당신은 단순히 기능 multiply 사용하여 곱할 수 : I가 설치 한 스칼라 플러그인을

val res = matrix_M.multiply(matrix_N)