Aggiunta di due RDD [mllib.linalg.Vector] '

Question

Ho bisogno di aggiungere due matrici che sono memorizzate in due file.

Il contenuto di latest1.txt e latest2.txt ha la prossima str:

 
1 2 3 
4 5 6 
7 8 9 

io sto leggendo i file come segue:

scala> val rows = sc.textFile(“latest1.txt”).map { line => val values = line.split(‘ ‘).map(_.toDouble) 
    Vectors.sparse(values.length,values.zipWithIndex.map(e => (e._2, e._1)).filter(_._2 != 0.0)) 
} 

scala> val r1 = rows 
r1: org.apache.spark.rdd.RDD[org.apache.spark.mllib.linalg.Vector] = MappedRDD[2] at map at :14 

scala> val rows = sc.textFile(“latest2.txt”).map { line => val values = line.split(‘ ‘).map(_.toDouble) 
    Vectors.sparse(values.length,values.zipWithIndex.map(e => (e._2, e._1)).filter(_._2 != 0.0)) 
} 

scala> val r2 = rows 
r2: org.apache.spark.rdd.RDD[org.apache.spark.mllib.linalg.Vector] = MappedRDD[2] at map at :14 

voglio aggiungere r1, r2. Quindi, c'è un modo per aggiungere questi due RDD[mllib.linalg.Vector] in Apache-Spark.

Answer 1

Questa è in realtà una buona domanda. Lavoro regolarmente con mllib e non mi sono reso conto che queste operazioni di algebra lineare di base non sono facilmente accessibili.

Il punto è che i vettori breeze di base dispongono di tutte le manipolazioni di algebra lineare che ci si aspetterebbe, incluso naturalmente l'aggiunta di elementi di base che si è specificamente menzionato.

Tuttavia l'attuazione brezza è nascosta dal mondo esterno tramite:

[private mllib] 

Dunque, dal punto di vista API mondo esterno/pubblico, come possiamo accedere a tali primitivi?

Alcuni di questi sono già esposti: ad es. somma dei quadrati:

/** 
* Returns the squared distance between two Vectors. 
* @param v1 first Vector. 
* @param v2 second Vector. 
* @return squared distance between two Vectors. 
*/ 
def sqdist(v1: Vector, v2: Vector): Double = { 
    ... 
} 

Tuttavia la selezione di tali metodi disponibili è limitato - e di fatto non non includono le operazioni di base tra elemento saggio addizione, sottrazione, moltiplicazione, ecc

Quindi ecco il migliore che ho potuto vedere:

• convertire i vettori di brezza:
• Eseguire le operazioni vettoriali in brezza
01.235.
• riconvertire da gioco da ragazzi per mllib Vector

Ecco alcuni esempi di codice:

val v1 = Vectors.dense(1.0, 2.0, 3.0) 
val v2 = Vectors.dense(4.0, 5.0, 6.0) 
val bv1 = new DenseVector(v1.toArray) 
val bv2 = new DenseVector(v2.toArray) 

val vectout = Vectors.dense((bv1 + bv2).toArray) 
vectout: org.apache.spark.mllib.linalg.Vector = [5.0,7.0,9.0] 

Answer 2

Il codice seguente espone i metodi asBreeze e fromBreeze Spark. Questa soluzione supporta SparseVector rispetto all'utilizzo di vector.toArray. Si noti che Spark potrebbe cambiare la propria API in futuro e ha già rinominato toBreeze in asBreeze.

package org.apache.spark.mllib.linalg 
import breeze.linalg.{Vector => BV} 
import org.apache.spark.sql.functions.udf 

/** expose vector.toBreeze and Vectors.fromBreeze 
    */ 
object VectorUtils { 

    def fromBreeze(breezeVector: BV[Double]): Vector = { 
    Vectors.fromBreeze(breezeVector) 
    } 

    def asBreeze(vector: Vector): BV[Double] = { 
    // this is vector.asBreeze in Spark 2.0 
    vector.toBreeze 
    } 

    val addVectors = udf { 
    (v1: Vector, v2: Vector) => fromBreeze(asBreeze(v1) + asBreeze(v2)) 
    } 

} 

Con questo si può fare df.withColumn("xy", addVectors($"x", $"y")).