Spark ML下实现的多分类AdaBoost + NaiveBayes算法 及其在文本分类上的应用

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

val pipeline = ModelUsage.createPipeline(dataRDDTrain) 
// train and test 
val combinedModel = pipeline.fit(dataRDDTrain) 
val predictResult = combinedModel.transform(dataRDDTest).select("predictedLabel", "label").rdd.map(row => (row.getDouble(0), row.getDouble(1))) 
val evaluator = newMulticlassMetrics(predictResult)

println("confusionMatrix:") 
println(evaluator.confusionMatrix) 
println(evaluator.accuracy)

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

class AdaboostNaiveBayes(override val uid: String) extends ProbabilisticClassifier[Vector, AdaboostNaiveBayes, AdaboostNaiveBayesModel] with AdaboostNaiveBayesParams { 
    def this() = this(Identifiable.randomUID("AdaboostNaiveBayes")) 
    //model parameters assignment 
    def setMaxIter(value: Int): this.type =set(maxIter, value) 
    def setThreshold(value: Double): this.type =set(threshold, value)  
    def setSmoothing(value: Double): this.type =set(smoothing, value) 
    def setModelType(value: String): this.type =set(modelType, value)

    setMaxIter(20) 
    setThreshold(0.02) 
    setSmoothing(1.0) 
    setModelType("multinomial")

    //method used by fit()
    override protected def train(dataset: Dataset[_]): AdaboostNaiveBayesModel ={ 
        val datasetSize =dataset.count().toInt 
        val labelSize = dataset.select("label").distinct().count()

        //各子模型及其权重 
        val modelWeights = newArray[Double]($(maxIter))  
        val modelArray = newArray[NaiveBayesModel]($(maxIter)) 
        var alpha = 0.0 
        //初始化各样本等权重 
        val dataWeight: (Double, Double, Double) => Double = (obsWeight: Double, labelIndex: Double, prediction: Double) =>{ 
            if (labelIndex ==prediction) { 
                obsWeight 
            }
            else{ 
                obsWeight *math.exp(alpha) 
            }
        } 
        val sqlfunc =udf(dataWeight) 
        //初始化还没有prediction 
        var temp = dataset.withColumn("obsWeights", lit(1.0)) 
        var i = 0 42 var error1 = 2.0 
        var error2 = 1.0//&& (error1 - error2) > $(threshold) 
        var weightSum = datasetSize.toDouble*datasetSize

        while (i <$(maxIter)) { 
            val naiveBayes = new NaiveBayes().setFeaturesCol($(featuresCol)).setLabelCol($(labelCol)).setWeightCol("obsWeights").setPredictionCol($(predictionCol)).setModelType($(modelType)).setSmoothing($(smoothing)).fit(temp) 
            temp =naiveBayes.transform(temp).cache() 
            var error = temp.select("labelIndex", "prediction", "obsWeights").rdd.map(row =>{ 
                if (row(0) != row(1)) 
                    row.getDouble(2) 
                else 
                    0.0
            }).sum()/(datasetSize) 
        error1 =error2 
        error2 =error 
        alpha = Math.log((labelSize - 1) * (1 - error) /error) 
        modelWeights(i) =alpha 
        modelArray(i) =naiveBayes 
        //更新权重 
        temp = temp.withColumn("obsWeights", sqlfunc(col("obsWeights"), col($(labelCol)), col($(predictionCol)))); 
        weightSum = temp.select("obsWeights").rdd.map(row => (row.getDouble(0))).sum() 
        temp =temp.drop($(predictionCol), $(rawPredictionCol), $(probabilityCol)) 
        temp = temp.withColumn("obsWeights", col("obsWeights")/(weightSum/datasetSize))

        i += 1 } 
        newAdaboostNaiveBayesModel(uid, i, modelWeights, modelArray)  } 
        override def copy(extra: ParamMap): AdaboostNaiveBayes =defaultCopy(extra) }

class AdaboostNaiveBayesModel(override val uid: String, val iternums: Int, val modelWeights: Array[Double], val modelArray: Array[NaiveBayesModel]) extendsProbabilisticClassificationModel[Vector, AdaboostNaiveBayesModel] with AdaboostNaiveBayesParams { 
    override val numClasses = modelArray(0).pi.size 
    private def multinomialCalculation(features: Vector): Vector = { 
        val result: Vector = new DenseVector(newArray(numClasses)) 
        for (i <- 0until iternums) { 
            val prob: Vector =modelArray(i).theta.multiply(features) 
            prob.foreachActive { (index, value) =>{ 
                prob.toArray(index) = value +modelArray(i).pi(index) 
            }   
        } 
        result.toArray(prob.argmax) = result(prob.argmax) +modelWeights(i) 
    } 
    result
} 
override def predictRaw(features: Vector): Vector ={ 
    multinomialCalculation(features) 
} 
override def raw2probabilityInPlace(rawPrediction: Vector): Vector = { 
    rawPrediction match { 
        case dv: DenseVector => 
            var i = 0 28 val size = dv.size 
            val maxLog = dv.values.max 
            for (i <- 0 until size) { 
                dv.values(i) = math.exp(dv.values(i) - maxLog)
            } 
            val probSum = dv.values.sum 
            for (i <- 0until size) { 
                dv.values(i) = dv.values(i) /probSum
            } 
            dv 
        case sv: SparseVector => throw new RuntimeException("Unexpected error in AdaboostNaiveBayesModel:  raw2probabilityInPlace encountered SparseVector")}}
    override def copy(extra: ParamMap) = {
        defaultCopy(extra)
    }
}