Java编写的Spark ALS协同过滤推荐算法的源代码能共享一下

import org.apache.spark.SparkConf;
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.mllib.recommendation.ALS;
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
import org.apache.spark.mllib.recommendation.Rating;
import scala.Tuple2;
public class myAls {
  public static void main(String[] args) {
    // TODO Auto-generated method stub
SparkConf conf=new SparkConf().setAppName("als").setMaster("local");
JavaSparkContext sc=new JavaSparkContext(conf);
JavaRDD<String> con=sc.textFile("file:///home/gyq/下载/spark-2.3.2-bin-hadoop2.7/data/mllib/als/sample.data");
JavaRDD<Rating> ratings=con.map(f->{
  return new Rating(
      new Integer(f.split("::")[0]),
      new Integer(f.split("::")[1]),
      new Double(f.split("::")[2]));
});//数据转换为javardd三元组
JavaRDD<Rating>[] rr=ratings.randomSplit(new double[]{0.3,0.7});
MatrixFactorizationModel model=ALS.train(rr[0].rdd(), 15, 10);//生成模型
JavaRDD<Tuple2<Object,Object>> up =rr[1].map(f->{//取0.7的数据要预测的用户和产品ID
  return new Tuple2<>(f.user(),f.product());
});
JavaPairRDD<String,Double> pupr= model.predict(up.rdd()).toJavaRDD().mapToPair(f->{
  return new Tuple2<String,Double>(f.user()+","+f.product(),f.rating());//用0.7的数据去预测得到一个键值对
});
//pupr.foreach(f->System.out.println(f+"gg"));
JavaPairRDD<String,Double> upr=rr[1].mapToPair(f->{//原来0.7的数据转化为键值对
  return new Tuple2<String,Double>(f.user()+","+f.product(),f.rating());
});
//upr.foreach(f->System.out.println(f+"ss"));
JavaPairRDD<String,Tuple2<Double,Double>> mm=upr.join(pupr);
//输出格式为（（用户，项目），（预测评分，实际评分）） 
//mm.foreach(f->System.out.println(f+"qq"));
//model.save(sc.sc(),"file:///home/gyq/下载/spark-2.3.2-bin-hadoop2.7/data/mllib/als/myals");
JavaPairRDD<String,Tuple2<Double,Double>> recom=mm.filter(f->{
  double f2_2=f._2._2;
  double f2_1=f._2._1;
  int a=(int)f2_1;
  int b=(int)f2_2;
  if (a==b){
      return true;}
  else
    return false;
});
double count=recom.count();
double counts=mm.count();
double accur=count/counts;
System.out.println("count="+count);
System.out.println("counts="+counts);
System.out.println("accur="+accur);
sc.stop();
  }
}