import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public JavaRDD<Tuple2<Object, Object>> predictForMetrics(String modelName, T model, JavaRDD<LabeledPoint> data, int numClasses){
    JavaRDD<Tuple2<Object, Object>> predictionAndLabels = null;
    if(modelName.equals("LogisticRegressionModel")){
      LogisticRegressionModel lrmodel = (LogisticRegressionModel) model;      
      if(numClasses==2){
        lrmodel.clearThreshold();
      } 
      //Predict
      predictionAndLabels = PredictUnit.predictForMetrics_LogisticRegressionModel(lrmodel, data);
    }
    else if(modelName.equals("SVMModel")){
      SVMModel svmmodel = (SVMModel) model;      
      if(numClasses==2){
        svmmodel.clearThreshold();
      }     
      //Predict
      predictionAndLabels = PredictUnit.predictForMetrics_SVMModel(svmmodel, data);
    }
    else if(modelName.equals("NaiveBayesModel")){
      NaiveBayesModel bayesmodel = (NaiveBayesModel) model;      
      //Predict
      predictionAndLabels = PredictUnit.predictForMetrics_NaiveBayesModel(bayesmodel, data);
    }
    return predictionAndLabels;
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public JavaRDD<Tuple2<Object, Object>> predictForOutput(String modelName, T model, JavaRDD<LabeledPoint> data, int numClasses, double threshold){
    JavaRDD<Tuple2<Object, Object>> FeaturesAndPrediction = null;
    if(modelName.equals("LogisticRegressionModel")){
      LogisticRegressionModel lrmodel = (LogisticRegressionModel) model; 
      if(numClasses==2){
        lrmodel.setThreshold(threshold);
      }  
      //Predict
      FeaturesAndPrediction = PredictUnit.predictForOutput_LogisticRegressionModel(lrmodel, data);
    }
    else if(modelName.equals("SVMModel")){
      SVMModel svmmodel = (SVMModel) model;     
      if(numClasses==2){
        svmmodel.setThreshold(threshold);
      }
      //Predict
      FeaturesAndPrediction = PredictUnit.predictForOutput_SVMModel(svmmodel, data);
    }
    else if(modelName.equals("NaiveBayesModel")){
      NaiveBayesModel bayesmodel = (NaiveBayesModel) model;    
      //Predict
      FeaturesAndPrediction = PredictUnit.predictForOutput_NaiveBayesModel(bayesmodel, data);
    }
    
    return FeaturesAndPrediction;
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public PredictWithModel(String modelName, String modelPath, String testFile, int numClasses, int minPartition, double threshold, SparkContext sc){
    this.numClasses = numClasses;
    this.threshold = threshold;

    if(modelName.equals("LogisticRegressionModel")){
      LogisticRegressionModel lrmodel = LogisticRegressionModel.load(sc, modelPath);
      this.model = (T)(Object) lrmodel;
    }
    else if(modelName.equals("SVMModel")){
      SVMModel svmmodel = SVMModel.load(sc, modelPath);
      this.model = (T)(Object) svmmodel;
    }
    else if(modelName.equals("NaiveBayesModel")){
      NaiveBayesModel bayesmodel = NaiveBayesModel.load(sc, modelPath);
      this.model = (T)(Object) bayesmodel;
    }
    
    //Load testing data
    LoadProcess loadProcess = new LoadProcess(sc, minPartition);    
    testingData = loadProcess.load(testFile, "Vector");  
    testingData.cache();    
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public OnlineFeatureHandler(FeatureConstraint featureConstraint,
                            DetectionModel detectionModel,
                            onlineMLEventListener onlineMLEventListener,
                            ControllerConnector controllerConnector) {
    this.featureConstraint = featureConstraint;
    this.detectionModel = detectionModel;
    setAthenaMLFeatureConfiguration(detectionModel.getAthenaMLFeatureConfiguration());

    if (detectionModel instanceof KMeansDetectionModel) {
        this.kMeansModel = (KMeansModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof GaussianMixtureDetectionModel) {
        this.gaussianMixtureModel = (GaussianMixtureModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof DecisionTreeDetectionModel) {
        this.decisionTreeModel = (DecisionTreeModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof NaiveBayesDetectionModel) {
        this.naiveBayesModel = (NaiveBayesModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof RandomForestDetectionModel) {
        this.randomForestModel = (RandomForestModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof GradientBoostedTreesDetectionModel) {
        this.gradientBoostedTreesModel = (GradientBoostedTreesModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof SVMDetectionModel) {
        this.svmModel = (SVMModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof LogisticRegressionDetectionModel) {
        this.logisticRegressionModel = (LogisticRegressionModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof LinearRegressionDetectionModel) {
        this.linearRegressionModel = (LinearRegressionModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof LassoDetectionModel) {
        this.lassoModel = (LassoModel) detectionModel.getDetectionModel();
    } else if (detectionModel instanceof RidgeRegressionDetectionModel) {
        this.ridgeRegressionModel = (RidgeRegressionModel) detectionModel.getDetectionModel();
    } else {
        //not supported ML model
        System.out.println("Not supported model");
    }

    this.eventDeliveryManager = new EventDeliveryManagerImpl(controllerConnector, new InternalAthenaFeatureEventListener());
    this.eventDeliveryManager.registerOnlineAthenaFeature(null, new QueryIdentifier(QUERY_IDENTIFIER), featureConstraint);
    this.onlineMLEventListener = onlineMLEventListener;
    System.out.println("Install handler!");
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public static JavaRDD<Tuple2<Object, Object>> predictForMetrics_NaiveBayesModel(NaiveBayesModel model, JavaRDD<LabeledPoint> data){
    JavaRDD<Tuple2<Object, Object>> predictionAndLabels = data.map(
      new Function<LabeledPoint, Tuple2<Object, Object>>() {
        private static final long serialVersionUID = 1L;
        public Tuple2<Object, Object> call(LabeledPoint p) {
          Double prediction = model.predict(p.features());
          return new Tuple2<Object, Object>(prediction, p.label());
        }
      }
    ); 
    return predictionAndLabels;
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public static JavaRDD<Tuple2<Object, Object>> predictForOutput_NaiveBayesModel(NaiveBayesModel model, JavaRDD<LabeledPoint> data){
    JavaRDD<Tuple2<Object, Object>> FeaturesAndPrediction = data.map(
      new Function<LabeledPoint, Tuple2<Object, Object>>() {
        private static final long serialVersionUID = 1L;
        public Tuple2<Object, Object> call(LabeledPoint p) {
          Double prediction = model.predict(p.features());
          return new Tuple2<Object, Object>(p.features(), prediction);
        }
      }
    );
    return FeaturesAndPrediction;    
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public T trainWithBayes(){
  //Train the model
  NaiveBayesModel bayesmodel = NaiveBayes.train(trainingData.rdd(), 1.0);  //This version of the method uses a default smoothing parameter of 1.0.
  this.model = (T)(Object) bayesmodel;

  //Evalute the trained model      
  EvaluateProcess<T> evalProcess = new EvaluateProcess<T>(model, modelName, validData, numClasses);
  evalProcess.evalute(numClasses);
  return model;
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public NaiveBayesModel generateModel(JavaRDD<LabeledPoint> parsedData,
                                     NaiveBayesDetectionAlgorithm naiveBayesDetectionAlgorithm,
                                     NaiveBayesModelSummary naiveBayesModelSummary) {
    NaiveBayesModel naiveBayesModel
            = NaiveBayes.train(parsedData.rdd(), naiveBayesDetectionAlgorithm.getLambda());

    naiveBayesModelSummary.setNaiveBayesDetectionAlgorithm(naiveBayesDetectionAlgorithm);
    return naiveBayesModel;
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public NaiveBayesModel generateModelWithPreprocessing(JavaPairRDD<Object, BSONObject> mongoRDD,
                                                        AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                        NaiveBayesDetectionAlgorithm naiveBayesDetectionAlgorithm,
                                                        Marking marking,
                                                        NaiveBayesModelSummary naiveBayesModelSummary) {

    return generateModel(
            rddPreProcessing(mongoRDD, athenaMLFeatureConfiguration, naiveBayesModelSummary,
                    marking),
            naiveBayesDetectionAlgorithm, naiveBayesModelSummary
    );
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public void setNaiveBayesModel(NaiveBayesModel naiveBayesModel) {
    this.naiveBayesModel = naiveBayesModel;
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public void validate(JavaPairRDD<Object, BSONObject> mongoRDD,
                     AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                     NaiveBayesDetectionModel naiveBayesDetectionModel,
                     NaiveBayesValidationSummary naiveBayesValidationSummary) {
    List<AthenaFeatureField> listOfTargetFeatures = athenaMLFeatureConfiguration.getListOfTargetFeatures();
    Map<AthenaFeatureField, Integer> weight = athenaMLFeatureConfiguration.getWeight();
    Marking marking = naiveBayesDetectionModel.getMarking();
    NaiveBayesModel model = (NaiveBayesModel) naiveBayesDetectionModel.getDetectionModel();
    Normalizer normalizer = new Normalizer();

    int numberOfTargetValue = listOfTargetFeatures.size();

    mongoRDD.foreach(new VoidFunction<Tuple2<Object, BSONObject>>() {
        public void call(Tuple2<Object, BSONObject> t) throws UnknownHostException {
            long start2 = System.nanoTime(); // <-- start
            BSONObject feature = (BSONObject) t._2().get(AthenaFeatureField.FEATURE);
            BSONObject idx = (BSONObject) t._2();
            int originLabel = marking.checkClassificationMarkingElements(idx,feature);

            double[] values = new double[numberOfTargetValue];
            for (int j = 0; j < numberOfTargetValue; j++) {
                values[j] = 0;
                if (feature.containsField(listOfTargetFeatures.get(j).getValue())) {
                    Object obj = feature.get(listOfTargetFeatures.get(j).getValue());
                    if (obj instanceof Long) {
                        values[j] = (Long) obj;
                    } else if (obj instanceof Double) {
                        values[j] = (Double) obj;
                    } else if (obj instanceof Boolean) {
                        values[j] = (Boolean) obj ? 1 : 0;
                    } else {
                        return;
                    }

                    //check weight
                    if (weight.containsKey(listOfTargetFeatures.get(j))) {
                        values[j] *= weight.get(listOfTargetFeatures.get(j));
                    }

                    //check absolute
                    if (athenaMLFeatureConfiguration.isAbsolute()){
                        values[j] = Math.abs(values[j]);
                    }
                }
            }

            Vector normedForVal;
            if (athenaMLFeatureConfiguration.isNormalization()) {
                normedForVal = normalizer.transform(Vectors.dense(values));
            } else {
                normedForVal = Vectors.dense(values);
            }

            LabeledPoint p = new LabeledPoint(originLabel,normedForVal);

            int validatedLabel = (int) model.predict(p.features());


            naiveBayesValidationSummary.updateSummary(validatedLabel,idx,feature);

            long end2 = System.nanoTime();
            long result2 = end2 - start2;
            naiveBayesValidationSummary.addTotalNanoSeconds(result2);
        }
    });
    naiveBayesValidationSummary.getAverageNanoSeconds();
    naiveBayesValidationSummary.setNaiveBayesDetectionAlgorithm((NaiveBayesDetectionAlgorithm) naiveBayesDetectionModel.getDetectionAlgorithm());
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
public NaiveBayesDetectionModel generateNaiveBayesAthenaDetectionModel(JavaSparkContext sc,
                                                                       FeatureConstraint featureConstraint,
                                                                       AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                       DetectionAlgorithm detectionAlgorithm,
                                                                       Indexing indexing,
                                                                       Marking marking) {

    NaiveBayesModelSummary naiveBayesModelSummary = new NaiveBayesModelSummary(
            sc.sc(), indexing, marking);

    long start = System.nanoTime(); // <-- start

    NaiveBayesDetectionAlgorithm naiveBayesDetectionAlgorithm = (NaiveBayesDetectionAlgorithm) detectionAlgorithm;

    NaiveBayesDetectionModel naiveBayesDetectionModel = new NaiveBayesDetectionModel();

    naiveBayesDetectionModel.setNaiveBayesDetectionAlgorithm(naiveBayesDetectionAlgorithm);
    naiveBayesModelSummary.setNaiveBayesDetectionAlgorithm(naiveBayesDetectionAlgorithm);
    naiveBayesDetectionModel.setFeatureConstraint(featureConstraint);
    naiveBayesDetectionModel.setAthenaMLFeatureConfiguration(athenaMLFeatureConfiguration);
    naiveBayesDetectionModel.setIndexing(indexing);
    naiveBayesDetectionModel.setMarking(marking);

    JavaPairRDD<Object, BSONObject> mongoRDD;
    mongoRDD = sc.newAPIHadoopRDD(
            mongodbConfig,            // Configuration
            MongoInputFormat.class,   // InputFormat: read from a live cluster.
            Object.class,             // Key class
            BSONObject.class          // Value class
    );

    NaiveBayesDistJob naiveBayesDistJob = new NaiveBayesDistJob();

    NaiveBayesModel naiveBayesModel = naiveBayesDistJob.generateModelWithPreprocessing(mongoRDD,
            athenaMLFeatureConfiguration, naiveBayesDetectionAlgorithm, marking, naiveBayesModelSummary);

    naiveBayesDetectionModel.setNaiveBayesModel(naiveBayesModel);
    long end = System.nanoTime(); // <-- start
    long time = end - start;
    naiveBayesModelSummary.setTotalLearningTime(time);
    naiveBayesDetectionModel.setClassificationModelSummary(naiveBayesModelSummary);

    return naiveBayesDetectionModel;
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
/**
 * This method builds a naive bayes model
 *
 * @param sparkContext JavaSparkContext initialized with the application
 * @param modelID Model ID
 * @param trainingData Training data as a JavaRDD of LabeledPoints
 * @param testingData Testing data as a JavaRDD of LabeledPoints
 * @param workflow Machine learning workflow
 * @param mlModel Deployable machine learning model
 * @throws MLModelBuilderException
 */
private ModelSummary buildNaiveBayesModel(JavaSparkContext sparkContext, long modelID,
                                          JavaRDD<LabeledPoint> trainingData, JavaRDD<LabeledPoint> testingData, Workflow workflow, MLModel mlModel,
                                          SortedMap<Integer, String> includedFeatures) throws MLModelBuilderException {
    try {
        Map<String, String> hyperParameters = workflow.getHyperParameters();
        NaiveBayesClassifier naiveBayesClassifier = new NaiveBayesClassifier();
        NaiveBayesModel naiveBayesModel = naiveBayesClassifier.train(trainingData,
                Double.parseDouble(hyperParameters.get(MLConstants.LAMBDA)));

        // remove from cache
        trainingData.unpersist();
        // add test data to cache
        testingData.cache();

        JavaPairRDD<Double, Double> predictionsAndLabels = naiveBayesClassifier.test(naiveBayesModel, testingData)
                .cache();
        ClassClassificationAndRegressionModelSummary classClassificationAndRegressionModelSummary = SparkModelUtils
                .getClassClassificationModelSummary(sparkContext, testingData, predictionsAndLabels);

        // remove from cache
        testingData.unpersist();

        mlModel.setModel(new MLClassificationModel(naiveBayesModel));

        classClassificationAndRegressionModelSummary.setFeatures(includedFeatures.values().toArray(new String[0]));
        classClassificationAndRegressionModelSummary.setAlgorithm(SUPERVISED_ALGORITHM.NAIVE_BAYES.toString());

        MulticlassMetrics multiclassMetrics = getMulticlassMetrics(sparkContext, predictionsAndLabels);

        predictionsAndLabels.unpersist();

        classClassificationAndRegressionModelSummary.setMulticlassConfusionMatrix(getMulticlassConfusionMatrix(
                multiclassMetrics, mlModel));
        Double modelAccuracy = getModelAccuracy(multiclassMetrics);
        classClassificationAndRegressionModelSummary.setModelAccuracy(modelAccuracy);
        classClassificationAndRegressionModelSummary.setDatasetVersion(workflow.getDatasetVersion());

        return classClassificationAndRegressionModelSummary;
    } catch (Exception e) {
        throw new MLModelBuilderException("An error occurred while building naive bayes model: " + e.getMessage(),
                e);
    }
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
@Override
protected NaiveBayesModel loadModel(SparkContext sc, String modelPath) {
	return NaiveBayesModel.load(sc, modelPath);
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
/**
 * This method trains a naive bayes model
 *
 * @param trainingData Training dataset as a JavaRDD of labeled points
 * @param lambda Lambda parameter
 * @return Naive bayes model
 */
public NaiveBayesModel train(JavaRDD<LabeledPoint> trainingData, double lambda) {
    return NaiveBayes.train(trainingData.rdd(), lambda);
}
 

import org.apache.spark.mllib.classification.NaiveBayesModel; //导入依赖的package包/类
/**
 * Instantiates a new naive bayes classification model.
 *
 * @param modelId the model id
 * @param model   the model
 */
public NaiveBayesClassificationModel(String modelId, NaiveBayesModel model) {
  super(modelId, model);
}