src/main/java/it/uniroma2/dicii/sabd/covidproject/Query3.java
/*
     * Apply a MLlib implementation of k-means. For the details of this implementation, see Apache Spark documentation
     * */
Apply a MLlib implementation of k-means. For the details of this implementation, see Apache Spark documentation
Apply a MLlib implementation of k - means . For the details of this implementation see Apache Spark documentation
private static void kMeansMLlib(JavaSparkContext sc, JavaPairRDD<Double,RegionData> trendRegionPairs, String outputDirectory) {
        SparkSession ss = SparkSession.builder().config(sc.getConf()).getOrCreate();
        /* Convert input RDD to DataFrame */
        Dataset<Row> trendRegionsDF = ss.createDataFrame(trendRegionPairs.map(x -> x._2), RegionData.class);
        /* Set feature column in dataset */
        String[] featureCols = {"trendLineCoefficient"};
        VectorAssembler assembler = new VectorAssembler().setInputCols(featureCols).setOutputCol("features");
        Dataset<Row> featuredDF = assembler.transform(trendRegionsDF);
        /* Apply K-Means */
        KMeans kMeans = new KMeans().setK(NUM_OF_K_MEANS_CLUSTERS).setSeed(K_MEANS_MLLIB_SEED);
        KMeansModel model = kMeans.fit(featuredDF);
        Dataset<Row> clusteredRegions = model.transform(featuredDF);
        Dataset<Row> formattedDF = clusteredRegions.withColumn("newTrendLineCoefficient",
                new Column("trendLineCoefficient").cast(DataTypes.createDecimalType(12, 6))).drop("trendLineCoefficient")
                .withColumnRenamed("newTrendLineCoefficient", "trendLineCoefficient").toDF();
        /* Save clustering results in a CSV file */
        formattedDF.select("month", "name", "prediction", "latitude", "longitude", "trendLineCoefficient")
                .write().format("csv").option("header", "false").save(outputDirectory);
    }
private static void kMeansMLlib ( JavaSparkContext sc , JavaPairRDD < Double , RegionData > trendRegionPairs , String outputDirectory ) { SparkSession ss = SparkSession . builder ( ) . config ( sc . getConf ( ) ) . getOrCreate ( ) ; /* Convert input RDD to DataFrame */ Dataset < Row > trendRegionsDF = ss . createDataFrame ( trendRegionPairs . map ( x -> x . _2 ) , RegionData . class ) ; /* Set feature column in dataset */ String [ ] featureCols = { "trendLineCoefficient" } ; VectorAssembler assembler = new VectorAssembler ( ) . setInputCols ( featureCols ) . setOutputCol ( "features" ) ; Dataset < Row > featuredDF = assembler . transform ( trendRegionsDF ) ; /* Apply K-Means */ KMeans kMeans = new KMeans ( ) . setK ( NUM_OF_K_MEANS_CLUSTERS ) . setSeed ( K_MEANS_MLLIB_SEED ) ; KMeansModel model = kMeans . fit ( featuredDF ) ; Dataset < Row > clusteredRegions = model . transform ( featuredDF ) ; Dataset < Row > formattedDF = clusteredRegions . withColumn ( "newTrendLineCoefficient" , new Column ( "trendLineCoefficient" ) . cast ( DataTypes . createDecimalType ( 12 , 6 ) ) ) . drop ( "trendLineCoefficient" ) . withColumnRenamed ( "newTrendLineCoefficient" , "trendLineCoefficient" ) . toDF ( ) ; /* Save clustering results in a CSV file */ formattedDF . select ( "month" , "name" , "prediction" , "latitude" , "longitude" , "trendLineCoefficient" ) . write ( ) . format ( "csv" ) . option ( "header" , "false" ) . save ( outputDirectory ) ; }
Apply a MLlib implementation of k-means.
Apply a MLlib implementation of k - means .