• 本例使用 MovieLens 数据集中的两个文件：“ ratings.dat.” 和“ movies.dat” 。所有的评分数据按照下面的格式存储“ ratings.dat” 中 。

• UserID ：： MovieID ：： Rating ：： Timestamp

• 在“ movies.dat” 中以下面的格式存储电影信息 。

• MovieID ：： Title ：： Genres

• 针对本例使用一个 standalone 项目模板。假设在用户的环境中，已经配置好所需路径和文件（实例下载地址为 https:// github.com/amplab/training/tree/ampcamp4/machine-learning/scala ）， 这些已经在 /root/machine-learning/ scala / 中设置 ，目录 下找到以下 选项：

• sbt ：包含 SBT 工具的目录。

• build.sbt ： SBT 项目文件。

• MovieLensALS.scala ：用户需要编译和运行的主要 Scala 主程序。

• solution ：包含 solution 代码的目录。

• 用户需要编辑、编译和运行的主要文件是 MovieLensALS.scala ，可以将下面的代码模板拷贝到文件中。

主要运行思路：先在本地编译好可执行jar包，然后spark-submit进行编译运行，中间遇到许多问题，特意记录与大家分享。

编译环境： scala-2.11 + JDK 1.8 + sbt 1.0.3 + spark 集群（注意黑体的版本对应要不然容易出错，有坑）

spark集群：如下图。

建立一个sbt工程，其中的sbt,scala,jdk的版本都可以进行选择。

然后把代码copy进去，然后修改相关配置，代码如下：

import java.util.Random
import org.apache.log4j.Logger
import org.apache.log4j.Level
import scala.io.Source
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.rdd._
import org.apache.spark.mllib.recommendation.{ALS, Rating, MatrixFactorizationModel}
object MovieLensALS{def main(args: Array[String]) {Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)}//  val sparkHome = "/zzti/libs/spark"
//  val master = "local"
  val conf = new SparkConf()
//    .setMaster(master)
//    .setSparkHome(sparkHome)
    .setAppName("MovieLensALS").set("spark.executor.memory", "2g")//  System.setProperty("hadoop.home.dir", "H:\\大三\\spark\\winutils")
  val sc = new SparkContext(conf)//H:\大三\spark\MLib算法\data_movies\ml-1m 本机
  ///movielens/medium/ratings.dat
  val ratings = sc.textFile("/movielens/medium/ratings.dat").map { line =>val fields = line.split("::")(fields(3).toInt % 10, Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble))}//H:\大三\spark\MLib算法\data_movies\ml-1m
   ///movielens/medium/movies.dat
  val movies = sc.textFile("/movielens/medium/movies.dat").map { line =>val fields = line.split("::")(fields(0).toInt, fields(1))}.collect.toMapval numRatings = ratings.countval numUsers = ratings.map(_._2.user).distinct.countval numMovies = ratings.map(_._2.product).distinct.countprintln("Got " + numRatings + " ratings from "
    + numUsers + " users on " + numMovies + " movies.")val mostRateMovieIds = ratings.map(_._2.product).countByValue().toSeq.sortBy(-_._2).take(50).map(_._1)//获它们的id
  val random = new Random(0)val seclectedMovies = mostRateMovieIds.filter(x => random.nextDouble() < 0.2).map(x => (x, movies(x))).toSeqval myRatings = elicitateRatings(seclectedMovies)val myRatingsRDD = sc.parallelize(myRatings)val numPartitions = 20
  val training = ratings.filter(x => x._1 < 6).values.union(myRatingsRDD).repartition(numPartitions).persistval validation = ratings.filter(x => x._1 >= 6 && x._1 < 8).values.repartition(numPartitions).persistval test = ratings.filter(x => x._1 >= 8).values.persistval numTraining = training.countval numValidation = validation.countval numTest = test.countprintln("Training:" + numTraining + ",validation: " + numValidation + ", test:" + numTest)val ranks = List(8, 12)val lambdas = List(0.1, 10.0)val numIters = List(10, 20)var bestModel: Option[MatrixFactorizationModel] = Nonevar bestValidationRmse = Double.MaxValue
  var bestRank = 0
  var bestLambda = -1.0
  var bestNumIter = -1
  for (rank <- ranks; lambda <- lambdas; numIter <- numIters) {val model = ALS.train(training, rank, numIter, lambda)val validationRmse = computeRmse(model, validation, numValidation)println("RMSE (validation)=" + validationRmse + "for the model trained with rand =" + rank + ", lambda=" + lambda + ", and numIter= " + numIter + ".")if (validationRmse < bestValidationRmse) {bestModel = Some(model)bestValidationRmse = validationRmsebestRank = rankbestLambda = lambdabestNumIter = numIter}}val testRmse = computeRmse(bestModel.get, test, numTest)println("The best model was trained with rank=" + bestRank + " and lambda =" + bestLambda + ", and numIter =" + bestNumIter + ", and itsRMSE on the test set is" + testRmse + ".")val myRateMoviesIds = myRatings.map(_.product).toSetval candidates = sc.parallelize(movies.keys.filter(!myRateMoviesIds.contains(_)).toSeq)val recommendations = bestModel.get.predict(candidates.map((0, _))).collect().sortBy((-_.rating)).take(50)var i = 1
  println("movies recommended for you:")recommendations.foreach { r =>println("%2d".format(i) + ":" + movies(r.product))i += 1

  }/** Compute RMSE (Root Mean Squared Error). */
  def computeRmse(model: MatrixFactorizationModel, data: RDD[Rating], n: Long) = {val predictions: RDD[Rating] = model.predict(data.map(x => (x.user, x.product)))val predictionsAndRatings = predictions.map(x => ((x.user, x.product), x.rating)).join(data.map(x => ((x.user, x.product), x.rating))).valuesmath.sqrt(predictionsAndRatings.map(x => (x._1 - x._2) * (x._1 - x._2)).reduce(_ + _) / n)}/** Elicitate ratings from command-line. */
  def elicitateRatings(movies: Seq[(Int, String)]) = {val prompt = "Please rate the following movie (1-5 (best), or 0 if not seen):"
    println(prompt)val ratings = movies.flatMap { x =>var rating: Option[Rating] = Nonevar valid = false
      while (!valid) {print(x._2 + ": ")try {val r = Console.readIntif (r < 0 || r > 5) {println(prompt)} else {valid = true
            if (r > 0) {rating = Some(Rating(0, x._1, r))}}} catch {case e: Exception => println(prompt)}}rating match {case Some(r) => Iterator(r)case None => Iterator.empty
      }}if (ratings.isEmpty) {error("No rating provided!")} else {ratings}}
}

然后运行结果报错2个：NoClassDefFoundError，以及找不到winutils.exe（不影响程序）

解决方案：因为版本错误，jdk 1.8与高版本的scala不兼容，所以在build.sbt里面修改scala版本号为2.11.8如下图然后重新build 一下 ok解决。

然后就可以打可执行jar包了，如图用idea打jar包

选择project Structure 在选择Atifacts->jar ->from modules with denpencies （不写了网上太多了， 具体百度吧），注意一下Main Class 要与Object 的scala 对象一致、还有MENIFEST.MF这一项要放到src->main里面，如果重复打jar，要删除之后  重新打。

效果如下：