TF-IDF

1 介绍

  词频-逆文档频率法(Term frequency-inverse document frequency,TF-IDF)是在文本挖掘中广泛使用的特征向量化方法。
它反映语料中词对文档的重要程度。假设用t表示词，d表示文档，D表示语料。词频TF(t,d)表示词t在文档d中出现的次数。文档频率DF(t,D)表示语料中出现词t的文档的个数。
如果我们仅仅用词频去衡量重要程度，这很容易过分强调出现频繁但携带较少文档信息的词，如of、the等。如果一个词在语料中出现很频繁，这意味着它不携带特定文档的特殊信息。逆文档频率数值衡量一个词提供多少信息。

1.1

  如果某个词出现在所有的文档中，它的IDF值为0。注意，上式有个平滑项，这是为了避免分母为0的情况发生。TF-IDF就是TF和IDF简单的相乘。

1.2

  词频和文档频率的定义有很多种不同的变种。在Mllib中，分别提供了TF和IDF的实现，以便有更好的灵活性。

  Mllib使用hashing trick实现词频。元素的特征应用一个hash函数映射到一个索引（即词），通过这个索引计算词频。这个方法避免计算全局的词-索引映射，因为全局的词-索引映射在大规模语料中花费较大。
但是，它会出现哈希冲突，这是因为不同的元素特征可能得到相同的哈希值。为了减少碰撞冲突，我们可以增加目标特征的维度，例如哈希表的桶数量。默认的特征维度是1048576。

2 实例

  • TF的计算
  1. import org.apache.spark.rdd.RDD
  2. import org.apache.spark.SparkContext
  3. import org.apache.spark.mllib.feature.HashingTF
  4. import org.apache.spark.mllib.linalg.Vector
  5. val sc: SparkContext = ...
  6. // Load documents (one per line).
  7. val documents: RDD[Seq[String]] = sc.textFile("...").map(_.split(" ").toSeq)
  8. val hashingTF = new HashingTF()
  9. val tf: RDD[Vector] = hashingTF.transform(documents)
  • IDF的计算
  1. import org.apache.spark.mllib.feature.IDF
  2. // ... continue from the previous example
  3. tf.cache()
  4. val idf = new IDF().fit(tf)
  5. val tfidf: RDD[Vector] = idf.transform(tf)
  6. //或者
  7. val idf = new IDF(minDocFreq = 2).fit(tf)
  8. val tfidf: RDD[Vector] = idf.transform(tf)

3 源码实现

  下面分别分析HashingTF和IDF的实现。

3.1 HashingTF

  1. def transform(document: Iterable[_]): Vector = {
  2.     val termFrequencies = mutable.HashMap.empty[Int, Double]
  3.     document.foreach { term =>
  4.       val i = indexOf(term)
  5.       termFrequencies.put(i, termFrequencies.getOrElse(i, 0.0) + 1.0)
  6.     }
  7.     Vectors.sparse(numFeatures, termFrequencies.toSeq)
  8.   }

  以上代码中，indexOf方法使用哈希获得索引。

  1. //为了减少碰撞，将numFeatures设置为1048576
  2. def indexOf(term: Any): Int = Utils.nonNegativeMod(term.##, numFeatures)
  3. def nonNegativeMod(x: Int, mod: Int): Int = {
  4.     val rawMod = x % mod
  5.     rawMod + (if (rawMod < 0) mod else 0)
  6.   }

  这里的term.##等价于term.hashCode，得到哈希值之后，作取余操作得到相应的索引。

3.2 IDF

  我们先看IDF的fit方法。

  1. def fit(dataset: RDD[Vector]): IDFModel = {
  2.     val idf = dataset.treeAggregate(new IDF.DocumentFrequencyAggregator(
  3.           minDocFreq = minDocFreq))(
  4.       seqOp = (df, v) => df.add(v),
  5.       combOp = (df1, df2) => df1.merge(df2)
  6.     ).idf()
  7.     new IDFModel(idf)
  8.   }

  该函数使用treeAggregate处理数据集，生成一个DocumentFrequencyAggregator对象，它用于计算文档频率。重点看add和merge方法。

  1. def add(doc: Vector): this.type = {
  2.       if (isEmpty) {
  3.         df = BDV.zeros(doc.size)
  4.       }
  5.       //计算
  6.       doc match {
  7.         case SparseVector(size, indices, values) =>
  8.           val nnz = indices.size
  9.           var k = 0
  10.           while (k < nnz) {
  11.             if (values(k) > 0) {
  12.               df(indices(k)) += 1L
  13.             }
  14.             k += 1
  15.           }
  16.         case DenseVector(values) =>
  17.           val n = values.size
  18.           var j = 0
  19.           while (j < n) {
  20.             if (values(j) > 0.0) {
  21.               df(j) += 1L
  22.             }
  23.             j += 1
  24.           }
  25.         case other =>
  26.           throw new UnsupportedOperationException
  27.       }
  28.       m += 1L
  29.       this
  30.     }

  df这个向量的每个元素都表示该索引对应的词出现的文档数。m表示文档总数。

  1. def merge(other: DocumentFrequencyAggregator): this.type = {
  2.       if (!other.isEmpty) {
  3.         m += other.m
  4.         if (df == null) {
  5.           df = other.df.copy
  6.         } else {
  7.           //简单的向量相加
  8.           df += other.df
  9.         }
  10.       }
  11.       this
  12.     }

  treeAggregate方法处理完数据之后，调用idf方法将文档频率低于给定值的词的idf置为0，其它的按照上面的公式计算。

  1.  def idf(): Vector = {
  2.       val n = df.length
  3.       val inv = new Array[Double](n)
  4.       var j = 0
  5.       while (j < n) {
  6.         if (df(j) >= minDocFreq) {
  7.           //计算得到idf
  8.           inv(j) = math.log((m + 1.0) / (df(j) + 1.0))
  9.         }
  10.         j += 1
  11.       }
  12.       Vectors.dense(inv)
  13. }

  最后使用transform方法计算tfidf值。

  1.  //这里的dataset指tf
  2.  def transform(dataset: RDD[Vector]): RDD[Vector] = {
  3.     val bcIdf = dataset.context.broadcast(idf)
  4.     dataset.mapPartitions(iter => iter.map(v => IDFModel.transform(bcIdf.value, v)))
  5.   }
  6.  def transform(idf: Vector, v: Vector): Vector = {
  7.      val n = v.size
  8.      v match {
  9.        case SparseVector(size, indices, values) =>
  10.          val nnz = indices.size
  11.          val newValues = new Array[Double](nnz)
  12.          var k = 0
  13.          while (k < nnz) {
  14.            //tf-idf = tf * idf
  15.            newValues(k) = values(k) * idf(indices(k))
  16.            k += 1
  17.          }
  18.          Vectors.sparse(n, indices, newValues)
  19.        case DenseVector(values) =>
  20.          val newValues = new Array[Double](n)
  21.          var j = 0
  22.          while (j < n) {
  23.            newValues(j) = values(j) * idf(j)
  24.            j += 1
  25.          }
  26.          Vectors.dense(newValues)
  27.        case other =>
  28.          throw new UnsupportedOperationException
  29.      }
  30.    }