spark 余弦相似度

余弦相似度是一种用于度量两个向量之间相似度的方法。在Spark中，可以使用MLlib库中的CosineSimilarity来计算两个向量之间的余弦相似度。下面是一个使用Spark MLlib计算余弦相似度的示例代码：

import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.ml.linalg.DenseVector
import org.apache.spark.ml.feature.Normalizer
import org.apache.spark.ml.feature.StandardScaler
import org.apache.spark.ml.linalg.Vectors

val data = Seq(
      (0, "hello world"),
      (1, "hello spark"),
      (2, "hello hadoop"),
      (3, "hello hadoop spark"),
      (4, "hello world hadoop")
    ).toDF("id", "text")

val tokenizer = new Tokenizer()
      .setInputCol("text")
      .setOutputCol("words")

val wordsData = tokenizer.transform(data)

val hashingTF = new HashingTF()
      .setInputCol("words")
      .setOutputCol("rawFeatures")
      .setNumFeatures(20)

val featurizedData = hashingTF.transform(wordsData)

val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
val idfModel = idf.fit(featurizedData)
val rescaledData = idfModel.transform(featurizedData)

val vectors = rescaledData.select("features").rdd.map(r => r(0).asInstanceOf[Vector]).collect()
val denseVectors = vectors.map(v => new DenseVector(v.toArray))

val normalize = new Normalizer()
      .setInputCol("features")
      .setOutputCol("normFeatures")
      .setP(2.0)

val normalizedData = normalize.transform(rescaledData)

val scaler = new StandardScaler()
      .setInputCol("features")
      .setOutputCol("scaledFeatures")
      .setWithStd(true)
      .setWithMean(false)

val scaledData = scaler.fit(normalizedData).transform(normalizedData)

val bVectors = spark.sparkContext.broadcast(denseVectors)

val cosineSimilarity = scaledData.map { row =>
      val vector = row.getAs[Vector]("scaledFeatures")
      val bDenseVectors = bVectors.value
      val similarities = bDenseVectors.map(dv => (dv, dv.dot(vector))).sortBy(-_._2)
      (row.getAs[Int]("id"), similarities)
    }

cosineSimilarity.collect().foreach(println)

该示例代码使用Tokenizer将文本数据分词，并使用HashingTF将词汇转换为特征向量。然后，使用IDF对特征向量进行加权，最后使用Normalizer和StandardScaler对特征向量进行标准化和缩放。最后，使用Spark的广播变量，将所有特征向量存储在内存中，并使用map函数计算每个向量与所有其他向量之间的余弦相似度。