1、Naive Bayes classification
朴素贝叶斯分类器在文本分类中使用很广泛，因为他简单、高效，在大量的样本集上具有较好的分类性能，但NB反应的只是一个统计意义上的信息，当每个类别信息不足时效果并不能保证，这篇文章主要是剖析NB在Spark MLlib中实现，以便在分类效果不好时进行问题分析和定位。给出NB分类的过程如下：
- 设 x = {a1, a2, ..., am}为待分类样本，其中ai为样本中的特征，针对于NLP领域，处理的数据均为文本，因此这里是经过向量化之后的数据，如何将text转换为模型可接受的数值向量会在另一篇文章中进行介绍。
- 2）类别集合C={c1, c2,...,cn},计算各个类别的先验概率并取对数()，如下
p(ci) = log(p(ci))= log((i类别的出现的次数 + 平滑因子) / (所有类别出现的总次数 + 平滑因子)) - 3）计算各类别下各个特征的条件概率，并取对数
theta(i)(j) = log(sumTermFreq(j) + 平滑因子) - thetaLogDenom
theta(i)(j)表示类别i下的第j个特征，sumTermFreq(j)表示该类别下特征j出现的次数，其实这里是特征j所在的这个位置的value，而这个值和向量化的方式有关，thetaLogDenom分为两种形式，
多项式模式
thetaLogDenom = log(sumTermFreq.values.sum + numFeatureslambda)
二项式模型
thetaLogDenom = log(n + 2.0*lambda)
其中，sumTermFreq.values.sum在文本分类中解释为，类别i下的所有单词的总数，numFeatures表示特征数量，lambda为平滑因子，n为总的文档/样本数量。
2、模型训练
NB的主要方法run方法，该方法位于spark\mllib\classification\NaiveBayes.scala中，代码如下：
代码的主题思路是，先对样本根据label进行聚合，结果为(label, (标签下样本数,features之和))，然后在根据label统计（label, (n, sumTermFreqs)）计算条件概率和先验概率。
@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 = { // 设置平滑因子，默认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 = { // 设置模式
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的关键方法，用于模型训练
@Since("0.9.0")
def run(data: RDD[LabeledPoint]): NaiveBayesModel = {
val requireNonnegativeValues: Vector => Unit = (v: Vector) => {
val values = v match { // 如果是Multinomial,向量的所有值，进行校验，所有值都必须非负
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.")
}
}
// 根据标签进行聚合，并统计标签下样本数
val aggregated = data.map(p => (p.label, p.features)).combineByKey[(Long, DenseVector)](
createCombiner = (v: Vector) => { // 创建combiner，用于聚合vectors
if (modelType == Bernoulli) {
requireZeroOneBernoulliValues(v)
} else {
requireNonnegativeValues(v)
}
(1L, v.copy.toDense) // 将样本vector转换为DenseVector并计次数为1，
},
mergeValue = (c: (Long, DenseVector), v: Vector) => { // 创建合并options，用于合并vector的值
requireNonnegativeValues(v)
BLAS.axpy(1.0, v, c._2) // 该方法的作用为c._2 = c._2 + v
(c._1 + 1L, c._2) // 计数加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, (样本数，features之和))
).collect().sortBy(_._1)
val numLabels = aggregated.length // 标签个数
var numDocuments = 0L
aggregated.foreach { case (_, (n, _)) => // 训练集样本数
numDocuments += n
}
// 获取样本特征数即样本向量的大小
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 // 类别的先验概率
val thetaLogDenom = modelType match { // sumTermFreqs.values.sum将vector中的所有values进行累计
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) { // 计算每个特征的条件概率
theta(i)(j) = math.log(sumTermFreqs(j) + lambda) - thetaLogDenom
j += 1
}
i += 1
}
new NaiveBayesModel(labels, pi, theta, modelType)
}
}
总结：spark中MLlib版本的NB，首先根据label对样本进行聚合，聚合的方式把样本向量转换为DenseVector，然后把vector.values累加，并计下该label下的样本数，即构成了(label， （个数， features之和）)，然后将所有个数相加就得到总的样本数，就可以计算类别先验概率和特征条件概率了。
