package ML.collaborativeFilltering;

 import org.apache.spark.SparkConf;

 import org.apache.spark.api.java.JavaDoubleRDD;

 import org.apache.spark.api.java.JavaPairRDD;

 import org.apache.spark.api.java.JavaRDD;

 import org.apache.spark.api.java.JavaSparkContext;

 import org.apache.spark.api.java.function.Function;

 import org.apache.spark.mllib.recommendation.ALS;

 import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;

 import org.apache.spark.mllib.recommendation.Rating;

 import scala.Tuple2;

 /**

  * TODO

  *

  * @ClassName: example

  * @author: DingH

  * @since: 2019/4/10 16:03

  */

 public class example {

     public static void main(String[] args) {

         SparkConf conf = new SparkConf().setAppName("Java Collaborative Filtering Example");

         JavaSparkContext jsc = new JavaSparkContext(conf);

         // Load and parse the data

         String path = "D:\\IdeaProjects\\SimpleApp\\src\\main\\resources\\data\\mllib\\als\\test.data";

         JavaRDD<String> data = jsc.textFile(path);

         JavaRDD<Rating> ratings = data.map(new Function<String, Rating>() {

             public Rating call(String s) {

                 String[] sarray = s.split(",");

                 return new Rating(Integer.parseInt(sarray[0]), Integer.parseInt(sarray[1]), Double.parseDouble(sarray[2]));

             }

           }

         );

         int ranks = 10;

         int numIterations = 10;

         MatrixFactorizationModel model = ALS.train(ratings.rdd(), ranks, numIterations);

         JavaRDD<Tuple2<Object, Object>> userProducts = ratings.map(new Function<Rating, Tuple2<Object, Object>>() {

             public Tuple2<Object, Object> call(Rating r) {

               return new Tuple2<Object, Object>(r.user(), r.product());

             }

           }

         );

         JavaPairRDD<Tuple2<Integer, Integer>, Double> predictions = JavaPairRDD.fromJavaRDD(model.predict(JavaRDD.toRDD(userProducts)).toJavaRDD().map(

             new Function<Rating, Tuple2<Tuple2<Integer, Integer>, Double>>() {

               public Tuple2<Tuple2<Integer, Integer>, Double> call(Rating r){

                 return new Tuple2<Tuple2<Integer, Integer>, Double>(

                   new Tuple2<Integer, Integer>(r.user(), r.product()), r.rating());

               }

             }

           ));

         JavaRDD<Tuple2<Double, Double>> ratesAndPreds = JavaPairRDD.fromJavaRDD(ratings.map(

             new Function<Rating, Tuple2<Tuple2<Integer, Integer>, Double>>() {

               public Tuple2<Tuple2<Integer, Integer>, Double> call(Rating r){

                 return new Tuple2<Tuple2<Integer, Integer>, Double>(

                   new Tuple2<Integer, Integer>(r.user(), r.product()), r.rating());

               }

             }

           )).join(predictions).values();

         double MSE = JavaDoubleRDD.fromRDD(ratesAndPreds.map(

           new Function<Tuple2<Double, Double>, Object>() {

             public Object call(Tuple2<Double, Double> pair) {

               Double err = pair._1() - pair._2();

               return err * err;

             }

           }

         ).rdd()).mean();

         System.out.println("Mean Squared Error = " + MSE);

     }

 }