朴素貝葉斯算法源碼分析及代碼實戰【python sklearn/spark ML】

一.簡介　　
　　貝葉斯定理是關於隨機事件A和事件B的條件概率的一個定理。通常在事件A發生的前提下事件B發生的概率，與在事件B發生的前提下事件A發生的概率是不一致的。然而，這兩者之間有確定的
關系，貝葉斯定理就是這種關系的陳述。其中，L(A|B)表示在B發生的前提下，A發生的概率。L表示要取對數的意思。
　　關鍵詞解釋：
　　　　1.p(A),p(B)表示A，B發生的概率，也稱先驗概率或邊緣概率。
　　　　2.p(B|A)表示在A發生的前提下，B發生的概率，也稱后驗概率。
　　基本公式：p(A|B) = p(AB)/p(B)
　　圖解：

　　　　　　

　　備注：p(AB) = p(BA)都是指A,B同時發生的概率，所以可得貝葉斯公式：p(B|A) = p(AB)/p(A) = p(A|B)p(B)/p(A)導入數據得 = 0.5*0.4/0.8 = 0.25
　　貝葉斯公式：p(B|A) = p(A|B)p(B)/p(A)
　　圖解：同上
　　朴素貝葉斯分類是一種十分簡單的分類算法，其算法基礎是對於給出的待分類項，求解在此項出現的條件下各類別出現的概率，哪個最大，就認為此待分類項屬於哪個類別。
　　實現步驟：

　　　　   　

二.代碼實現【python】

 1 # -*- coding: utf-8 -*-
 2 """
 3 Created on Tue Oct 28 14:40:38 2018
 4 
 5 @author: zhen
 6 """
 7 from sklearn.datasets import fetch_20newsgroups
 8 from sklearn.model_selection import train_test_split
 9 from sklearn.feature_extraction.text import CountVectorizer
10 from sklearn.naive_bayes import MultinomialNB
11 from sklearn.metrics import classification_report
12 # 數據獲取
13 news = fetch_20newsgroups(subset='all')
14 
15 # 數據預處理:分割訓練集和測試集
16 x_train, x_test, y_train, y_test = train_test_split(news.data, news.target, test_size=0.25, random_state=33)
17 # 文本特征向量化
18 vec = CountVectorizer()
19 x_train = vec.fit_transform(x_train)
20 x_test = vec.transform(x_test)
21 
22 # 使用朴素貝葉斯進行訓練(多項式模型)
23 mnb = MultinomialNB()
24 mnb.fit(x_train, y_train)
25 y_predict = mnb.predict(x_test)
26 
27 # 獲取預測結果
28 print(classification_report(y_test, y_predict, target_names = news.target_names))
29 print("the accuracy of MultinomialNB is:", mnb.score(x_test, y_test))

三.結果【python】

　　

 四.代碼實現【Spark】

package big.data.analyse.ml

import org.apache.log4j.{Level,Logger}
import org.apache.spark.NaiveBayes
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.feature.LabeledPoint
import org.apache.spark.sql.{SparkSession}

/**
  * Created by zhen on 2019/9/9.
  */
object NaiveBayesAnalyse {
  Logger.getLogger("org").setLevel(Level.WARN)
  def main(args: Array[String]) {
    val spark = SparkSession.builder().appName("NaiveBayes").master("local[2]").getOrCreate()

    /**
      * 加載數據
      */
    val test_data_array = Array("0,1.2-0.5-0.2","0,2.1-0.3-0.2","0,3.6-0.1-1.0","0,4.6-0.3-0.2",
          "1,0.4-1.5-0.2","1,0.2-2.6-0.8","1,0.6-3.3-0.6","1,0.1-4.3-0.4",
          "2,0.1-0.4-1.8","2,0.2-0.4-2.1","2,0.3-0.1-3.3","2,0.5-0.2-4.1")
    
    val sc = spark.sparkContext
    val test_data = sc.parallelize(test_data_array).map(row => {
      val array = row.split(",")
      LabeledPoint(array(0).toDouble,Vectors.dense(array(1).split("-").map(_.toDouble)))
    })

    /**
      * 拆分數據為訓練數據和測試數據
      */
    val splits = test_data.randomSplit(Array(0.8, 0.2), seed = 11L)
    val train = splits(0)
    val test = splits(1)

    /**
      * 創建朴素貝葉斯模型並訓練
      * 使用多項式模型
      */
    val model = NaiveBayes.train(train, lambda = 1.0, modelType = "multinomial")

    /**
      * 預測
      */
    val predict = test.map(row => (row.label, model.predict(row.features)))
    val predict_show = predict.take(20)
    val test_take = test.take(20)
    println("預測結果：")
    println("label" + "\t" + "features" + "\t" + "predict")
    for(i <- 0 until predict_show.length){
      println(predict_show(i)._1 + "\t" + test_take(i).features +"\t" +  predict_show(i)._2)
    }

    val acc = 1.0 * predict.filter(row => row._1 == row._2).count() / test.count()
    println("預測准確度："+acc)
  }
}

五.模擬源碼實現【Spark】

NaiveBayes朴素貝葉斯類：

package org.apache.spark

import org.apache.spark.ml.feature.LabeledPoint
import org.apache.spark.ml.linalg.{BLAS, DenseVector, SparseVector, Vector}
import org.apache.spark.rdd.RDD

/**
  * Created by zhen on 2019/9/11.
  */
object NaiveBayes{
  /**
    * 多項式模型類別
    */
  val Multinomial : String = "multinomial"

  /**
    * 伯努利模式類型
    */
  val Bernoulli : String = "bernoulli"

  /**
    * 設置模型支持的類別
    */
  val supportedModelTypes = Set(Multinomial, Bernoulli)

  /**
    * 訓練一個朴素貝葉斯模型
    * @param input 樣本RDD
    * @return
    */
  def train(input : RDD[LabeledPoint]) : NaiveBayesModel = {
    new NaiveBayes().run(input)
  }

  /**
    * 訓練一個朴素貝葉斯模型
    * @param input 樣本RDD
    * @param lambda 平滑系數
    * @return
    */
  def train(input : RDD[LabeledPoint], lambda : Double) : NaiveBayesModel = {
    new NaiveBayes(lambda, Multinomial).run(input)
  }

  /**
    * 訓練一個朴素貝葉斯模型
    * @param input 樣本RDD
    * @param lambda 平滑系數
    * @param modelType 模型類型
    * @return
    */
  def train(input : RDD[LabeledPoint], lambda : Double, modelType : String) : NaiveBayesModel = {
    require(supportedModelTypes.contains(modelType), s"NaiveBayes was created with an unknown modelType:$modelType.")
    new NaiveBayes(lambda, modelType).run(input)
  }
}

/**
  * 貝葉斯分類類
  * @param lambda 平滑系數
  * @param modelType 模型類型
  */
class NaiveBayes private(private var lambda : Double,
                         private var modelType : String) extends Serializable{

  import NaiveBayes.{Bernoulli, Multinomial}

  def this(lambda : Double) = this(lambda, NaiveBayes.Multinomial)

  def this() = this(1.0, NaiveBayes.Multinomial)

  /**
    * 設置平滑參數
    * @param lambda
    * @return
    */
  def setLambda(lambda : Double) : NaiveBayes = {
    this.lambda = lambda
    this
  }

  /**
    * 獲取平滑參數
    * @return
    */
  def getLambda : Double = this.lambda

  /**
    * 設置模型類型
    * @param modelType
    * @return
    */
  def setModelType(modelType : String) : NaiveBayes = {
    require(NaiveBayes.supportedModelTypes.contains(modelType), s"NaiveBayes was created with an unknown modelType :$modelType.")
    this.modelType = modelType
    this
  }

  /**
    * 獲取模型類型
    * @return
    */
  def getModelType : String  = this.modelType

  /**
    * 運行算法
    * @param data
    * @return
    */
  def run(data : RDD[LabeledPoint]) : NaiveBayesModel = {
    val requireNonnegativeValues : Vector => Unit = (v : Vector) => {
      val values = v match {
        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{
        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.")
      }
    }


    /**
      * 對每個標簽進行聚合操作，求得每個標簽標簽對應特征的頻數
      * 以label為key，聚合同一個label的features，返回(label, (計數, features之和))
      */
    println("訓練數據：")
    data.foreach(println)
    val aggregated = data.map(row => (row.label, row.features))
　　　　　　.combineByKey[(Long, DenseVector)](
      createCombiner = (v : Vector) => { //完成樣本從v到c的轉化：(v:Vector) -> (c:(Long, DenseVector))
        if(modelType == Bernoulli){
          requireZeroOneBernoulliValues(v)
        }else{
          requireNonnegativeValues(v)
        }
        (1L, v.copy.toDense)
      },
      mergeValue = (c : (Long, DenseVector), v : Vector) => { // 合並
        requireNonnegativeValues(v)
        BLAS.axpy(1.0, v, c._2)
        (c._1 + 1L, c._2)
      },
      mergeCombiners = (c1 : (Long, DenseVector), c2 : (Long, DenseVector)) => {
        BLAS.axpy(1.0, c2._2, c1._2)
        (c1._1 + c2._1, c1._2)
      }
    ).collect()

    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) // pi類別的先驗概率

    val theta = Array.fill(numLabels)(new Array[Double](numFeatures)) // theta各個特征在類別中的條件概率

    val piLogDenom = math.log(numDocuments + numLabels * lambda) //聚合計算theta

    var i = 0
    aggregated.foreach{case (label, (n, sumTermFreqs)) =>
      labels(i) = label
        pi(i) = math.log(n + lambda) - piLogDenom // 計算先驗概率，並取log
      val thetaLogDenom = modelType match {
        case Multinomial => math.log(sumTermFreqs.values.sum + numFeatures * lambda) // 多項式模型
        case Bernoulli => math.log(n + 2.0 * lambda) // 伯努利模型
        case _ => throw new UnknownError(s"Invalid modeType: $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)
  }
}　　

NaiveBayesModel朴素貝葉斯模型類：

package org.apache.spark

import org.apache.spark.ml.linalg.{BLAS, Vector, DenseMatrix, DenseVector}
import org.apache.spark.mllib.util.Saveable
import org.apache.spark.rdd.RDD

/**
  * Created by zhen on 2019/9/12.
  */
class NaiveBayesModel private[spark](
                                      val labels : Array[Double],
                                      val pi : Array[Double],
                                      val theta : Array[Array[Double]],
                                      val modelType : String
                                    ) extends Serializable with Saveable{

  import NaiveBayes.{Bernoulli, Multinomial, supportedModelTypes}

  private val piVector = new DenseVector(pi) // 類別的先驗概率

  private val thetaMatrix = new DenseMatrix(labels.length, theta(0).length, theta.flatten, true) // 各個特征在各個類別的條件概率

  private[spark] def this(labels:Array[Double], pi:Array[Double], theta:Array[Array[Double]]) = this(labels, pi, theta, NaiveBayes.Multinomial)

  /**
    * java接口的構造函數
    */
  private[spark] def this(
                           labels : Iterable[Double],
                           pi : Iterable[Double],
                           theta : Iterable[Iterable[Double]]
                         ) = this(labels.toArray, pi.toArray, theta.toArray.map(_.toArray))

  require(supportedModelTypes.contains(modelType), s"Invalid modelType $modelType.Supported modelTypes are $supportedModelTypes.")

  /**
    * 伯努利模型額外處理
    */
  private val (thetaMinusNegTheta, negThetaSum) = modelType match {
    case Multinomial => (None, None)
    case Bernoulli =>
      val negTheta = thetaMatrix.map(value => math.log(1.0 - math.exp(value)))
      val ones = new DenseVector(Array.fill(thetaMatrix.numCols){1.0})
      val thetaMinusNegTheta = thetaMatrix.map{value => value - math.log(1.0 - math.exp(value))}
      (Option(thetaMinusNegTheta), Option(negTheta.multiply(ones)))
    case _ => throw new UnknownError(s"Involid modelType: $modelType.")
  }

  /**
    * 對樣本RDD進行預測
    */
  def predict(testData : RDD[Vector]) : RDD[Double] = {
    val bcModel = testData.context.broadcast(this)
    testData.mapPartitions{ iter =>
      val model = bcModel.value
      iter.map(model.predict) // 調用參數為一個樣本的predict
    }
  }

  /**
    * 根據一個樣本的特征向量進行預測
    */
  def predict(testData : Vector) : Double = {
    modelType match {
      case Multinomial =>
        val prob = thetaMatrix.multiply(testData)
        RBLAS.axpy(1.0, piVector, prob)
        labels(prob.argmax)
      case Bernoulli =>
        testData.foreachActive{(index, value) =>
          if(value != 0.0 && value != 1.0){
            throw new SparkException(s"Bernouslli naive Bayes requires 0 or 1 feature values but found $testData.")
          }
        }
        val prob = thetaMinusNegTheta.get.multiply(testData)
        BLAS.axpy(1.0, piVector, prob)
        BLAS.axpy(1.0, negThetaSum.get, prob)
        labels(prob.argmax)
      case _ =>
        throw new UnknownError(s"Involid modelType: $modelType.")
    }
  }

  /**
    * 保存模型
    */
  def save(sc : SparkContext, path : String) : Unit = {
    //val data = NaiveBayesModel.SaveLoadV2_0.Data(labels, pi, theta, modelType)
    //NaiveBayesModel.SaveLoadV2_0.save(sc, path, data)
  }

  override protected def formatVersion : String = "2.0"
}

六.結果【Spark】