0
0

educoder-Spark机器学习

第1关：基于物品的推荐算法

  1.     给用户2推荐2个商品。利用spark.mllib中的矩阵计算库，构建用户与物品的打分矩阵，然后计算物品之间的相似分数，进行推荐。实现基于用户（User CF）的协同过滤算法。
  1. import org.apache.log4j.{Level, Logger}
  2. import org.apache.spark.mllib.linalg.SparseVector
  3. import org.apache.spark.mllib.linalg.distributed.{CoordinateMatrix, IndexedRow, MatrixEntry, RowMatrix}
  4. import org.apache.spark.rdd.RDD
  5. import org.apache.spark.{SparkConf, SparkContext}
  6. object ItemBasedCF {
  7.   def main(args: Array[String]): Unit = {
  9.     Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
  10.     Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
  12.     //读入数据
  13.     val conf = new SparkConf().setAppName("ItemBasedCFModel").setMaster("local")
  14.     val sc = new SparkContext(conf)
  15.     val data = sc.textFile("/root/data/als/ratingdata.txt")
  17.     /*MatrixEntry代表一个分布式矩阵中的每一行(Entry)
  18.      * 这里的每一项都是一个(i: Long, j: Long, value: Double) 指示行列值的元组tuple。
  19.      * 其中i是行坐标，j是列坐标，value是值。*/
  20.     val parseData: RDD[MatrixEntry] =
  21.       data.map(_.split(",") match { case Array(user, item, rate) => MatrixEntry(user.toLong, item.toLong, rate.toDouble) })
  23.     //CoordinateMatrix是Spark MLLib中专门保存user_item_rating这种数据样本的
  24.     val ratings = new CoordinateMatrix(parseData)
  26.     /* 由于CoordinateMatrix没有columnSimilarities方法，所以我们需要将其转换成RowMatrix矩阵，调用他的columnSimilarities计算其相似性
  27.      * RowMatrix的方法columnSimilarities是计算，列与列的相似度，现在是user_item_rating，与基于用户的CF不同的是，这里不需要进行矩阵的转置，直接就是物品的相似*/
  28.      
  29.     val matrix: RowMatrix = ratings.toRowMatrix()
  31.     //需求：为某一个用户推荐商品。基本的逻辑是：首先得到某个用户评价过（买过）的商品，然后计算其他商品与该商品的相似度，并排序；从高到低，把不在用户评价过
  32.     //商品里的其他商品推荐给用户。
  33.     //例如：为用户2推荐商品
  35.     //第一步：得到用户2评价过（买过）的商品  take(5)表示取出所有的5个用户  2:表示第二个用户
  36.     //解释：SparseVector：稀疏矩阵
  37.     val user2pred = matrix.rows.take(5)(2)
  38.     val prefs: SparseVector = user2pred.asInstanceOf[SparseVector]
  39.     val uitems = prefs.indices //得到了用户2评价过（买过）的商品的ID
  40.     val ipi = (uitems zip prefs.values) //得到了用户2评价过（买过）的商品的ID和评分，即：(物品ID,评分)
  42.     //计算物品的相似性，并输出
  43.     val similarities = matrix.columnSimilarities()
  44.     val indexdsimilar = similarities.toIndexedRowMatrix().rows.map {
  45.       case IndexedRow(idx, vector) => (idx.toInt, vector)
  46.     }
  48.     //ij表示：其他用户购买的商品与用户2购买的该商品的相似度
  49.     val ij = sc.parallelize(ipi).join(indexdsimilar).flatMap {
  50.       case (i, (pi, vector: SparseVector)) => (vector.indices zip vector.values)
  51.     }
  53.     /********** begin **********/
  54.     //ij1表示：其他用户购买过，但不在用户2购买的商品的列表中的商品和评分
  55.     val ij1 = ij.filter { case (item, pref) => !uitems.contains(item) }
  57.     //将这些商品的评分求和，并降序排列，并推荐前两个物品
  58.     val ij2 = ij1.reduceByKey(_ + _).sortBy(_._2, false).take(2)
  59.     
  60.     /********** end **********/
  61.     
  62.     // crgjl
  63.     //取消以下1行注释
  64.     for (id <- ij2) print(id._1 + " ")
  65.     sc.stop()
  66.   }
  67. }

第2关：基于用户的推荐算法

  1.     根据提示，在右侧编辑器补充代码.实现：找出与用户1最相似的2个用户。
  1. import org.apache.log4j.{Level, Logger}
  2. import org.apache.spark.mllib.linalg.distributed.{CoordinateMatrix, MatrixEntry, RowMatrix}
  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.{SparkConf, SparkContext}
  6. object UserBasedCF {
  7.   def main(args: Array[String]): Unit = {
  8.     Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
  9.     Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
  10.     // 创建一个SparkContext
  11.     val conf = new SparkConf().setAppName("UserBasedCF").setMaster("local")
  12.     val sc = new SparkContext(conf)
  13.     // 读入数据
  14.     val data = sc.textFile("/root/data/als/ratingdata.txt")
  15.     // 解析出评分矩阵的每一行
  16.     val parseData: RDD[MatrixEntry] = data.map(_.split(",")
  17.     match { case Array(user, item, rate) =>
  18.         MatrixEntry(user.toLong, item.toLong, rate.toDouble)
  19.     })
  20.     // 构建关联矩阵
  21.     val ratings = new CoordinateMatrix(parseData)
  23.     // 转置矩阵以计算列(用户)的相似性
  24.     val matrix: RowMatrix = ratings.transpose().toRowMatrix()
  26.     // 计算得到用户的相似度矩阵
  27.     val similarities = matrix.columnSimilarities()
  29.     // 得到某个用户对所有物品的评分
  30.     val ratingOfUser1 = ratings.entries.filter(_.i == 1).
  31.       map(x => (x.j, x.value)).
  32.       sortBy(_._1).
  33.       map(_._1).
  34.       collect().
  35.       toList.
  36.       toArray
  38.     // 得到用户1相对于其他用户的相似性
  39.     val similarityOfUser1 = similarities.entries.filter(_.i == 1).
  40.       sortBy(_.value, false).
  41.       map(_.value).
  42.       collect
  44.     // 需求：为用户1推荐2个商品
  45.     // 思路：找到与用户1相似性最高的两个用户，将这两个用户评过分的物品，用户1没有评过分的物品推荐给用户1
  46.     /********** begin **********/
  48.     //找到与用户1相似性最高的两个用户
  49.     val similarityTopUser = similarities.entries.filter(_.i == 1).
  50.       sortBy(_.value, false).
  51.       map(x=>(x.j, x.value)).
  52.       collect.
  53.       take(2)
  54.     
  55.     //println("与用户1最相似的两个用户如下：")
  56.     //取消以下2行注释
  57.     for (s <- similarityTopUser) print(s._1 + " ")
  58.     
  59.     for (s <- similarityTopUser) {
  60.       // 找到这两个用户评过分的商品，与用户1没有评过分的物品
  61.       val userId = s._1
  62.       val ratingOfTemp = ratings.entries.filter(_.i == userId).
  63.         map(x => (x.j, x.value)).
  64.         sortBy(_._1).
  65.         map(_._1).
  66.         collect().
  67.         toList.
  68.         toArray
  69.       // 用户1与当前用户求差集
  70.       val dis = ratingOfTemp diff ratingOfUser1
  71.       //println("用户" + userId + "要推荐给用户1的商品id为")
  72.       for (id <- dis) print(id + " ")
  73.     }
  74.     
  75.     /********** end **********/
  77.     sc.stop()
  78.   }
  79. }

第3关：基于ALS的推荐算法

  1.     根据提示，在右侧编辑器补充代码。创建一个
  1. ALS

模型，使用调用fit方法，使用

  1. training

训练生成

  1. model

。

  1. import org.apache.spark.ml.evaluation.RegressionEvaluator
  2. import org.apache.spark.ml.recommendation.ALS
  3. import org.apache.spark.sql.{DataFrame, SparkSession}
  5. object ALS {
  6.   case class Rating(userId: Int, movieId: Int, rating: Float)
  7.   def parseRating(str: String): Rating = {
  8.     val fields = str.split(",")
  9.     assert(fields.size == 3)
  10.     Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat)
  11.   }
  13.   def main(args: Array[String]) {
  14.     val spark = SparkSession
  15.       .builder
  16.       .master("local")
  17.       .appName("ALS")
  18.       .getOrCreate()
  19.     import spark.implicits._
  21.     val ratings = spark.read.textFile("data/als/ratingdata.txt")
  22.       .map(parseRating)
  23.       .toDF()
  24.     val Array(training, test) = ratings.randomSplit(Array(0.8, 0.2))
  25.     /********** begin **********/
  26.     // Build the recommendation model using ALS on the training data
  27.     
  28.     val als = new ALS()  
  29.       .setMaxIter(5)  
  30.       .setRegParam(0.01)  
  31.       .setUserCol("userId")  
  32.       .setItemCol("movieId")  
  33.       .setRatingCol("rating")  
  34.     val model = als.fit(training) 
  35.     
  37.     /********** end **********/
  38.     // "Evaluate the model by computing the RMSE on the test data"
  39.     // "Note we set cold start strategy to 'drop' to ensure we don't get NaN evaluation metrics"
  40.     //取消以下3行注释
  41.       model.setColdStartStrategy("drop")
  42.     val predictions = model.transform(test)
  43.     evaluatingRMSE(predictions)
  44.     spark.stop()
  45.   }
  46.   def evaluatingRMSE(predictions:DataFrame):Unit = {
  47.     val evaluator = new RegressionEvaluator()
  48.       .setMetricName("rmse")
  49.       .setLabelCol("rating")
  50.       .setPredictionCol("prediction")
  51.     val rmse = evaluator.evaluate(predictions)
  52.     if (rmse <= 2){
  53.       print("\n" + "good")
  54.     }else{
  55.       println()
  56.       predictions.show(false)
  57.     }
  58.   }
  59. }

第4关：基于随机森林预测贷款风险

  1.     编写一个预测贷款风险的随机森林二分类模型。
  1. import org.apache.spark.rdd.RDD
  2. import org.apache.spark.sql.{DataFrame,SparkSession}
  3. import org.apache.spark.ml.classification.RandomForestClassifier
  4. import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
  5. import org.apache.spark.ml.feature.StringIndexer
  6. import org.apache.spark.ml.feature.VectorAssembler
  7. import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder}
  8. import org.apache.spark.ml.{Pipeline, PipelineStage}
  9. import org.apache.spark.mllib.evaluation.RegressionMetrics
  10. object Credit {
  11.   case class Credit(
  12.                      creditability: Double,
  13.                      balance: Double, duration: Double, history: Double, purpose: Double, amount: Double,
  14.                      savings: Double, employment: Double, instPercent: Double, sexMarried: Double, guarantors: Double,
  15.                      residenceDuration: Double, assets: Double, age: Double, concCredit: Double, apartment: Double,
  16.                      credits: Double, occupation: Double, dependents: Double, hasPhone: Double, foreign: Double
  17.                    )
  19.   def parseCredit(line: Array[Double]): Credit = {
  20.     Credit(
  21.       line(0),
  22.       line(1) - 1, line(2), line(3), line(4), line(5),
  23.       line(6) - 1, line(7) - 1, line(8), line(9) - 1, line(10) - 1,
  24.       line(11) - 1, line(12) - 1, line(13), line(14) - 1, line(15) - 1,
  25.       line(16) - 1, line(17) - 1, line(18) - 1, line(19) - 1, line(20) - 1
  26.     )
  27.   }
  29.   def parseRDD(rdd: RDD[String]): RDD[Array[Double]] = {
  30.     rdd.map(_.split(",")).map(_.map(_.toDouble))
  31.   }
  33.   def evaluatingAUC(predictedResultDF:DataFrame, labelstring:String):Unit = {
  34.     val evaluator = new BinaryClassificationEvaluator().setLabelCol(labelstring).setRawPredictionCol("prediction")
  35.     val predictionAUC = evaluator.setMetricName("areaUnderROC").evaluate(predictedResultDF)
  36.     if(predictionAUC > 0.6){
  37.         print("\n" + "good")
  38.     }else{
  39.         print(s"areaUnderROC: $predictionAUC")
  40.     }
  41.   }
  43.   def main(args: Array[String]) {
  45.     val spark = SparkSession
  46.       .builder
  47.       .appName("Credit")
  48.       .master("local")
  49.       .getOrCreate()
  51.     import spark.implicits._
  52.     val creditDF = parseRDD(spark.sparkContext.textFile("/root/data/germancredit.csv")).map(parseCredit).toDF()
  53.     creditDF.createTempView("credit")
  54.     
  55.     val featureCols = Array("balance", "duration", "history", "purpose", "amount",
  56.       "savings", "employment", "instPercent", "sexMarried", "guarantors",
  57.       "residenceDuration", "assets", "age", "concCredit", "apartment",
  58.       "credits", "occupation", "dependents", "hasPhone", "foreign")
  60.     /********** begin **********/
  61.     
  62.     // 合并特征列。
  63.     val assembler = new VectorAssembler().setInputCols(featureCols).setOutputCol("features")  
  64.     val df2 = assembler.transform(creditDF)
  65.     
  66.     /********** end **********/
  68.     //取消以下4行注释
  69.     val labelIndexer = new StringIndexer().setInputCol("creditability").setOutputCol("label")
  70.     val df3 = labelIndexer.fit(df2).transform(df2)
  71.     val splitSeed = 5043
  72.     val Array(trainingData, testData) = df3.randomSplit(Array(0.7, 0.3), splitSeed)
  74.     /********** begin **********/
  75.     
  76.     //调用随机森林API，使用trainingData训练生成模型model
  77.      val classifier = new RandomForestClassifier().setImpurity("gini").setMaxDepth(5).setNumTrees(20).setFeatureSubsetStrategy("auto").setSeed(5043)  
  78.     val model = classifier.fit(trainingData)  
  79.     
  80.     /********** end **********/
  82.     //取消以下2行注释
  83.     val predictions = model.transform(testData)
  84.       evaluatingAUC(predictions,"label")
  85.     spark.stop()
  86.   }
  87. }

第5关：基于多层感知器的手机短信分类

  1.     编写一个短信文本分类的程序。使用Spark.ml中的多层感知器（MLP，Multi Layer Perceptron Classifier）API——MultilayerPerceptronClassifer。
  1. import org.apache.spark.ml.Pipeline
  2. import org.apache.spark.ml.classification.MultilayerPerceptronClassifier
  3. import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
  4. import org.apache.spark.ml.feature.{IndexToString, StringIndexer, Word2Vec}
  5. import org.apache.spark.sql.{DataFrame,SparkSession}
  7. object SMSClassifier {
  8.   final val VECTOR_SIZE = 100
  10.   def evaluatingAUC(predictedResultDF:DataFrame, labelcol: String):Unit = {
  11.     val evaluator = new BinaryClassificationEvaluator().setLabelCol(labelcol).setRawPredictionCol("prediction")
  12.     val predictionAUC = evaluator.setMetricName("areaUnderROC").evaluate(predictedResultDF)
  13.     if(predictionAUC > 0.8){
  14.       print("\n" + "good")
  15.     }else{
  16.       print(s"areaUnderROC: $predictionAUC")
  17.     }
  18.   }
  20.   def main(args: Array[String]) {
  22.     val spark = SparkSession
  23.       .builder
  24.       .master("local")
  25.       .appName("SMS Message Classification (HAM or SPAM)")
  26.       .getOrCreate()
  28.     val parsedRDD = spark.sparkContext.textFile("data/smsspamcollection/SMSSpamCollection").map(_.split("\t")).map(eachRow => {
  29.       (eachRow(0),eachRow(1).split(" "))
  30.     })
  32.     val msgDF = spark.createDataFrame(parsedRDD).toDF("label","message")
  34.     val labelIndexer = new StringIndexer()
  35.       .setInputCol("label")
  36.       .setOutputCol("indexedLabel")
  37.       .fit(msgDF)
  39.     /********** begin **********/
  40.      val word2Vec = new Word2Vec()  
  41.       .setInputCol("message")  
  42.       .setOutputCol("features")  
  43.       .setVectorSize(VECTOR_SIZE)  
  44.       .setMinCount(1)  
  45.     /********** end **********/
  47.     val layers = Array[Int](VECTOR_SIZE,6,5,2)
  48.     
  49.     /********** begin **********/
  50.     val mlpc = new MultilayerPerceptronClassifier()  
  51.       .setLayers(layers)  
  52.       .setBlockSize(512)  
  53.       .setSeed(1234L)  
  54.       .setMaxIter(128)  
  55.       .setFeaturesCol("features")  
  56.       .setLabelCol("indexedLabel")  
  57.       .setPredictionCol("prediction") 
  59.     /********** end **********/
  61.     val labelConverter = new IndexToString()
  62.       .setInputCol("prediction")
  63.       .setOutputCol("predictedLabel")
  64.       .setLabels(labelIndexer.labels)
  66.     val Array(trainingData, testData) = msgDF.randomSplit(Array(0.8, 0.2))
  68.     /********** begin **********/
  69.     val pipeline = new Pipeline().setStages(Array(labelIndexer,word2Vec,mlpc,labelConverter))  
  70.     val model = pipeline.fit(trainingData)  
  71.     
  72.     /********** end **********/
  73.   
  74.     //取消以下两行注释
  75.       val predictionResultDF = model.transform(testData)
  76.     evaluatingAUC(predictionResultDF,"indexedLabel")
  77.     spark.stop()
  78.   }
  79. }
标签： spark 机器学习 大数据
本文转载自: https://blog.csdn.net/qq_48664727/article/details/125177812
版权归原作者 多多读书~ 所有， 如有侵权，请联系我们删除。
登录后发布评论

“educoder-Spark机器学习”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。