// 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

     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)

     new Pipeline().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 = new MulticlassMetrics(predictResult)

 println("confusionMatrix:")

 println(evaluator.confusionMatrix)

 println(evaluator.accuracy)

 trait AdaboostNaiveBayesParams extends Params {

   // 进行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)

 }

 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 = new Array[Double]($(maxIter))

     val modelArray = new Array[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

     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)

       val t5 = System.nanoTime()

       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

     }

     new AdaboostNaiveBayesModel(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])

   extends ProbabilisticClassificationModel[Vector, AdaboostNaiveBayesModel]

   with AdaboostNaiveBayesParams {

   override val numClasses = modelArray(0).pi.size

   private def multinomialCalculation(features: Vector): Vector = {

     val result: Vector = new DenseVector(new Array(numClasses))

     for (i <- 0 until 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

         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 <- 0 until 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)
　　}
　}