Spark ML下实现的多分类adaboost+naivebayes算法在文本分类上的应用

 1   // pipeline for train
 2   def createPipiline(dataset: Dataset[_]): Pipeline = {
 3     // step 1 sentence 拆成 words
 4     val tokenizer = new RegexTokenizer().setInputCol("sentence").setOutputCol("words").setPattern(",")
 5     // step 2 label 转化为以0开始的labelIndex 为了适应spark.ml
 6     val indexer = new StringIndexer().setInputCol("label").setOutputCol("labelIndex").fit(dataset)
 7     // step3 统计tf词频
 8     val countModel = new CountVectorizer().setInputCol("words").setOutputCol("rawFeatures")
 9     // step4 tf-idf
10     val idfModel = new IDF().setInputCol("rawFeatures").setOutputCol("features")
11     // step5 normalize tf-idf vector
12     val normalizer = new Normalizer().setInputCol("features").setOutputCol("normalizedFeatures")
13     // step6 naive bayes model
14     val naiveBayes = new NaiveBayes().setFeaturesCol("normalizedFeatures").setLabelCol("labelIndex").setWeightCol("obsWeights").setPredictionCol("prediction").setModelType("multinomial").setSmoothing(1.0)
15     // step7 predict label to real label
16     val labelConverter = new IndexToString().setInputCol("prediction").setOutputCol("predictedLabel").setLabels(indexer.labels)
17 
18     new Pipeline().setStages(Array(tokenizer, indexer, countModel, idfModel, normalizer, naiveBayes, labelConverter))
19   }

1 val pipeline = ModelUsage.createPipeline(dataRDDTrain)
2 // train and test
3 val combinedModel = pipeline.fit(dataRDDTrain)
4 val predictResult = combinedModel.transform(dataRDDTest).select("predictedLabel", "label").rdd.map(row => (row.getDouble(0), row.getDouble(1)))
5 val evaluator = new MulticlassMetrics(predictResult)
6 
7 println("confusionMatrix:")
8 println(evaluator.confusionMatrix)
9 println(evaluator.accuracy)

 1 trait AdaboostNaiveBayesParams extends Params {
 2   // 进行adaboost时的最高迭代次数
 3   final val maxIter: IntParam = new IntParam(this, "maxIter", "max number of iterations")
 4   def getMaxIter: Int = $(maxIter)
 5   // 进行adaboost时准确率变化小于某个阈值时迭代提前终止
 6   final val threshold: DoubleParam = new DoubleParam(this, "threshold", "improvement threshold among iterations")
 7   def getThreshold: Double = $(threshold)
 8   // 朴素Bayes的平滑系数
 9   final val smoothing : DoubleParam = new DoubleParam(this, "smoothing", "naive bayes smooth")
10   def getSmoothing : Double = $(smoothing)
11   // 朴素Bayes类型"multinomial"(default) and "bernoulli"
12   final val modelType : Param[String] = new Param[String](this, "modelType", "naive bayes model type")
13   def getModelType : String = $(modelType)
14 }

  1 class AdaboostNaiveBayes(override val uid: String)
  2   extends ProbabilisticClassifier[Vector, AdaboostNaiveBayes, AdaboostNaiveBayesModel]
  3   with AdaboostNaiveBayesParams {
  4 
  5   def this() = this(Identifiable.randomUID("AdaboostNaiveBayes"))
  6 
  7   // model parameters assignment
  8   def setMaxIter(value: Int): this.type = set(maxIter, value)
  9   def setThreshold(value: Double): this.type = set(threshold, value)
 10   def setSmoothing(value: Double): this.type = set(smoothing, value)
 11   def setModelType(value: String): this.type = set(modelType, value)
 12 
 13   setMaxIter(20)
 14   setThreshold(0.02)
 15   setSmoothing(1.0)
 16   setModelType("multinomial")
 17 
 18   // method used by fit()
 19   override protected def train(dataset: Dataset[_]): AdaboostNaiveBayesModel = {
 20 
 21     val datasetSize = dataset.count().toInt
 22     val labelSize = dataset.select("label").distinct().count()
 23 
 24     // 各子模型及其权重
 25     val modelWeights = new Array[Double]($(maxIter))
 26     val modelArray = new Array[NaiveBayesModel]($(maxIter))
 27 
 28     var alpha = 0.0
 29     // 初始化各样本等权重
 30     val dataWeight: (Double, Double, Double) => Double = (obsWeight: Double, labelIndex: Double, prediction: Double) => {
 31       if (labelIndex == prediction) {
 32         obsWeight
 33       }
 34       else {
 35         obsWeight * math.exp(alpha)
 36       }
 37     }
 38     val sqlfunc = udf(dataWeight)
 39     // 初始化还没有prediction
 40     var temp = dataset.withColumn("obsWeights", lit(1.0))
 41     var i = 0
 42     var error1 = 2.0
 43     var error2 = 1.0// && (error1 - error2) > $(threshold)
 44     var weightSum = datasetSize.toDouble*datasetSize
 45 
 46     while (i < $(maxIter)) {
 47       val naiveBayes = new NaiveBayes().setFeaturesCol($(featuresCol)).setLabelCol($(labelCol)).setWeightCol("obsWeights")
 48         .setPredictionCol($(predictionCol)).setModelType($(modelType)).setSmoothing($(smoothing)).fit(temp)
 49       temp = naiveBayes.transform(temp).cache() 
 50 
 51       var error = temp.select("labelIndex", "prediction", "obsWeights").rdd.map(row => {
 52         if (row(0) != row(1))
 53           row.getDouble(2)
 54         else
 55           0.0
 56       }
 57       ).sum()/(datasetSize)
 58       val t5 = System.nanoTime()
 59       error1 = error2
 60       error2 = error
 61       alpha = Math.log((labelSize - 1) * (1 - error) / error)
 62 
 63       modelWeights(i) = alpha
 64       modelArray(i) = naiveBayes
 65       // 更新权重
 66       temp = temp.withColumn("obsWeights", sqlfunc(col("obsWeights"), col($(labelCol)), col($(predictionCol))));
 67       weightSum = temp.select("obsWeights").rdd.map(row => (row.getDouble(0))).sum()
 68       temp = temp.drop($(predictionCol), $(rawPredictionCol), $(probabilityCol))
 69       temp = temp.withColumn("obsWeights", col("obsWeights")/(weightSum/datasetSize)) 70 
 71       i += 1
 72     }
 73 
 74     new AdaboostNaiveBayesModel(uid, i, modelWeights, modelArray)
 75   }
 76 
 77   override def copy(extra: ParamMap): AdaboostNaiveBayes = defaultCopy(extra)
 78 }

 1 class AdaboostNaiveBayesModel(override val uid: String, val iternums: Int, val modelWeights: Array[Double], val modelArray: Array[NaiveBayesModel])
 2   extends ProbabilisticClassificationModel[Vector, AdaboostNaiveBayesModel]
 3   with AdaboostNaiveBayesParams {
 4 
 5   override val numClasses = modelArray(0).pi.size
 6 
 7   private def multinomialCalculation(features: Vector): Vector = {
 8     val result: Vector = new DenseVector(new Array(numClasses))
 9     for (i <- 0 until iternums) {
10       val prob: Vector = modelArray(i).theta.multiply(features)
11       prob.foreachActive { (index, value) => {
12         prob.toArray(index) = value + modelArray(i).pi(index)
13         }
14       }
15       result.toArray(prob.argmax) = result(prob.argmax) + modelWeights(i)
16     }
17     result
18   }
19 
20   override def predictRaw(features: Vector): Vector = {
21     multinomialCalculation(features)
22   }
23 
24   override def raw2probabilityInPlace(rawPrediction: Vector): Vector = {
25     rawPrediction match {
26       case dv: DenseVector =>
27         var i = 0
28         val size = dv.size
29         val maxLog = dv.values.max
30         for (i <- 0 until size) {
31           dv.values(i) = math.exp(dv.values(i) - maxLog)
32         }
33         val probSum = dv.values.sum
34 
35         for (i <- 0 until size) {
36           dv.values(i) = dv.values(i) / probSum
37         }
38         dv
39       case sv: SparseVector =>
40         throw new RuntimeException("Unexpected error in AdaboostNaiveBayesModel:" +
41         " raw2probabilityInPlace encountered SparseVector")
42     }
43   }
44　　override def copy(extra: ParamMap) = {
45　　　defaultCopy(extra)
46　　}
47　}