Calcolo della similarità di Spark Jaccard per hashing min lento rispetto all'approccio banale

Question

Dato 2 elenco enorme di valori, sto provando a calcolare jaccard similarity tra di loro in Spark usando Scala.

Si supponga che colHashed1 contenga il primo elenco di valori e colHashed2 contenga il secondo elenco.

Approccio 1 (approccio banale):

val jSimilarity = colHashed1.intersection(colHashed2).distinct.count/(colHashed1.union(colHashed2).distinct.count.toDouble) 

Approccio 2 (utilizzando minHashing):

Ho usato il metodo spiegato here.

import java.util.zip.CRC32 

def getCRC32 (s : String) : Int = 
{ 
    val crc=new CRC32 
    crc.update(s.getBytes) 
    return crc.getValue.toInt & 0xffffffff 
} 

val maxShingleID = Math.pow(2,32)-1 
def pickRandomCoeffs(kIn : Int) : Array[Int] = 
{ 
    var k = kIn 
    val randList = Array.fill(k){0} 

    while(k > 0) 
    { 
    // Get a random shingle ID. 

    var randIndex = (Math.random()*maxShingleID).toInt 

    // Ensure that each random number is unique. 
    while(randList.contains(randIndex)) 
    { 
     randIndex = (Math.random()*maxShingleID).toInt 
    } 

    // Add the random number to the list. 
    k = k - 1 
    randList(k) = randIndex 
    } 

    return randList 
} 

val colHashed1 = list1Values.map(a => getCRC32(a)) 
val colHashed2 = list2Values.map(a => getCRC32(a)) 

val nextPrime = 4294967311L 
val numHashes = 10 

val coeffA = pickRandomCoeffs(numHashes) 
val coeffB = pickRandomCoeffs(numHashes) 

var signature1 = Array.fill(numHashes){0} 
for (i <- 0 to numHashes-1) 
{ 
    // Evaluate the hash function. 
    val hashCodeRDD = colHashed1.map(ele => ((coeffA(i) * ele + coeffB(i)) % nextPrime)) 

    // Track the lowest hash code seen. 
    signature1(i) = hashCodeRDD.min.toInt 
} 

var signature2 = Array.fill(numHashes){0} 
for (i <- 0 to numHashes-1) 
{ 
    // Evaluate the hash function. 
    val hashCodeRDD = colHashed2.map(ele => ((coeffA(i) * ele + coeffB(i)) % nextPrime)) 

    // Track the lowest hash code seen. 
    signature2(i) = hashCodeRDD.min.toInt 
} 


var count = 0 
// Count the number of positions in the minhash signature which are equal. 
for(k <- 0 to numHashes-1) 
{ 
    if(signature1(k) == signature2(k)) 
    count = count + 1 
} 
val jSimilarity = count/numHashes.toDouble 

Approccio 1 sembra a sovraperformare Approccio 2 in termini di tempo sempre. Quando ho analizzato il codice, la chiamata di funzione min() nello RDD in Approccio 2 richiede un tempo significativo e tale funzione viene chiamata più volte in base a quante funzioni hash vengono utilizzate.

Le operazioni di intersezione e unione utilizzate in Approccio 1 sembrano funzionare più velocemente rispetto alle ripetute chiamate di funzione min().

Non capisco perché minHashing non aiuti qui. Mi aspettavo che minHashing funzionasse più velocemente rispetto all'approccio banale. C'è qualcosa che sto facendo di sbagliato qui?

dati campione possono essere visualizzate here

Answer 1

JaccardSimilarity con MinHash non sta dando risultati coerenti:

import java.util.zip.CRC32 

object Jaccard { 
    def getCRC32(s: String): Int = { 
    val crc = new CRC32 
    crc.update(s.getBytes) 
    return crc.getValue.toInt & 0xffffffff 
    } 

    def pickRandomCoeffs(kIn: Int, maxShingleID: Double): Array[Int] = { 
    var k = kIn 
    val randList = Array.ofDim[Int](k) 

    while (k > 0) { 
     // Get a random shingle ID. 
     var randIndex = (Math.random() * maxShingleID).toInt 
     // Ensure that each random number is unique. 
     while (randList.contains(randIndex)) { 
     randIndex = (Math.random() * maxShingleID).toInt 
     } 
     // Add the random number to the list. 
     k = k - 1 
     randList(k) = randIndex 
    } 
    return randList 
    } 


    def approach2(list1Values: List[String], list2Values: List[String]) = { 

    val maxShingleID = Math.pow(2, 32) - 1 

    val colHashed1 = list1Values.map(a => getCRC32(a)) 
    val colHashed2 = list2Values.map(a => getCRC32(a)) 

    val nextPrime = 4294967311L 
    val numHashes = 10 

    val coeffA = pickRandomCoeffs(numHashes, maxShingleID) 
    val coeffB = pickRandomCoeffs(numHashes, maxShingleID) 

    val signature1 = for (i <- 0 until numHashes) yield { 
     val hashCodeRDD = colHashed1.map(ele => (coeffA(i) * ele + coeffB(i)) % nextPrime) 
     hashCodeRDD.min.toInt // Track the lowest hash code seen. 
    } 

    val signature2 = for (i <- 0 until numHashes) yield { 
     val hashCodeRDD = colHashed2.map(ele => (coeffA(i) * ele + coeffB(i)) % nextPrime) 
     hashCodeRDD.min.toInt // Track the lowest hash code seen 
    } 

    val count = (0 until numHashes) 
     .map(k => if (signature1(k) == signature2(k)) 1 else 0) 
     .fold(0)(_ + _) 


    val jSimilarity = count/numHashes.toDouble 
    jSimilarity 
    } 


    // def approach1(list1Values: List[String], list2Values: List[String]) = { 
    // val colHashed1 = list1Values.toSet 
    // val colHashed2 = list2Values.toSet 
    // 
    // val jSimilarity = colHashed1.intersection(colHashed2).distinct.count/(colHashed1.union(colHashed2).distinct.count.toDouble) 
    // jSimilarity 
    // } 


    def approach1(list1Values: List[String], list2Values: List[String]) = { 
    val colHashed1 = list1Values.toSet 
    val colHashed2 = list2Values.toSet 

    val jSimilarity = (colHashed1 & colHashed2).size/(colHashed1 ++ colHashed2).size.toDouble 
    jSimilarity 
    } 

    def main(args: Array[String]) { 

    val list1Values = List("a", "b", "c") 
    val list2Values = List("a", "b", "d") 

    for (i <- 0 until 5) { 
     println(s"Iteration ${i}") 
     println(s" - Approach 1: ${approach1(list1Values, list2Values)}") 
     println(s" - Approach 2: ${approach2(list1Values, list2Values)}") 
    } 

    } 
} 

USCITA:

Iteration 0 
- Approach 1: 0.5 
- Approach 2: 0.5 
Iteration 1 
- Approach 1: 0.5 
- Approach 2: 0.5 
Iteration 2 
- Approach 1: 0.5 
- Approach 2: 0.8 
Iteration 3 
- Approach 1: 0.5 
- Approach 2: 0.8 
Iteration 4 
- Approach 1: 0.5 
- Approach 2: 0.4 

Perché stai usando?

Answer 2

Mi sembra che il costo generale per l'approccio minHashing superi solo la sua funzionalità in Spark. Soprattutto come aumenta numHashes. Ecco alcune osservazioni che ho trovato nel tuo codice:

Prima, while (randList.contains(randIndex)) questa parte rallenterà sicuramente il tuo processo come numHashes (che è comunque uguale alla dimensione di randList) aumenta.

In secondo luogo, è possibile risparmiare un po 'di tempo se si riscrive questo codice:

var signature1 = Array.fill(numHashes){0} 
for (i <- 0 to numHashes-1) 
{ 
    // Evaluate the hash function. 
    val hashCodeRDD = colHashed1.map(ele => ((coeffA(i) * ele + coeffB(i)) % nextPrime)) 

    // Track the lowest hash code seen. 
    signature1(i) = hashCodeRDD.min.toInt 
} 

var signature2 = Array.fill(numHashes){0} 
for (i <- 0 to numHashes-1) 
{ 
    // Evaluate the hash function. 
    val hashCodeRDD = colHashed2.map(ele => ((coeffA(i) * ele + coeffB(i)) % nextPrime)) 

    // Track the lowest hash code seen. 
    signature2(i) = hashCodeRDD.min.toInt 
} 


var count = 0 
// Count the number of positions in the minhash signature which are equal. 
for(k <- 0 to numHashes-1) 
{ 
    if(signature1(k) == signature2(k)) 
    count = count + 1 
} 

in

var count = 0 
for (i <- 0 to numHashes - 1) 
{ 
    val hashCodeRDD1 = colHashed1.map(ele => ((coeffA(i) * ele + coeffB(i)) % nextPrime)) 
    val hashCodeRDD2 = colHashed2.map(ele => ((coeffA(i) * ele + coeffB(i)) % nextPrime)) 

    val sig1 = hashCodeRDD1.min.toInt 
    val sig2 = hashCodeRDD2.min.toInt 

    if (sig1 == sig2) { count = count + 1 } 
} 

Questo metodo semplifica i tre anelli in uno solo. Tuttavia, non sono sicuro che ciò darebbe un enorme impulso nel tempo di calcolo.

Un altro suggerimento che ho, partendo dal presupposto che il primo approccio risulta ancora fuori per essere molto più veloce è quello di utilizzare la proprietà di set per modificare il primo approccio:

val colHashed1_dist = colHashed1.distinct 
val colHashed2_dist = colHashed2.distinct 
val intersect_cnt = colHashed1_dist.intersection(colHashed2_dist).distinct.count 

val jSimilarity = intersect_cnt/(colHashed1_dist.count + colHashed2_dist.count - intersect_cnt).toDouble 

con questo, invece di ottenere l'unione , puoi semplicemente riutilizzare il valore dell'intersezione.

Answer 3

In realtà, nell'applicativo LSH si calcola minHash una sola volta per ciascuno dei propri documenti e quindi si confrontano due minHasi per ogni possibile coppia di documenti. E in caso di approccio banale eseguirai un confronto completo dei documenti per ogni possibile coppia di documenti. Che è approssimativamente il numero N/2/2 di confronti. Quindi il costo aggiuntivo del calcolo di minshashes è trascurabile per un numero sufficiente di documenti.

si dovrebbe effettivamente confrontare le prestazioni di un approccio banale:

val jSimilarity = colHashed1.intersection(colHashed2).distinct.count/(colHashed1.union(colHashed2).distinct.count.toDouble) 

e le prestazioni del calcolo Jaccard distanza (ultime righe nel codice):

var count = 0 
// Count the number of positions in the minhash signature which are equal. 
for(k <- 0 to numHashes-1) 
{ 
    if(signature1(k) == signature2(k)) 
    count = count + 1 
} 
val jSimilarity = count/numHashes.toDouble