import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public LassoValidationSummary validateLassoAthenaFeatures(JavaSparkContext sc,
                                                          FeatureConstraint featureConstraint,
                                                          AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                          LassoDetectionModel lassoDetectionModel,
                                                          Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    LassoDetectionAlgorithm lassoDetectionAlgorithm =
            (LassoDetectionAlgorithm) lassoDetectionModel.getDetectionAlgorithm();

    LassoValidationSummary lassoValidationSummary = new LassoValidationSummary();
    lassoValidationSummary.setLassoDetectionAlgorithm(lassoDetectionAlgorithm);
    LassoDistJob lassoDistJob = new LassoDistJob();

    lassoDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            lassoDetectionModel,
            lassoValidationSummary);


    long end = System.nanoTime(); // <-- start
    long time = end - start;
    lassoValidationSummary.setValidationTime(time);

    return lassoValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public RidgeRegressionValidationSummary validateRidgeRegressionAthenaFeatures(JavaSparkContext sc,
                                                                              FeatureConstraint featureConstraint,
                                                                              AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                              RidgeRegressionDetectionModel ridgeRegressionDetectionModel,
                                                                              Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    RidgeRegressionDetectionAlgorithm ridgeRegressionDetectionAlgorithm =
            (RidgeRegressionDetectionAlgorithm) ridgeRegressionDetectionModel.getDetectionAlgorithm();

    RidgeRegressionValidationSummary ridgeRegressionValidationSummary = new RidgeRegressionValidationSummary();
    ridgeRegressionValidationSummary.setRidgeRegressionDetectionAlgorithm(ridgeRegressionDetectionAlgorithm);
    RidgeRegressionDistJob ridgeRegressionDistJob = new RidgeRegressionDistJob();

    ridgeRegressionDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            ridgeRegressionDetectionModel,
            ridgeRegressionValidationSummary);


    long end = System.nanoTime(); // <-- start
    long time = end - start;
    ridgeRegressionValidationSummary.setValidationTime(time);
    return ridgeRegressionValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public LinearRegressionValidationSummary validateLinearRegressionAthenaFeatures(JavaSparkContext sc,
                                                                                FeatureConstraint featureConstraint,
                                                                                AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                                LinearRegressionDetectionModel linearRegressionDetectionModel,
                                                                                Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    LinearRegressionDetectionAlgorithm linearRegressionDetectionAlgorithm =
            (LinearRegressionDetectionAlgorithm) linearRegressionDetectionModel.getDetectionAlgorithm();

    LinearRegressionValidationSummary linearRegressionValidationSummary =
            new LinearRegressionValidationSummary();
    linearRegressionValidationSummary.setLinearRegressionDetectionAlgorithm(linearRegressionDetectionAlgorithm);

    LinearRegressionDistJob linearRegressionDistJob = new LinearRegressionDistJob();

    linearRegressionDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            linearRegressionDetectionModel,
            linearRegressionValidationSummary);


    long end = System.nanoTime(); // <-- start
    long time = end - start;
    linearRegressionValidationSummary.setValidationTime(time);


    return linearRegressionValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public LogisticRegressionValidationSummary validateLogisticRegressionAthenaFeatures(JavaSparkContext sc,
                                                                                    FeatureConstraint featureConstraint,
                                                                                    AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                                    LogisticRegressionDetectionModel logisticRegressionDetectionModel,
                                                                                    Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    LogisticRegressionDetectionAlgorithm logisticRegressionDetectionAlgorithm =
            (LogisticRegressionDetectionAlgorithm) logisticRegressionDetectionModel.getDetectionAlgorithm();

    LogisticRegressionValidationSummary logisticRegressionValidationSummary =
            new LogisticRegressionValidationSummary(sc.sc(), logisticRegressionDetectionAlgorithm.getNumClasses(), indexing, marking);

    LogisticRegressionDistJob logisticRegressionDistJob = new LogisticRegressionDistJob();

    logisticRegressionDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            logisticRegressionDetectionModel,
            logisticRegressionValidationSummary);


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

    logisticRegressionValidationSummary.setTotalValidationTime(time);
    return logisticRegressionValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public SVMValidationSummary validateSVMAthenaFeatures(JavaSparkContext sc,
                                                      FeatureConstraint featureConstraint,
                                                      AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                      SVMDetectionModel svmDetectionModel,
                                                      Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    SVMDetectionAlgorithm svmDetectionAlgorithm =
            (SVMDetectionAlgorithm) svmDetectionModel.getDetectionAlgorithm();

    SVMValidationSummary svmValidationSummary =
            new SVMValidationSummary(sc.sc(),
                    2, indexing, marking);

    SVMDistJob svmDistJob = new SVMDistJob();

    svmDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            svmDetectionModel,
            svmValidationSummary);


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

    svmValidationSummary.setTotalValidationTime(time);
    return svmValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public GradientBoostedTreesValidationSummary validateGradientBoostedTreesAthenaFeatures(JavaSparkContext sc,
                                                                                        FeatureConstraint featureConstraint,
                                                                                        AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                                        GradientBoostedTreesDetectionModel gradientBoostedTreesDetectionModel,
                                                                                        Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    GradientBoostedTreesDetectionAlgorithm gradientBoostedTreesDetectionAlgorithm =
            (GradientBoostedTreesDetectionAlgorithm) gradientBoostedTreesDetectionModel.getDetectionAlgorithm();

    GradientBoostedTreesValidationSummary gradientBoostedTreesValidationSummary =
            new GradientBoostedTreesValidationSummary(sc.sc(),
                    gradientBoostedTreesDetectionAlgorithm.getNumClasses(), indexing, marking);

    GradientBoostedTreesDistJob gradientBoostedTreesDistJob = new GradientBoostedTreesDistJob();

    gradientBoostedTreesDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            gradientBoostedTreesDetectionModel,
            gradientBoostedTreesValidationSummary);


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

    gradientBoostedTreesValidationSummary.setTotalValidationTime(time);
    return gradientBoostedTreesValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public RandomForestValidationSummary validateRandomForestAthenaFeatures(JavaSparkContext sc,
                                                                        FeatureConstraint featureConstraint,
                                                                        AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                        RandomForestDetectionModel randomForestDetectionModel,
                                                                        Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    RandomForestDetectionAlgorithm randomForestDetectionAlgorithm = (RandomForestDetectionAlgorithm) randomForestDetectionModel.getDetectionAlgorithm();

    RandomForestValidationSummary randomForestValidationSummary =
            new RandomForestValidationSummary(sc.sc(), randomForestDetectionAlgorithm.getNumClasses(), indexing, marking);

    RandomForestDistJob randomForestDistJob = new RandomForestDistJob();

    randomForestDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            randomForestDetectionModel,
            randomForestValidationSummary);


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

    randomForestValidationSummary.setTotalValidationTime(time);
    return randomForestValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public NaiveBayesValidationSummary validateNaiveBayesAthenaFeatures(JavaSparkContext sc,
                                                                    FeatureConstraint featureConstraint,
                                                                    AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                    NaiveBayesDetectionModel naiveBayesDetectionModel,
                                                                    Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    NaiveBayesDetectionAlgorithm naiveBayesDetectionAlgorithm = (NaiveBayesDetectionAlgorithm) naiveBayesDetectionModel.getDetectionAlgorithm();

    NaiveBayesValidationSummary naiveBayesValidationSummary =
            new NaiveBayesValidationSummary(sc.sc(), naiveBayesDetectionAlgorithm.getNumClasses(), indexing, marking);

    NaiveBayesDistJob naiveBayesDistJob = new NaiveBayesDistJob();


    naiveBayesDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            naiveBayesDetectionModel,
            naiveBayesValidationSummary);


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

    naiveBayesValidationSummary.setTotalValidationTime(time);
    return naiveBayesValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public DecisionTreeValidationSummary validateDecisionTreeAthenaFeatures(JavaSparkContext sc,
                                                                        FeatureConstraint featureConstraint,
                                                                        AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                        DecisionTreeDetectionModel decisionTreeDetectionModel,
                                                                        Indexing indexing, Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    DecisionTreeDetectionAlgorithm decisionTreeDetectionAlgorithm = (DecisionTreeDetectionAlgorithm) decisionTreeDetectionModel.getDetectionAlgorithm();

    DecisionTreeValidationSummary decisionTreeValidationSummary =
            new DecisionTreeValidationSummary(sc.sc(), decisionTreeDetectionAlgorithm.getNumClasses(), indexing, marking);

    DecisionTreeDistJob decisionTreeDistJob = new DecisionTreeDistJob();

    decisionTreeDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            decisionTreeDetectionModel,
            decisionTreeValidationSummary);


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

    decisionTreeValidationSummary.setTotalValidationTime(time);
    return decisionTreeValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public GaussianMixtureValidationSummary validateGaussianMixtureAthenaFeatures(JavaSparkContext sc,
                                                                              FeatureConstraint featureConstraint,
                                                                              AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                              GaussianMixtureDetectionModel gaussianMixtureDetectionModel,
                                                                              Indexing indexing,
                                                                              Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    GaussianMixtureDetectionAlgorithm gaussianMixtureDetectionAlgorithm = (GaussianMixtureDetectionAlgorithm) gaussianMixtureDetectionModel.getDetectionAlgorithm();
    GaussianMixtureValidationSummary gaussianMixtureValidationSummary =
            new GaussianMixtureValidationSummary(sc.sc(), gaussianMixtureDetectionAlgorithm.getK(), indexing, marking);


    GaussianMixtureDistJob gaussianMixtureDistJob = new GaussianMixtureDistJob();

    gaussianMixtureDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            gaussianMixtureDetectionModel,
            gaussianMixtureValidationSummary);


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

    gaussianMixtureValidationSummary.setTotalValidationTime(time);
    return gaussianMixtureValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public KmeansValidationSummary validateKMeansAthenaFeatures(JavaSparkContext sc,
                                                            FeatureConstraint featureConstraint,
                                                            AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                            KMeansDetectionModel kMeansDetectionModel,
                                                            Indexing indexing,
                                                            Marking marking) {
    long start = System.nanoTime(); // <-- start

    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
    );

    KMeansDetectionAlgorithm kMeansDetectionAlgorithm = (KMeansDetectionAlgorithm) kMeansDetectionModel.getDetectionAlgorithm();

    KmeansValidationSummary kmeansValidationSummary =
            new KmeansValidationSummary(sc.sc(), kMeansDetectionAlgorithm.getK(), indexing, marking);

    KMeansDistJob KMeansDistJob = new KMeansDistJob();

    KMeansDistJob.validate(mongoRDD,
            athenaMLFeatureConfiguration,
            kMeansDetectionModel,
            kmeansValidationSummary);
    long end = System.nanoTime(); // <-- start
    long time = end - start;
    kmeansValidationSummary.setTotalValidationTime(time);
    return kmeansValidationSummary;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
/**
 * Uses the Hadoop API to get the input objects. Then uses the geneesc framewes to do tes evaluesion.es
 *
 * @param transformation
 * @return
 */
@Override
public JavaRDD<ObjectValue> evaluate(Transformation transformation) {
    JavaSparkContext sc = NotaQL.SparkFactory.getSparkContext();

    String mongoDBHost = "mongodb://" + NotaQL.prop.getProperty("mongodb_host", "localhost") + ":27017/";

    Configuration config = new Configuration();
    config.set("mongo.input.uri", mongoDBHost + databaseName + "." + collectionName);

    // add partial filter to query in mongodb
    if(!noQuery && transformation.getInPredicate() != null) {
        final BSONObject query = FilterTranslator.toMongoDBQuery(transformation.getInPredicate());
        logger.info("Sending query to MongoDB: " + query.toString());
        config.set("mongo.input.query", query.toString());
    }

    final SparkTransformationEvaluator evaluator = new SparkTransformationEvaluator(transformation);

    JavaPairRDD<Object, BSONObject> mongoRDD = sc.newAPIHadoopRDD(config, MongoInputFormat.class, Object.class, BSONObject.class);

    // convert all objects in rdd to inner format
    final JavaRDD<Value> converted = mongoRDD.map(t -> ValueConverter.convertToNotaQL(t._2));
    // filter the ones not fulfilling the input filter (queries of MongoDB are less expressive than NotaQL)
    final JavaRDD<Value> filtered = converted.filter(v -> transformation.satisfiesInPredicate((ObjectValue) v));

    // process all input
    return evaluator.process(filtered);
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
private void setupJob() {
  _job.setInputFormatClass(MongoInputFormat.class);
  _job.setMapperClass(BulkImportMapper.class);
  _job.setMapOutputKeyClass(ImmutableBytesWritable.class);
  _job.setMapOutputValueClass(Put.class);

  MongoConfigUtil.setInputURI(getConfiguration(), _mongoURI);
  MongoConfigUtil.setReadSplitsFromSecondary(getConfiguration(), true);
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public LassoDetectionModel generateLassoAthenaDetectionModel(JavaSparkContext sc,
                                                             FeatureConstraint featureConstraint,
                                                             AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                             DetectionAlgorithm detectionAlgorithm,
                                                             Indexing indexing,
                                                             Marking marking) {
    LassoModelSummary lassoModelSummary = new LassoModelSummary(
            sc.sc(), indexing, marking);

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

    LassoDetectionAlgorithm lassoDetectionAlgorithm = (LassoDetectionAlgorithm) detectionAlgorithm;

    LassoDetectionModel lassoDetectionModel = new LassoDetectionModel();

    lassoDetectionModel.setLassoDetectionAlgorithm(lassoDetectionAlgorithm);
    lassoModelSummary.setLassoDetectionAlgorithm(lassoDetectionAlgorithm);
    lassoDetectionModel.setFeatureConstraint(featureConstraint);
    lassoDetectionModel.setAthenaMLFeatureConfiguration(athenaMLFeatureConfiguration);
    lassoDetectionModel.setIndexing(indexing);
    lassoDetectionModel.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
    );

    LassoDistJob lassoDistJob = new LassoDistJob();

    LassoModel lassoModel = lassoDistJob.generateDecisionTreeWithPreprocessing(mongoRDD,
            athenaMLFeatureConfiguration, lassoDetectionAlgorithm, marking, lassoModelSummary);


    lassoDetectionModel.setModel(lassoModel);
    long end = System.nanoTime(); // <-- start
    long time = end - start;
    lassoModelSummary.setTotalLearningTime(time);
    lassoDetectionModel.setClassificationModelSummary(lassoModelSummary);

    return lassoDetectionModel;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public RidgeRegressionDetectionModel generateRidgeRegressionAthenaDetectionModel(JavaSparkContext sc,
                                                                                 FeatureConstraint featureConstraint,
                                                                                 AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                                 DetectionAlgorithm detectionAlgorithm,
                                                                                 Indexing indexing,
                                                                                 Marking marking) {
    RidgeRegressionModelSummary ridgeRegressionModelSummary = new RidgeRegressionModelSummary(
            sc.sc(), indexing, marking);

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

    RidgeRegressionDetectionAlgorithm ridgeRegressionDetectionAlgorithm = (RidgeRegressionDetectionAlgorithm) detectionAlgorithm;

    RidgeRegressionDetectionModel ridgeRegressionDetectionModel = new RidgeRegressionDetectionModel();

    ridgeRegressionDetectionModel.setRidgeRegressionDetectionAlgorithm(ridgeRegressionDetectionAlgorithm);
    ridgeRegressionModelSummary.setRidgeRegressionDetectionAlgorithm(ridgeRegressionDetectionAlgorithm);
    ridgeRegressionDetectionModel.setFeatureConstraint(featureConstraint);
    ridgeRegressionDetectionModel.setAthenaMLFeatureConfiguration(athenaMLFeatureConfiguration);
    ridgeRegressionDetectionModel.setIndexing(indexing);
    ridgeRegressionDetectionModel.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
    );

    RidgeRegressionDistJob ridgeRegressionDistJob = new RidgeRegressionDistJob();

    RidgeRegressionModel ridgeRegressionModel = ridgeRegressionDistJob.generateDecisionTreeWithPreprocessing(mongoRDD,
            athenaMLFeatureConfiguration, ridgeRegressionDetectionAlgorithm, marking, ridgeRegressionModelSummary);


    ridgeRegressionDetectionModel.setModel(ridgeRegressionModel);
    long end = System.nanoTime(); // <-- start
    long time = end - start;
    ridgeRegressionModelSummary.setTotalLearningTime(time);
    ridgeRegressionDetectionModel.setClassificationModelSummary(ridgeRegressionModelSummary);

    return ridgeRegressionDetectionModel;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public LinearRegressionDetectionModel generateLinearRegressionAthenaDetectionModel(JavaSparkContext sc,
                                                                                   FeatureConstraint featureConstraint,
                                                                                   AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                                   DetectionAlgorithm detectionAlgorithm,
                                                                                   Indexing indexing,
                                                                                   Marking marking) {
    LinearRegressionModelSummary linearRegressionModelSummary = new LinearRegressionModelSummary(
            sc.sc(), indexing, marking);

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

    LinearRegressionDetectionAlgorithm linearRegressionDetectionAlgorithm = (LinearRegressionDetectionAlgorithm) detectionAlgorithm;

    LinearRegressionDetectionModel linearRegressionDetectionModel = new LinearRegressionDetectionModel();

    linearRegressionDetectionModel.setLinearRegressionDetectionAlgorithm(linearRegressionDetectionAlgorithm);
    linearRegressionModelSummary.setLinearRegressionDetectionAlgorithm(linearRegressionDetectionAlgorithm);
    linearRegressionDetectionModel.setFeatureConstraint(featureConstraint);
    linearRegressionDetectionModel.setAthenaMLFeatureConfiguration(athenaMLFeatureConfiguration);
    linearRegressionDetectionModel.setIndexing(indexing);
    linearRegressionDetectionModel.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
    );

    LinearRegressionDistJob linearRegressionDistJob = new LinearRegressionDistJob();

    LinearRegressionModel linearRegressionModel = linearRegressionDistJob.generateDecisionTreeWithPreprocessing(mongoRDD,
            athenaMLFeatureConfiguration, linearRegressionDetectionAlgorithm, marking, linearRegressionModelSummary);


    linearRegressionDetectionModel.setModel(linearRegressionModel);
    long end = System.nanoTime(); // <-- start
    long time = end - start;
    linearRegressionModelSummary.setTotalLearningTime(time);
    linearRegressionDetectionModel.setClassificationModelSummary(linearRegressionModelSummary);

    return linearRegressionDetectionModel;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public LogisticRegressionDetectionModel generateLogisticRegressionAthenaDetectionModel(JavaSparkContext sc,
                                                                                       FeatureConstraint featureConstraint,
                                                                                       AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                                       DetectionAlgorithm detectionAlgorithm,
                                                                                       Indexing indexing,
                                                                                       Marking marking) {
    LogisticRegressionModelSummary logisticRegressionModelSummary = new LogisticRegressionModelSummary(
            sc.sc(), indexing, marking);

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

    LogisticRegressionDetectionAlgorithm logisticRegressionDetectionAlgorithm = (LogisticRegressionDetectionAlgorithm) detectionAlgorithm;

    LogisticRegressionDetectionModel logisticRegressionDetectionModel = new LogisticRegressionDetectionModel();

    logisticRegressionDetectionModel.setLogisticRegressionDetectionAlgorithm(logisticRegressionDetectionAlgorithm);
    logisticRegressionModelSummary.setLogisticRegressionDetectionAlgorithm(logisticRegressionDetectionAlgorithm);
    logisticRegressionDetectionModel.setFeatureConstraint(featureConstraint);
    logisticRegressionDetectionModel.setAthenaMLFeatureConfiguration(athenaMLFeatureConfiguration);
    logisticRegressionDetectionModel.setIndexing(indexing);
    logisticRegressionDetectionModel.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
    );

    LogisticRegressionDistJob logisticRegressionDistJob = new LogisticRegressionDistJob();

    LogisticRegressionModel logisticRegressionModel = logisticRegressionDistJob.generateDecisionTreeWithPreprocessing(mongoRDD,
            athenaMLFeatureConfiguration, logisticRegressionDetectionAlgorithm, marking, logisticRegressionModelSummary);


    logisticRegressionDetectionModel.setModel(logisticRegressionModel);
    long end = System.nanoTime(); // <-- start
    long time = end - start;
    logisticRegressionModelSummary.setTotalLearningTime(time);
    logisticRegressionDetectionModel.setClassificationModelSummary(logisticRegressionModelSummary);

    return logisticRegressionDetectionModel;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public SVMDetectionModel generateSVMAthenaDetectionModel(JavaSparkContext sc,
                                                         FeatureConstraint featureConstraint,
                                                         AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                         DetectionAlgorithm detectionAlgorithm,
                                                         Indexing indexing,
                                                         Marking marking) {
    SVMModelSummary svmModelSummary = new SVMModelSummary(
            sc.sc(), indexing, marking);

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

    SVMDetectionAlgorithm svmDetectionAlgorithm = (SVMDetectionAlgorithm) detectionAlgorithm;

    SVMDetectionModel svmDetectionModel = new SVMDetectionModel();

    svmDetectionModel.setSVMDetectionAlgorithm(svmDetectionAlgorithm);
    svmModelSummary.setSVMDetectionAlgorithm(svmDetectionAlgorithm);
    svmDetectionModel.setFeatureConstraint(featureConstraint);
    svmDetectionModel.setAthenaMLFeatureConfiguration(athenaMLFeatureConfiguration);
    svmDetectionModel.setIndexing(indexing);
    svmDetectionModel.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
    );

    SVMDistJob svmDistJob = new SVMDistJob();

    SVMModel svmModel = svmDistJob.generateDecisionTreeWithPreprocessing(mongoRDD,
            athenaMLFeatureConfiguration, svmDetectionAlgorithm, marking, svmModelSummary);


    svmDetectionModel.setSVMModel(svmModel);
    long end = System.nanoTime(); // <-- start
    long time = end - start;
    svmModelSummary.setTotalLearningTime(time);
    svmDetectionModel.setClassificationModelSummary(svmModelSummary);

    return svmDetectionModel;
}
 

import com.mongodb.hadoop.MongoInputFormat; //导入依赖的package包/类
public GradientBoostedTreesDetectionModel generateGradientBoostedTreesAthenaDetectionModel(JavaSparkContext sc,
                                                                                           FeatureConstraint featureConstraint,
                                                                                           AthenaMLFeatureConfiguration athenaMLFeatureConfiguration,
                                                                                           DetectionAlgorithm detectionAlgorithm,
                                                                                           Indexing indexing,
                                                                                           Marking marking) {
    GradientBoostedTreesModelSummary gradientBoostedTreesModelSummary = new GradientBoostedTreesModelSummary(
            sc.sc(), indexing, marking);

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

    GradientBoostedTreesDetectionAlgorithm gradientBoostedTreesDetectionAlgorithm = (GradientBoostedTreesDetectionAlgorithm) detectionAlgorithm;

    GradientBoostedTreesDetectionModel gradientBoostedTreesDetectionModel = new GradientBoostedTreesDetectionModel();

    gradientBoostedTreesDetectionModel.setGradientBoostedTreesDetectionAlgorithm(gradientBoostedTreesDetectionAlgorithm);
    gradientBoostedTreesModelSummary.setGradientBoostedTreesDetectionAlgorithm(gradientBoostedTreesDetectionAlgorithm);
    gradientBoostedTreesDetectionModel.setFeatureConstraint(featureConstraint);
    gradientBoostedTreesDetectionModel.setAthenaMLFeatureConfiguration(athenaMLFeatureConfiguration);
    gradientBoostedTreesDetectionModel.setIndexing(indexing);
    gradientBoostedTreesDetectionModel.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
    );

    GradientBoostedTreesDistJob gradientBoostedTreesDistJob = new GradientBoostedTreesDistJob();

    GradientBoostedTreesModel decisionTreeModel = gradientBoostedTreesDistJob.generateDecisionTreeWithPreprocessing(mongoRDD,
            athenaMLFeatureConfiguration, gradientBoostedTreesDetectionAlgorithm, marking, gradientBoostedTreesModelSummary);


    gradientBoostedTreesDetectionModel.setGradientBoostedTreestMod