1、Naive Bayes classification

樸素貝葉斯分類器在文本分類中使用很廣泛，因?yàn)樗?jiǎn)單、高效，在大量的樣本集上具有較好的分類性能，但NB反應(yīng)的只是一個(gè)統(tǒng)計(jì)意義上的信息，當(dāng)每個(gè)類別信息不足時(shí)效果并不能保證，這篇文章主要是剖析NB在Spark MLlib中實(shí)現(xiàn)，以便在分類效果不好時(shí)進(jìn)行問(wèn)題分析和定位。給出NB分類的過(guò)程如下：

• 1. 設(shè) x = {a1, a2, ..., am}為待分類樣本，其中ai為樣本中的特征，針對(duì)于NLP領(lǐng)域，處理的數(shù)據(jù)均為文本，因此這里是經(jīng)過(guò)向量化之后的數(shù)據(jù)，如何將text轉(zhuǎn)換為模型可接受的數(shù)值向量會(huì)在另一篇文章中進(jìn)行介紹。
• 2）類別集合C={c1, c2,...,cn},計(jì)算各個(gè)類別的先驗(yàn)概率并取對(duì)數(shù)()，如下
  p(ci) = log(p(ci))= log((i類別的出現(xiàn)的次數(shù) + 平滑因子) / (所有類別出現(xiàn)的總次數(shù) + 平滑因子))
• 3）計(jì)算各類別下各個(gè)特征的條件概率，并取對(duì)數(shù)
  theta(i)(j) = log(sumTermFreq(j) + 平滑因子) - thetaLogDenom
  theta(i)(j)表示類別i下的第j個(gè)特征，sumTermFreq(j)表示該類別下特征j出現(xiàn)的次數(shù)，其實(shí)這里是特征j所在的這個(gè)位置的value，而這個(gè)值和向量化的方式有關(guān)，thetaLogDenom分為兩種形式，

多項(xiàng)式模式
thetaLogDenom = log(sumTermFreq.values.sum + numFeatureslambda)
二項(xiàng)式模型
thetaLogDenom = log(n + 2.0*lambda)
其中，sumTermFreq.values.sum在文本分類中解釋為，類別i下的所有單詞的總數(shù)，numFeatures表示特征數(shù)量，lambda為平滑因子，n為總的文檔/樣本數(shù)量。

2、模型訓(xùn)練

NB的主要方法run方法，該方法位于spark\mllib\classification\NaiveBayes.scala中，代碼如下：
代碼的主題思路是，先對(duì)樣本根據(jù)label進(jìn)行聚合，結(jié)果為(label, (標(biāo)簽下樣本數(shù),features之和))，然后在根據(jù)label統(tǒng)計(jì)（label, (n, sumTermFreqs)）計(jì)算條件概率和先驗(yàn)概率。

@Since("0.9.0")
class NaiveBayes private (
    private var lambda: Double,  // 平滑因子
    private var modelType: String) extends Serializable with Logging {

  import NaiveBayes.{Bernoulli, Multinomial}     // 兩種分類模式，樣本向量化的格式不同，

  @Since("1.4.0")
  def this(lambda: Double) = this(lambda, NaiveBayes.Multinomial)

  @Since("0.9.0")
  def this() = this(1.0, NaiveBayes.Multinomial)

  /** Set the smoothing parameter. Default: 1.0. */
  @Since("0.9.0")
  def setLambda(lambda: Double): NaiveBayes = {  // 設(shè)置平滑因子，默認(rèn)1.0
    require(lambda >= 0,
      s"Smoothing parameter must be nonnegative but got $lambda")
    this.lambda = lambda
    this
  }

  /** Get the smoothing parameter. */
  @Since("1.4.0")
  def getLambda: Double = lambda

  /**
   * Set the model type using a string (case-sensitive).
   * Supported options: "multinomial" (default) and "bernoulli".
   */
  @Since("1.4.0")
  def setModelType(modelType: String): NaiveBayes = { // 設(shè)置模式
    require(NaiveBayes.supportedModelTypes.contains(modelType),
      s"NaiveBayes was created with an unknown modelType: $modelType.")
    this.modelType = modelType
    this
  }

  /** Get the model type. */
  @Since("1.4.0")
  def getModelType: String = this.modelType

// NB的關(guān)鍵方法，用于模型訓(xùn)練
  @Since("0.9.0")
  def run(data: RDD[LabeledPoint]): NaiveBayesModel = {
    val requireNonnegativeValues: Vector => Unit = (v: Vector) => {
      val values = v match {    // 如果是Multinomial,向量的所有值，進(jìn)行校驗(yàn)，所有值都必須非負(fù)
        case sv: SparseVector => sv.values
        case dv: DenseVector => dv.values
      }
      if (!values.forall(_ >= 0.0)) {
        throw new SparkException(s"Naive Bayes requires nonnegative feature values but found $v.")
      }
    }

    val requireZeroOneBernoulliValues: Vector => Unit = (v: Vector) => {
      val values = v match { // 如果是Bernoulli模型，向量的所有值只能為0或1
        case sv: SparseVector => sv.values
        case dv: DenseVector => dv.values
      }
      if (!values.forall(v => v == 0.0 || v == 1.0)) {
        throw new SparkException(
          s"Bernoulli naive Bayes requires 0 or 1 feature values but found $v.")
      }
    }

    // 根據(jù)標(biāo)簽進(jìn)行聚合，并統(tǒng)計(jì)標(biāo)簽下樣本數(shù)
    val aggregated = data.map(p => (p.label, p.features)).combineByKey[(Long, DenseVector)](
      createCombiner = (v: Vector) => {   // 創(chuàng)建combiner，用于聚合vectors
        if (modelType == Bernoulli) {
          requireZeroOneBernoulliValues(v)
        } else {
          requireNonnegativeValues(v)
        }
        (1L, v.copy.toDense)  // 將樣本vector轉(zhuǎn)換為DenseVector并計(jì)次數(shù)為1，
      },
      mergeValue = (c: (Long, DenseVector), v: Vector) => {  // 創(chuàng)建合并options，用于合并vector的值
        requireNonnegativeValues(v)
        BLAS.axpy(1.0, v, c._2)  // 該方法的作用為c._2 = c._2 + v
        (c._1 + 1L, c._2)  // 計(jì)數(shù)加1，(c._1 + 1, c._2 + v)
      },
      mergeCombiners = (c1: (Long, DenseVector), c2: (Long, DenseVector)) => {
        BLAS.axpy(1.0, c2._2, c1._2)  // 用法同上，c1._2 = c1._2 + c2._2
        (c1._1 + c2._1, c1._2)
      }  // 最終的形式為(label, (樣本數(shù)，features之和))
    ).collect().sortBy(_._1)

    val numLabels = aggregated.length  // 標(biāo)簽個(gè)數(shù)
    var numDocuments = 0L
    aggregated.foreach { case (_, (n, _)) =>    // 訓(xùn)練集樣本數(shù)
      numDocuments += n
    }
  // 獲取樣本特征數(shù)即樣本向量的大小
    val numFeatures = aggregated.head match { case (_, (_, v)) => v.size }

    val labels = new Array[Double](numLabels)
    val pi = new Array[Double](numLabels)
    val theta = Array.fill(numLabels)(new Array[Double](numFeatures))

    val piLogDenom = math.log(numDocuments + numLabels * lambda)
    var i = 0
    aggregated.foreach { case (label, (n, sumTermFreqs)) =>
      labels(i) = label
      pi(i) = math.log(n + lambda) - piLogDenom  // 類別的先驗(yàn)概率
      val thetaLogDenom = modelType match { // sumTermFreqs.values.sum將vector中的所有values進(jìn)行累計(jì)
        case Multinomial => math.log(sumTermFreqs.values.sum + numFeatures * lambda)
        case Bernoulli => math.log(n + 2.0 * lambda)
        case _ =>
          // This should never happen.
          throw new UnknownError(s"Invalid modelType: $modelType.")
      }
      var j = 0
      while (j < numFeatures) {  // 計(jì)算每個(gè)特征的條件概率
        theta(i)(j) = math.log(sumTermFreqs(j) + lambda) - thetaLogDenom
        j += 1
      }
      i += 1
    }

    new NaiveBayesModel(labels, pi, theta, modelType)
  }
}

總結(jié)：spark中MLlib版本的NB，首先根據(jù)label對(duì)樣本進(jìn)行聚合，聚合的方式把樣本向量轉(zhuǎn)換為DenseVector，然后把vector.values累加，并計(jì)下該label下的樣本數(shù)，即構(gòu)成了(label， （個(gè)數(shù)， features之和）)，然后將所有個(gè)數(shù)相加就得到總的樣本數(shù)，就可以計(jì)算類別先驗(yàn)概率和特征條件概率了。

3、樣本預(yù)測(cè)