devbox
在线问答5
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

大数据之Spark（七）--- Spark机器学习，朴素贝叶斯，酒水评估和分类案例学习，垃圾邮件过滤学习案例，电商商品推荐，电影推荐学习案例_在分布式系统中使用spark框架中实现朴素贝叶斯-垃圾邮件过滤算法

作者：在线问答5 | 2024-08-05 13:59:54

赞

踩

在分布式系统中使用spark框架中实现朴素贝叶斯-垃圾邮件过滤算法
一、Saprk机器学习介绍
------------------------------------------------------------------
    1.监督学习
        a.有训练数据集,符合规范的数据
        b.根据数据集，产生一个推断函数
        c.根据产生的推断函数，处理新的数据,从而产生预测结果
        d.常见的监督学习：电子邮件的垃圾分类；按照内容标记网页；声音识别等
        e.常见监督学习的算法
            神经网络
            SVM
            贝叶斯分类器等

    2.非监督学习
        a.没有训练数据
        b.分析可用数据，找出数据的模式和趋势，从而将数据聚类，分组
        c.常用的手段
            kmean
            自我组织的map
            层次聚类等

    3.推荐
        协同过滤，基于之前的购买点击和分级行为提供最接近的推荐：猜你喜欢，推荐你认识的人


二、朴素贝叶斯算法
-----------------------------------------------
    1.公式
        P(B|A) = P(A|B) * P(B) / P(A)

    2.解释
        A：事件A
        B：事件B
        P(B|A)：A事件发生时，B事件发生的概率，依赖事件
        P(A|B)：B事件发生时，A事件发生的概率，依赖事件
        P(B)：B事件发生的概率,独立事件
        P(A)：A事件发生的概率,独立事件


三、Spark机器学习库
----------------------------------------------------
    [Estimator]
        评估器：操作在数据框DataFrame上的算法
        运行在包含了feature和label(结果)的dataFrame之上，对数据进行训练创建model模型。
        该模型用于以后的预测。

    [Transformer]
        数据框转换器
        将包含feature的Dataframe变换成了包含了预测的dataframe.
        由Estimator创建的model就是Transformer。

    [Parameter]
        Estimator和Transformer使用的数据，通常和机器学习的算法相关。
        Spark API给出了一致性API针对算法。

    [Pipeline]
        将Estimators和Transformers组合在一起，形成机器学习工作流.


四、酒水评估[线性回归算法]和分类[逻辑回归算法]案例学习
----------------------------------------------------------
    1.数据集下载地址：http://archive.ics.uci.edu/ml/datasets/Wine+Quality

    2.线性回归实现酒水评估 ---scala版
        

  1. import org.apache.spark.ml.classification.LogisticRegression
  2.         import org.apache.spark.ml.param.ParamMap
  3.         import org.apache.spark.ml.linalg.{Vector, Vectors}
  4.         import org.apache.spark.ml.regression.LinearRegressionModel
  5.         import org.apache.spark.sql.{Row, SparkSession}
  6.  
  7.         //线性回归实现酒水等级评估0-10
  8.         object SparkScalaML1 {
  9.  
  10.             def main(args: Array[String]): Unit = {
  11.                 val sess = SparkSession.builder()
  12.                         .appName("ml")
  13.                         .master("local[4]")
  14.                         .getOrCreate();
  15.  
  16.                 val sc = sess.sparkContext;
  17.                 //数据目录
  18.                 val dataDir = "file:///D:/share/spark/ml/winequality-white.csv"
  19.                 //定义样例类
  20.                 case class Wine(FixedAcidity: Double, VolatileAcidity: Double,
  21.                                 CitricAcid: Double, ResidualSugar: Double, Chlorides: Double,
  22.                                 FreeSulfurDioxide: Double, TotalSulfurDioxide: Double, Density: Double, PH:
  23.                                 Double, Sulphates: Double, Alcohol: Double, Quality: Double)
  24.  
  25.                 //变换
  26.                 val wineDataRDD = sc.textFile(dataDir)
  27.                         .map(_.split(";"))
  28.                         .map(
  29.                             w => Wine(w(0).toDouble, w(1).toDouble,
  30.                                 w(2).toDouble, w(3).toDouble, w(4).toDouble, w(5).toDouble, w(6).toDouble, w(7).toDouble, w(8).toDouble
  31.                                 , w(9).toDouble, w(10).toDouble, w(11).toDouble)
  32.                         )
  33.  
  34.                 //转换RDD成DataFrame
  35.                 import sess.implicits._
  36.                 val trainingDF = wineDataRDD.map(w => (w.Quality,
  37.                         Vectors.dense(w.FixedAcidity, w.VolatileAcidity, w.CitricAcid,
  38.                             w.ResidualSugar, w.Chlorides, w.FreeSulfurDioxide, w.TotalSulfurDioxide,
  39.                             w.Density, w.PH, w.Sulphates, w.Alcohol))).toDF("label", "features")
  40.  
  41.                 trainingDF.show()
  42.                 println("=============================================")
  43.  
  44.                 //创建线性回归对象
  45.                 val lr = new LinearRegression()
  46.                 //设置最大迭代次数
  47.                 //lr.setMaxIter(10).setRegParam(0.01)
  48.                 lr.setMaxIter(10)
  49.                 //通过线性回归，拟合训练数据，生成model
  50.                 val model = lr.fit(trainingDF)
  51.  
  52.                 //创建测试Dataframe
  53.         //        val testDF = sess.createDataFrame(Seq((1.0,Vectors.dense(6.1, 0.32, 0.24, 1.5, 0.036, 43, 140, 0.9894, 3.36, 0.64, 10.7)),
  54.         //            (0.0, Vectors.dense(5.2, 0.44, 0.04, 1.4, 0.036, 38, 124, 0.9898, 3.29, 0.42, 12.4)),
  55.         //            (0.0,Vectors.dense(7.2, 0.32, 0.47, 5.1, 0.044, 19, 65, 0.9951, 3.38, 0.36, 9)),
  56.         //            (0.0, Vectors.dense(6.4, 0.595, 0.14, 5.2, 0.058, 15, 97, 0.991, 3.03, 0.41, 12.6)))
  57.         //        ).toDF("label", "features")
  58.  
  59.                 val testDF = sess.createDataFrame(Seq((5.0000, Vectors.dense(7.4,
  60.                     0.7, 0.0, 1.9, 0.076, 25.0, 67.0, 0.9968, 3.2, 0.68,9.8)),(5.00000,
  61.                         Vectors.dense(7.8, 0.88, 0.0, 2.6, 0.098, 11.0, 34.0, 0.9978, 3.51, 0.56,
  62.                             9.4)),(7.00000, Vectors.dense(7.3, 0.65, 0.0, 1.2, 0.065, 15.0, 18.0, 0.9968,
  63.                     3.36, 0.57, 9.5)))).toDF("label", "features")
  64.  
  65.                 //显式测试数据
  66.                 testDF.show();
  67.                 println("========================")
  68.                 //预测测试数据(带标签),评测模型的质量。
  69.                 testDF.createOrReplaceTempView("test")
  70.                 val tested = model.transform(testDF).select("features", "label", "prediction")
  71.                 tested.show();
  72.  
  73.                 println("========================")
  74.                 //预测无标签的测试数据。
  75.                 val predictDF = sess.sql("SELECT features FROM test")
  76.                 //预测结果
  77.                 val predicted = model.transform(predictDF).select("features", "prediction")
  78.                 predicted.show();
  79.             }
  80.         }
    3.Java代码实现 =====》 TODO

    4.Wine案例机器学习步骤分析
        a.读取训练数据，生成训练数据框
        b.创建LinearRegression线性回归对象
        c.通过训练数据拟合出模型，完成评估管线
        d.读取带评级结果的测试数据，生成测试数据框,用于测试[测试数据要包含正确的评级结果，以便于校验模型是否完美预测]
        e.使用模型对测试数据进行变换，产生新的数据框，抽取特征，完成预测评级，输出预测评级结果
        f.使用OK的模型，对生产数据进行变换，完成生产数据的评级

    5.逻辑回归实现酒水分类
        

  1. import org.apache.spark.ml.classification.LogisticRegression
  2.             import org.apache.spark.ml.linalg.Vectors
  3.             import org.apache.spark.ml.regression.LinearRegression
  4.             import org.apache.spark.sql.SparkSession
  5.  
  6.  
  7.             /**
  8.               * 逻辑回归实现酒水分类 --好/坏 0/1
  9.               */
  10.             object SparkML2 {
  11.  
  12.                 def main(args: Array[String]): Unit = {
  13.                     val sess = SparkSession.builder()
  14.                             .appName("ml")
  15.                             .master("local[4]")
  16.                             .getOrCreate();
  17.  
  18.                     val sc = sess.sparkContext;
  19.                     //数据目录
  20.                     val dataDir = "file:///D:/share/spark/ml/winequality-white.csv"
  21.                     //定义样例类
  22.                     case class Wine(FixedAcidity: Double, VolatileAcidity: Double,
  23.                                     CitricAcid: Double, ResidualSugar: Double, Chlorides: Double,
  24.                                     FreeSulfurDioxide: Double, TotalSulfurDioxide: Double, Density: Double, PH:
  25.                                     Double, Sulphates: Double, Alcohol: Double, Quality: Double)
  26.  
  27.                     //变换
  28.                     val wineDataRDD = sc.textFile(dataDir)
  29.                             .map(_.split(";"))
  30.                             .map(
  31.                                 w => Wine(w(0).toDouble, w(1).toDouble,
  32.                                     w(2).toDouble, w(3).toDouble, w(4).toDouble, w(5).toDouble, w(6).toDouble, w(7).toDouble, w(8).toDouble
  33.                                     , w(9).toDouble, w(10).toDouble, w(11).toDouble)
  34.                             )
  35.  
  36.                     //转换RDD成DataFrame
  37.                     import sess.implicits._
  38.                     val trainingDF = wineDataRDD.map(w => (if(w.Quality < 7) 0D else
  39.                         1D, Vectors.dense(w.FixedAcidity, w.VolatileAcidity, w.CitricAcid,
  40.                         w.ResidualSugar, w.Chlorides, w.FreeSulfurDioxide, w.TotalSulfurDioxide,
  41.                         w.Density, w.PH, w.Sulphates, w.Alcohol))).toDF("label", "features")
  42.  
  43.                     trainingDF.show()
  44.                     println("=============================================")
  45.  
  46.                     //创建线性回归对象
  47.                     val lr = new LogisticRegression()
  48.                     //设置最大迭代次数
  49.                     //lr.setMaxIter(10).setRegParam(0.01)
  50.                     lr.setMaxIter(10).setRegParam(0.01)
  51.                     //通过线性回归，拟合训练数据，生成model
  52.                     val model = lr.fit(trainingDF)
  53.  
  54.                     //val loadmodel = LinearRegressionModel.load("file:///d:/share/spark/model");
  55.  
  56.                     //model.save("file:///d:/share/spark/model")
  57.  
  58.                     //创建测试Dataframe
  59.                     //        val testDF = sess.createDataFrame(Seq((1.0,Vectors.dense(6.1, 0.32, 0.24, 1.5, 0.036, 43, 140, 0.9894, 3.36, 0.64, 10.7)),
  60.                     //            (0.0, Vectors.dense(5.2, 0.44, 0.04, 1.4, 0.036, 38, 124, 0.9898, 3.29, 0.42, 12.4)),
  61.                     //            (0.0,Vectors.dense(7.2, 0.32, 0.47, 5.1, 0.044, 19, 65, 0.9951, 3.38, 0.36, 9)),
  62.                     //            (0.0, Vectors.dense(6.4, 0.595, 0.14, 5.2, 0.058, 15, 97, 0.991, 3.03, 0.41, 12.6)))
  63.                     //        ).toDF("label", "features")
  64.  
  65.                     val testDF = sess.createDataFrame(Seq((5.0000, Vectors.dense(7.4,
  66.                         0.7, 0.0, 1.9, 0.076, 25.0, 67.0, 0.9968, 3.2, 0.68, 9.8)), (5.00000,
  67.                             Vectors.dense(7.8, 0.88, 0.0, 2.6, 0.098, 11.0, 34.0, 0.9978, 3.51, 0.56,
  68.                                 9.4)), (7.00000, Vectors.dense(7.3, 0.65, 0.0, 1.2, 0.065, 15.0, 18.0, 0.9968,
  69.                         3.36, 0.57, 9.5)))).toDF("label", "features")
  70.  
  71.                     //显式测试数据
  72.                     testDF.show();
  73.                     println("========================")
  74.                     //预测测试数据(带标签),评测模型的质量。
  75.                     testDF.createOrReplaceTempView("test")
  76.                     val tested = model.transform(testDF).select("features", "label", "prediction")
  77.                     tested.show();
  78.  
  79.                     println("========================")
  80.                     //预测无标签的测试数据。
  81.                     val predictDF = sess.sql("SELECT features FROM test")
  82.                     //预测结果
  83.                     val predicted = model.transform(predictDF).select("features", "prediction")
  84.                     predicted.show();
  85.                 }
  86.  
  87.             }
    6.Java代码实现 =====》 TODO


五、模型的保存和加载
------------------------------------------------------
    1.模型持久化Save
        //创建线性回归对象
        val lr = new LogisticRegression()
        //设置最大迭代次数
        //lr.setMaxIter(10).setRegParam(0.01)
        lr.setMaxIter(10)
        //通过线性回归，拟合训练数据，生成model
        val model = lr.fit(trainingDF)
        model.save("file:///d:/share/spark/model")

    2.模型的加载
        val loadmodel = LinearRegressionModel.load("file:///d:/share/spark/model");


六、垃圾邮件过滤[Hash词频 + 逻辑回归]
------------------------------------------------------
    

  1. import org.apache.spark.ml.classification.LogisticRegression
  2.     import org.apache.spark.ml.param.ParamMap
  3.     import org.apache.spark.ml.linalg.{Vector, Vectors}
  4.     import org.apache.spark.sql.{Row, SparkSession}
  5.     import org.apache.spark.ml.Pipeline
  6.     import org.apache.spark.ml.feature.{HashingTF, RegexTokenizer, StopWordsRemover, Tokenizer, Word2Vec}
  7.     /**
  8.       * 垃圾邮件过滤
  9.       */
  10.     object SparkMLSpamFilter {
  11.  
  12.         def main(args: Array[String]): Unit = {
  13.  
  14.             val spark = SparkSession.builder()
  15.                     .appName("ml")
  16.                     .master("local[4]")
  17.                     .getOrCreate();
  18.  
  19.             val sc = spark.sparkContext;
  20.  
  21.             //训练数据
  22.             val training = spark.createDataFrame(Seq(
  23.                 ("you@example.com", "hope you are well", 0.0),
  24.                 ("raj@example.com", "nice to hear from you", 0.0),
  25.                 ("thomas@example.com", "happy holidays", 0.0),
  26.                 ("mark@example.com", "see you tomorrow", 0.0),
  27.                 ("xyz@example.com", "save money", 1.0),
  28.                 ("top10@example.com", "low interest rate", 1.0),
  29.                 ("marketing@example.com", "cheap loan", 1.0))
  30.             ).toDF("email", "message", "label")
  31.  
  32.             training.show();
  33.  
  34.             //分词器：将输入的数据转成小写，然后按照若干空格分割开来，输出新的列
  35.             val tokenizer = new Tokenizer().setInputCol("message").setOutputCol("words")
  36.  
  37.             //hash词频  --- 设置桶数，设置输入列，设置输出列
  38.             val hashingTF = new HashingTF().setNumFeatures(1000).setInputCol("words").setOutputCol("features")
  39.  
  40.             //逻辑回归对象 --- 迭代10次，参数0.01
  41.             val lr = new LogisticRegression().setMaxIter(10).setRegParam(0.01)
  42.  
  43.             //新建管线
  44.             val pipeline = new Pipeline().setStages(Array(tokenizer, hashingTF, lr));
  45.  
  46.             //管线拟合数据，产生模型
  47.             val model = pipeline.fit(training)
  48.  
  49.             //测试数据
  50.             val test = spark.createDataFrame(Seq(
  51.                 ("you@example.com", "how are you"),
  52.                 ("jain@example.com", "hope doing well"),
  53.                 ("caren@example.com", "want some money"),
  54.                 ("zhou@example.com", "secure loan"),
  55.                 ("ted@example.com", "need loan"))
  56.             ).toDF("email", "message")
  57.  
  58.             test.show()
  59.  
  60.             //测试数据的结果
  61.             val prediction = model.transform(test).select("email", "message", "prediction")
  62.             //展示测试结果
  63.             prediction.show()
  64.  
  65.             println("==================================")
  66.             //分词，输出words
  67.             val wordsDF = tokenizer.transform(training)
  68.             wordsDF.show();
  69.             println("==================================")
  70.  
  71.             val featurizedDF = hashingTF.transform(wordsDF)
  72.             featurizedDF.show()
  73.             println("==================================")
  74.  
  75.             featurizedDF.createOrReplaceTempView("featurized")
  76.             val selectedFeaturizedFieldstDF = spark.sql("SELECT words, features FROM featurized")
  77.             selectedFeaturizedFieldstDF.show()
  78.         }
  79.     }
七、推荐[最小二乘法]
-------------------------------------------------------
    1.最小二乘法ALS
        点距离求和的最小值

    2.训练数据[test.data]
        1,0,1.0
        1,1,2.0
        1,2,5.0
        1,3,5.0
        1,4,5.0
        2,0,1.0
        2,1,2.0
        2,2,5.0
        2,5,5.0
        2,6,4.5
        3,1,2.5
        3,2,5.0
        3,3,4.0
        3,4,3.0
        4,0,5.0
        4,1,5.0
        4,2,5.0
        4,3,0.0

    3.scala实现
        

  1. package test.spark.examples.mllib
  2.  
  3.         import org.apache.spark.{SparkConf, SparkContext}
  4.         // $example on$
  5.         import org.apache.spark.mllib.recommendation.ALS
  6.         import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
  7.         import org.apache.spark.mllib.recommendation.Rating
  8.  
  9.         /**
  10.           * 推荐
  11.           */
  12.         object RecomendationDemo {
  13.             def main(args: Array[String]): Unit = {
  14.  
  15.                 val conf = new SparkConf().setAppName("CollaborativeFilteringExample").setMaster("local[*]")
  16.  
  17.                 val sc = new SparkContext(conf)
  18.                 // $example on$
  19.                 // 加载和解析数据
  20.                 val data = sc.textFile("file:///D:/share/spark/ml/data/mllib/als/test.data")
  21.  
  22.                 //将数据转换成Rating评分对象
  23.                 val ratings = data.map(
  24.                     _.split(',') match
  25.                     {
  26.                         case Array(user, item, rate) =>
  27.                             Rating(user.toInt, item.toInt, rate.toDouble)
  28.                     }
  29.                 )
  30.  
  31.                 println("==========    ratings：原始数据集    ===============")
  32.                 ratings.collect().foreach(println);
  33.  
  34.                 // 使用最小二乘法，构建模型
  35.                 val rank = 10
  36.                 val numIterations = 10
  37.                 val model = ALS.train(ratings, rank, numIterations, 0.01)
  38.  
  39.                 //通过测试数据，测试模型的准确性
  40.                 //准备测试数据，去掉评分
  41.                 val usersProducts = ratings.map { case Rating(user, product, rate) =>
  42.                     (user, product)
  43.                 }
  44.  
  45.                 println("==========    usersProducts：测试数据    ===============")
  46.                 usersProducts.collect().foreach(println)
  47.  
  48.                 //对usersProducts进行预测，产生rate
  49.                 val predictions =
  50.                     model.predict(usersProducts).map { case Rating(user, product, rate) =>
  51.                         ((user, product), rate)
  52.                     }
  53.                 println("==========    predictions：预测结果    ===============")
  54.                 predictions.collect().foreach(println)
  55.  
  56.                 //对比
  57.                 val ratesAndPreds = ratings.map { case Rating(user, product, rate) =>
  58.                     ((user, product), rate)
  59.                 }.join(predictions)
  60.                 println("==========    ratings + predictions:对比真实和预测结果    ===============")
  61.                 ratesAndPreds.collect().foreach(println)
  62.  
  63.                 //给2号用户推荐5款商品
  64.                 val res = model.recommendProducts(2,5);
  65.                 println("==========    res：给2号客户的推荐结果    ===============")
  66.                 res.foreach(println)
  67.  
  68.                 //计算误差
  69.                 val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =>
  70.                     val err = (r1 - r2)
  71.                     err * err
  72.                 }.mean()
  73.                 println("Mean Squared Error = " + MSE)
  74.  
  75.                 // 保存和加载模型的方式
  76.                 model.save(sc, "target/tmp/myCollaborativeFilter")
  77.                 val sameModel = MatrixFactorizationModel.load(sc, "target/tmp/myCollaborativeFilter")
  78.                 // $example off$
  79.             }
  80.         }
    4.java实现
        

  1. package test.spark.examples.mllib;
  2.  
  3.         import scala.Tuple2;
  4.         import org.apache.spark.api.java.*;
  5.         import org.apache.spark.api.java.function.Function;
  6.         import org.apache.spark.mllib.recommendation.ALS;
  7.         import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
  8.         import org.apache.spark.mllib.recommendation.Rating;
  9.         import org.apache.spark.SparkConf;
  10.  
  11.         public class RecomendationDemoJava {
  12.  
  13.             public static void main(String[] args) {
  14.                 // $example on$
  15.                 SparkConf conf = new SparkConf().setAppName("Java Collaborative Filtering Example").setMaster("local[*]");
  16.                 JavaSparkContext jsc = new JavaSparkContext(conf);
  17.  
  18.                 //加载和解析数据
  19.                 String path = "file:///D:/share/spark/ml/data/mllib/als/test.data";
  20.                 JavaRDD<String> data = jsc.textFile(path);
  21.                 JavaRDD<Rating> ratings = data.map(
  22.                         new Function<String, Rating>() {
  23.                             public Rating call(String s) {
  24.                                 String[] sarray = s.split(",");
  25.                                 return new Rating(Integer.parseInt(sarray[0]), Integer.parseInt(sarray[1]),
  26.                                         Double.parseDouble(sarray[2]));
  27.                             }
  28.                         }
  29.                 );
  30.                 System.out.println("==========    ratings：原始数据集    ===============");
  31.                 ratings.collect().forEach( x -> System.out.println(x));
  32.  
  33.  
  34.                 //使用最小二乘法构建模型
  35.                 int rank = 10;
  36.                 int numIterations = 10;
  37.                 MatrixFactorizationModel model = ALS.train(JavaRDD.toRDD(ratings), rank, numIterations, 0.01);
  38.  
  39.                 //通过测试数据，测试模型的准确性
  40.                 //准备测试数据，去掉评分
  41.                 JavaRDD<Tuple2<Object, Object>> userProducts = ratings.map(
  42.                         new Function<Rating, Tuple2<Object, Object>>() {
  43.                             public Tuple2<Object, Object> call(Rating r) {
  44.                                 return new Tuple2<Object, Object>(r.user(), r.product());
  45.                             }
  46.                         }
  47.                 );
  48.  
  49.                 System.out.println(("==========    usersProducts：测试数据    ==============="));
  50.                 userProducts.collect().forEach(x -> System.out.println(x._1 + ":" + x._2));
  51.  
  52.                 //对usersProducts进行预测，产生rate
  53.                 JavaPairRDD<Tuple2<Integer, Integer>, Double> predictions = JavaPairRDD.fromJavaRDD(
  54.                         model.predict(JavaRDD.toRDD(userProducts)).toJavaRDD().map(
  55.                                 new Function<Rating, Tuple2<Tuple2<Integer, Integer>, Double>>() {
  56.                                     public Tuple2<Tuple2<Integer, Integer>, Double> call(Rating r){
  57.                                         return new Tuple2<>(new Tuple2<>(r.user(), r.product()), r.rating());
  58.                                     }
  59.                                 }
  60.                         ));
  61.  
  62.                 System.out.println(("==========    predictions：预测结果    ==============="));
  63.                 predictions.collect().forEach(x-> System.out.println(x._1 + ":" + x._2));
  64.  
  65.                 System.out.println(("==========    给2号客户推荐5款商品    ==============="));
  66.                 Rating[] ratings1 = model.recommendProducts(2, 5);
  67.                 for(Rating r : ratings1)
  68.                 {
  69.                     System.out.println(r.user() + ":" + r.product() + ":" + r.rating());
  70.                 }
  71.  
  72.                 JavaRDD<Tuple2<Double, Double>> ratesAndPreds =
  73.                         JavaPairRDD.fromJavaRDD(ratings.map(
  74.                                 new Function<Rating, Tuple2<Tuple2<Integer, Integer>, Double>>() {
  75.                                     public Tuple2<Tuple2<Integer, Integer>, Double> call(Rating r){
  76.                                         return new Tuple2<>(new Tuple2<>(r.user(), r.product()), r.rating());
  77.                                     }
  78.                                 }
  79.                         )).join(predictions).values();
  80.  
  81.                 System.out.println(("==========    ratings + predictions:对比真实和预测结果    ==============="));
  82.                 ratesAndPreds.collect().forEach(x -> System.out.println(x._1 + ":" + x._2));
  83.  
  84.                 //计算误差
  85.                 double MSE = JavaDoubleRDD.fromRDD(ratesAndPreds.map(
  86.                         new Function<Tuple2<Double, Double>, Object>() {
  87.                             public Object call(Tuple2<Double, Double> pair) {
  88.                                 Double err = pair._1() - pair._2();
  89.                                 return err * err;
  90.                             }
  91.                         }
  92.                 ).rdd()).mean();
  93.                 System.out.println("Mean Squared Error = " + MSE);
  94.  
  95.                 // 模型的保存和加载
  96.                 //model.save(jsc.sc(), "target/tmp/myCollaborativeFilter");
  97.                 //MatrixFactorizationModel sameModel = MatrixFactorizationModel.load(jsc.sc(),
  98.                 //"target/tmp/myCollaborativeFilter");
  99.                 // $example off$
  100.  
  101.                 jsc.stop();
  102.             }
  103.         }
    5.猜你喜欢--推荐
        /*******向用户推荐n款商品********/
        val res = model.recommendProducts(5,8);

        /*******将指定的商品推荐给n个用户********/
        val res = model.recommendUsers(3,5)

        /*******向所有用户推荐3种商品********/
        val res = model.recommendProductsForUsers(3)


八、电影推荐案例
---------------------------------------------------------------
    1.元数据
        0::2::3::1424380312
        0::3::1::1424380312
        0::5::2::1424380312
        0::9::4::1424380312
        0::11::1::1424380312
        0::12::2::1424380312
        1::15::1::1424380312
        2::17::1::1424380312
        2::19::1::1424380312
        ...

    2.scala实现
        

  1. import org.apache.spark.{SparkConf, SparkContext}
  2.         import org.apache.spark.ml.evaluation.RegressionEvaluator
  3.         import org.apache.spark.ml.recommendation.ALS
  4.         import org.apache.spark.sql.SparkSession
  5.  
  6.  
  7.         object MovieRecc {
  8.  
  9.             //定义评级样例类【注意：样例类千万不要定义到main函数内部，会解析失败】
  10.             case class Rating0(userId: Int, movieId: Int, rating: Float, timestamp: Long)
  11.  
  12.             def main(args: Array[String]): Unit = {
  13.                 val conf = new SparkConf();
  14.                 conf.setAppName("movieRecomm");
  15.                 conf.setMaster("local[4]")
  16.  
  17.                 val spark = SparkSession.builder().config(conf).getOrCreate() ;
  18.                 import spark.implicits._
  19.  
  20.                 //解析评级
  21.                 def parseRating(str: String): Rating0 = {
  22.                     val fields = str.split("::")
  23.                     assert(fields.size == 4)
  24.                     Rating0(fields(0).toInt, fields(1).toInt, fields(2).toFloat, fields(3).toLong)
  25.                 }
  26.  
  27.                 //转换成Rating的DF对象
  28.                 var ratings = spark.sparkContext.textFile("file:///D:\\share\\spark\\ml\\data\\mllib\\als\\sample_movielens_ratings.txt");
  29.                 val ratings0 = ratings.map(parseRating)
  30.                 val df = ratings0.toDF()
  31.                 println("=========  df:数据源  ==========================")
  32.                 df.collect().foreach(println)
  33.  
  34.                 //随机切割训练数据，生成两个一个数组，第一个元素是training,第二个是test
  35.                 val Array(training, test) = df.randomSplit(Array(0.99, 0.01))
  36.  
  37.                 //构建ALS推荐算法并设置参数
  38.                 val als = new ALS().setMaxIter(5)
  39.                         .setRegParam(0.01)
  40.                         .setUserCol("userId")
  41.                         .setItemCol("movieId")
  42.                         .setRatingCol("rating")
  43.  
  44.                 //通过als对象对训练数据进行拟合,生成推荐模型
  45.                 val model = als.fit(training)
  46.  
  47.                 //使用模型对test，进行结果预测
  48.                 val predictions = model.transform(test);
  49.                 println("=========  predictions 预测结果  ==========================")
  50.                 predictions.collect().foreach(println)
  51.             }
  52.         }
    3.java实现
        














  1. package mllib;
  2.         import java.io.Serializable;
  3.  
  4.         import org.apache.spark.SparkConf;
  5.         import org.apache.spark.SparkContext;
  6.         import org.apache.spark.api.java.JavaRDD;
  7.         import org.apache.spark.ml.evaluation.RegressionEvaluator;
  8.         import org.apache.spark.ml.recommendation.ALS;
  9.         import org.apache.spark.ml.recommendation.ALSModel;
  10.         import org.apache.spark.sql.Dataset;
  11.         import org.apache.spark.sql.Row;
  12.         import org.apache.spark.sql.SparkSession;
  13.  
  14.         public class MovieReccJava {
  15.  
  16.             public static void main(String[] args) {
  17.                 SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("movie");
  18.                 SparkContext sc = new SparkContext(conf);
  19.                 SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
  20.  
  21.                 JavaRDD<Rating0> ratingsRDD = spark
  22.                         .read().textFile("file:///D:\\share\\spark\\ml\\data\\mllib\\als\\sample_movielens_ratings.txt").javaRDD()
  23.                         .map(Rating0::parseRating);
  24.  
  25.                 Dataset<Row> ratings = spark.createDataFrame(ratingsRDD, Rating0.class);
  26.                 Dataset<Row>[] splits = ratings.randomSplit(new double[]{0.99, 0.01});
  27.                 Dataset<Row> training = splits[0];
  28.                 Dataset<Row> test = splits[1];
  29.  
  30.                 System.out.println(("=========  trains 训练数据  =========================="));
  31.                 training.show();
  32.                 System.out.println(("=========  test 测试数据  =========================="));
  33.                 test.show();
  34.  
  35.  
  36.                 // Build the recommendation model using ALS on the training data
  37.                 ALS als = new ALS()
  38.                         .setMaxIter(5)
  39.                         .setRegParam(0.01)
  40.                         .setUserCol("userId")
  41.                         .setItemCol("movieId")
  42.                         .setRatingCol("rating");
  43.                 ALSModel model = als.fit(training);
  44.  
  45.                 Dataset<Row> predictions = model.transform(test);
  46.  
  47.                 System.out.println(("=========  predictions 预测结果  =========================="));
  48.                 predictions.show();
  49.             }
  50.         }

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/在线问答5/article/detail/932737
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号