spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Meeraj Kunnumpurath <mee...@servicesymphony.com>
Subject Re: Nearest neighbour search
Date Sun, 13 Nov 2016 20:07:12 GMT
That was a bit of a brute force search, so I changed the code to use a UDF
to create the dot product between the two IDF vectors, and do a sort on the
new column.

package com.ss.ml.clustering

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.ml.feature.{IDF, Tokenizer, HashingTF}
import org.apache.spark.ml.linalg.Vector

object ClusteringBasics extends App {

  val spark = SparkSession.builder().appName("Clustering
Basics").master("local").getOrCreate()
  import spark.implicits._

  val df = spark.read.option("header", "false").csv("data")

  val tk = new Tokenizer().setInputCol("_c2").setOutputCol("words")
  val tf = new HashingTF().setInputCol("words").setOutputCol("tf")
  val idf = new IDF().setInputCol("tf").setOutputCol("tf-idf")

  val df1 = tf.transform(tk.transform(df))
  val idfs = idf.fit(df1).transform(df1)

  val nn = nearestNeighbour("<http://dbpedia.org/resource/Barack_Obama>", idfs)
  println(nn)

  def nearestNeighbour(uri: String, ds: DataFrame) : String = {
    val tfIdfSrc = ds.filter(s"_c0 ==
'$uri'").take(1)(0).getAs[Vector]("tf-idf")
    def dorProduct(vectorA: Vector) = {
      var dp = 0.0
      var index = vectorA.size - 1
      for (i <- 0 to index) {
        dp += vectorA(i) * tfIdfSrc(i)
      }
      dp
    }
    val dpUdf = udf((v1: Vector, v2: Vector) => dorProduct(v1))
    ds.filter(s"_c0 != '$uri'").withColumn("dp",
dpUdf('tf-idf)).sort("dp").take(1)(0).getString(1)
  }

}


However, that is generating the exception below,

Exception in thread "main" java.lang.RuntimeException: Unsupported literal
type class org.apache.spark.ml.feature.IDF idf_e49381a285dd
at
org.apache.spark.sql.catalyst.expressions.Literal$.apply(literals.scala:57)
at org.apache.spark.sql.functions$.lit(functions.scala:101)
at org.apache.spark.sql.Column.$minus(Column.scala:672)
at
com.ss.ml.clustering.ClusteringBasics$.nearestNeighbour(ClusteringBasics.scala:36)
at
com.ss.ml.clustering.ClusteringBasics$.delayedEndpoint$com$ss$ml$clustering$ClusteringBasics$1(ClusteringBasics.scala:22)
at
com.ss.ml.clustering.ClusteringBasics$delayedInit$body.apply(ClusteringBasics.scala:8)
at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App$$anonfun$main$1.apply(App.scala:76)
at scala.App$$anonfun$main$1.apply(App.scala:76)
at scala.collection.immutable.List.foreach(List.scala:381)
at
scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35)
at scala.App$class.main(App.scala:76)
at com.ss.ml.clustering.ClusteringBasics$.main(ClusteringBasics.scala:8)
at com.ss.ml.clustering.ClusteringBasics.main(ClusteringBasics.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:483)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140)

On Sun, Nov 13, 2016 at 10:56 PM, Meeraj Kunnumpurath <
meeraj@servicesymphony.com> wrote:

> This is what I have done, is there a better way of doing this?
>
>   val df = spark.read.option("header", "false").csv("data")
>
>
>   val tk = new Tokenizer().setInputCol("_c2").setOutputCol("words")
>
>   val tf = new HashingTF().setInputCol("words").setOutputCol("tf")
>
>   val idf = new IDF().setInputCol("tf").setOutputCol("tf-idf")
>
>
>   val df1 = tf.transform(tk.transform(df))
>
>   val idfs = idf.fit(df1).transform(df1)
>
>
>   println(nearestNeighbour("http://dbpedia.org/resource/Barack_Obama",
> idfs))
>
>
>   def nearestNeighbour(uri: String, ds: DataFrame) : String = {
>
>     var res : Row = null
>
>     var metric : Double = 0
>
>     val tfIdfSrc = ds.filter(s"_c0 == '$uri'").take(1)(0).getAs[
> Vector]("tf-idf")
>
>     ds.filter("_c0 != '" + uri + "'").foreach { r =>
>
>       val tfIdfDst = r.getAs[Vector]("tf-idf")
>
>       val dp = dorProduct(tfIdfSrc, tfIdfDst)
>
>       if (dp > metric) {
>
>         res = r
>
>         metric = dp
>
>       }
>
>     }
>
>     return res.getAs[String]("_c1")
>
>   }
>
>
>   def cosineSimilarity(vectorA: Vector, vectorB: Vector) = {
>
>     var dotProduct = 0.0
>
>     var normA = 0.0
>
>     var normB = 0.0
>
>     var index = vectorA.size - 1
>
>     for (i <- 0 to index) {
>
>       dotProduct += vectorA(i) * vectorB(i)
>
>       normA += Math.pow(vectorA(i), 2)
>
>       normB += Math.pow(vectorB(i), 2)
>
>     }
>
>     (dotProduct / (Math.sqrt(normA) * Math.sqrt(normB)))
>
>   }
>
>
>   def dorProduct(vectorA: Vector, vectorB: Vector) = {
>
>     var dp = 0.0
>
>     var index = vectorA.size - 1
>
>     for (i <- 0 to index) {
>
>       dp += vectorA(i) * vectorB(i)
>
>     }
>
>     dp
>
>   }
>
> On Sun, Nov 13, 2016 at 7:04 PM, Meeraj Kunnumpurath <
> meeraj@servicesymphony.com> wrote:
>
>> Hello,
>>
>> I have a dataset containing TF-IDF vectors for a corpus of documents. How
>> do I perform a nearest neighbour search on the dataset, using cosine
>> similarity?
>>
>>   val df = spark.read.option("header", "false").csv("data")
>>
>>   val tk = new Tokenizer().setInputCol("_c2").setOutputCol("words")
>>
>>   val tf = new HashingTF().setInputCol("words").setOutputCol("tf")
>>
>>   val idf = new IDF().setInputCol("tf").setOutputCol("tf-idf")
>>
>>   val df1 = tf.transform(tk.transform(df))
>>
>>   idf.fit(df1).transform(df1).select("tf-idf").show(10)
>> Thank you
>>
>> --
>> *Meeraj Kunnumpurath*
>>
>>
>> *Director and Executive PrincipalService Symphony Ltd00 44 7702 693597*
>>
>> *00 971 50 409 0169meeraj@servicesymphony.com
>> <meeraj@servicesymphony.com>*
>>
>
>
>
> --
> *Meeraj Kunnumpurath*
>
>
> *Director and Executive PrincipalService Symphony Ltd00 44 7702 693597*
>
> *00 971 50 409 0169meeraj@servicesymphony.com <meeraj@servicesymphony.com>*
>



-- 
*Meeraj Kunnumpurath*


*Director and Executive PrincipalService Symphony Ltd00 44 7702 693597*

*00 971 50 409 0169meeraj@servicesymphony.com <meeraj@servicesymphony.com>*

Mime
View raw message