基于Spark的TF-IDF算法的中文文本相似度实现

Spark version：spark 2.2.0 Hadoop version:Hadoop 2.6.5 Scala
version:scala 2.11
 ansj version：5.1.5

第一、先说下应用场景吧，用户给出一段文字然后我返回十个与这段文字最相似的文件名称。

第二、什么是TF-IDF算法？我就简单介绍一下，因为百度上也有许多的介绍，TF-IDF用中文来解释就是词频、逆文档频率的意思，TF-IDF体现了词项与文档关联度的直观理解，一个词在文档中出现的越多则越重要，但是词项是不平等的文档中出现罕见词项的意义比常见词更大，因此就要取词项出现次数的倒数，词项在语料库中的分布式呈指数型的一个常用词出现次数往往是罕见词的数十倍，假如直接除以原始文档的频率则罕见词的权重就会过大，所以算法应对逆文档率取对数，让文档频率差别由乘数级变为加数级。

第三、由于spark MLlib库已经有TF-IDF算法的实现我们就不亲自实现了，而是去调用它的，对于整个功能的逻辑如下：

（1）获取数据、（2）用中文分词工具分词（这里使用的是ansj）、（3）计算TF、IDF、（4）计算向量间的余弦相似度

废话不多说直接上代码：

import org.ansj.recognition.impl.StopRecognition
import org.ansj.splitWord.analysis.ToAnalysis
import org.apache.spark.mllib.feature.{ HashingTF, IDF }
import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.mllib.linalg.{ SparseVector => SV }

import scala.collection.mutable.ArrayBuffer
import scala.util.matching.Regex
object TfIdfs {

  def main(args: Array[String]): Unit = {
    val conf =new SparkConf().setAppName("tdidf")
    val sc =new SparkContext(conf)
//读取2600份法律案例
    valrdd =
sc.wholeTextFiles("hdfs://master:8020/data2/*")

    val text =rdd.map {
case (file,text) =>
text }

    val title =rdd.map {
case (title,text) =>
title }.collect()

    val dim = math.pow(2,18).toInt
    val hashingTF =new HashingTF(dim)

    val filter =new StopRecognition()
    filter.insertStopNatures("w")//过滤掉标点
//使用ansj对文本进行分词
    val tokens2 =text.map(doc =>
ToAnalysis.parse(doc).recognition(filter).toStringWithOutNature(" ").split(" ").toSeq)
//tf计算
    val tf =hashingTF.transform(tokens2)

    // cache data in memory
    tf.cache
    //idf计算
    val idf =new IDF().fit(tf)
    val tfidf =idf.transform(tf)
    val vectorArr =tfidf.collect()
//需要匹配相似度的文本
    valrdd3 =
sc.parallelize(Seq("被告人王乃胜，高中文化，户籍所在地:白山市，居住地:白山市。"))

val predictTF2 =rdd3.map(doc =>
hashingTF.transform(ToAnalysis.parse(doc).recognition(filter).toStringWithOutNature(" ").split("
").toSeq))

    valpredictTfIdf =
idf.transform(predictTF2)

    import breeze.linalg._
    val predictSV =predictTfIdf.first.asInstanceOf[SV]
    val c =new ArrayBuffer[(String, Double)]()
  
    valbreeze1 =
new SparseVector(predictSV.indices,predictSV.values,predictSV.size)
    var tag =
4000
0
   
    for (i <-vectorArr) {

      val Svector =i.asInstanceOf[SV]
      val breeze2 =new SparseVector(Svector.indices,Svector.values,
Svector.size)
      valcosineSim =
breeze1.dot(breeze2) / (norm(breeze1) * norm(breeze2))
      c ++= Array((title(tag),cosineSim))
      tag +=
1

    }
    val cc =c.toArray.sortBy(_._2).reverse.take(10)
    println(cc.toBuffer)
    sc.stop()

  }
}