快上网专注成都网站设计 成都网站制作 成都网站建设
成都网站建设公司服务热线:028-86922220

网站建设知识

十年网站开发经验 + 多家企业客户 + 靠谱的建站团队

量身定制 + 运营维护+专业推广+无忧售后,网站问题一站解决

SparkALS实现的步骤是什么

这篇文章主要讲解了“Spark ALS实现的步骤是什么”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“Spark ALS实现的步骤是什么”吧!

公司主营业务:网站设计、做网站、移动网站开发等业务。帮助企业客户真正实现互联网宣传,提高企业的竞争能力。创新互联是一支青春激扬、勤奋敬业、活力青春激扬、勤奋敬业、活力澎湃、和谐高效的团队。公司秉承以“开放、自由、严谨、自律”为核心的企业文化,感谢他们对我们的高要求,感谢他们从不同领域给我们带来的挑战,让我们激情的团队有机会用头脑与智慧不断的给客户带来惊喜。创新互联推出吴川免费做网站回馈大家。

spark ALS算法是做个性推荐用的,它所需要的数据集是类似用户对商品的打分表之类的数据集。实现步骤主要以下几步:

1、定义输入数据

2、输入数据转换成评分数据格式,如case class Rating(user: Int, movie: Int, rating: Float)

3、设计ALS模型训练数据

4、计算推荐数据,存储起来供业务系统直接使用。

下面看看具体的代码:

package recommend
import org.apache.spark.sql.SparkSession
import java.util.Properties
import org.apache.spark.rdd.RDD

import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.ml.feature.StringIndexer
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.Row
import org.apache.spark.ml.feature.IndexToString
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.TaskContext
import org.apache.spark.ml.Pipeline
import org.apache.spark.sql.SaveMode

/**
 * 个性化推荐ALS算法
 * 用户对资源的点击率作为评分
 *
 */
object Recommend {

  case class Rating(user: Int, movie: Int, rating: Float)
 
  
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("Java Spark MySQL Recommend")
      .master("local")
      .config("es.nodes", "127.0.0.1")
      .config("es.port", "9200")
      .config("es.mapping.date.rich", "false") //不解析日期类型
      .getOrCreate()

    trainModel(spark)

    spark.close()
  }

  def trainModel(spark: SparkSession): Unit = {
    import spark.implicits._

    val MAX = 3 // 最大推荐数目
    val rank = 10 // 向量大小,默认10
    val iterations = 10 // 迭代次数,默认10


    val url = "jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf8"
    val table = "clicks"
    val user = "root"
    val pass = "123456"

    val props = new Properties()
    props.setProperty("user", user) // 设置用户名
    props.setProperty("password", pass) // 设置密码

    val clicks = spark.read.jdbc(url, table, props).repartition(4)
    clicks.createOrReplaceGlobalTempView("clicks")

    val agg = spark.sql("SELECT userId ,resId ,COUNT(id)  AS clicks FROM global_temp.clicks GROUP BY userId,resId")
   
    val userIndexer = new StringIndexer()
      .setInputCol("userId")
      .setOutputCol("userIndex")

    val resIndexer = new StringIndexer()
      .setInputCol("resId")
      .setOutputCol("resIndex")

    val indexed1 = userIndexer.fit(agg).transform(agg)
    val indexed2 = resIndexer.fit(indexed1).transform(indexed1)
    indexed2.show()

    val ratings = indexed2.map(x => Rating(x.getDouble(3).toInt, x.getDouble(4).toInt, x.getLong(2).toFloat))
    ratings.show()

    val Array(training, test) = ratings.randomSplit(Array(0.9, 0.1))
    println("training:")
    training.show()
    println("test:")
    test.show()

    //隐性反馈和显示反馈
    val als = new ALS()
      .setMaxIter(iterations)
      .setRegParam(0.01)
      .setImplicitPrefs(false)
      .setUserCol("user")
      .setItemCol("movie")
      .setRatingCol("rating")

    val model = als.fit(ratings)

    // Evaluate the model by computing the RMSE on the test data
    // Note we set cold start strategy to 'drop' to ensure we don't get NaN evaluation metrics
    model.setColdStartStrategy("drop")
    val predictions = model.transform(test)

    val r2 = model.recommendForAllUsers(MAX)

    println(r2.schema)

    val result = r2.rdd.flatMap(row => {
      val userId = row.getInt(0)
      val arrayPredict: Seq[Row] = row.getSeq(1)

      var result = ArrayBuffer[Rating]()
      arrayPredict.foreach(rowPredict => {
        val p = rowPredict(0).asInstanceOf[Int]
        val score = rowPredict(1).asInstanceOf[Float]
        val sql = "insert into recommends(userId,resId,score) values (" +
          userId + "," +
          rowPredict(0) + "," +
          rowPredict(1) +
          ")"
        println("sql:" + sql)
        result.append(Rating(userId, p, score))
      })
      for (i <- result) yield {
        i
      }
    })

    println("推荐结果RDD已展开")
    result.toDF().show()

    //资源id隐射
    val resInt2Index = new IndexToString()
      .setInputCol("movie")
      .setOutputCol("resId")
      .setLabels(resIndexer.fit(indexed1).labels)

    //userId映射
    val userInt2Index = new IndexToString()
      .setInputCol("user")
      .setOutputCol("userId")
      .setLabels(userIndexer.fit(agg).labels)

    val rc = userInt2Index.transform(resInt2Index.transform(result.toDF()))
    rc.show()

    rc.withColumnRenamed("rating","score").select("userId", "resId","score").write.mode(SaveMode.Overwrite)
      .format("jdbc")
      .option("url", url)
      .option("dbtable", "recommends")
      .option("user", user)
      .option("password", pass)
      .option("batchsize", "5000")
      .option("truncate", "true")
      .save

      println("finished!!!")
  }

}

DataFrame写入mysql还有另一种写法,就是原生写入:

    //分区写推荐结果到mysql
        r2.foreachPartition(p => {
          @transient val conn = ConnectionPool.getConnection
          p.foreach(row => {
            val userId = row.getInt(0)
            val arrayPredict: Seq[Row] = row.getSeq(1)
            arrayPredict.foreach(rowPredict => {
              println(rowPredict(0) + "@" + rowPredict(1))
              val sql = "insert into recommends(userId,resId,score) values (" +
                  userId+"," +
                rowPredict(0)+","+
               rowPredict(1) +
                ")"
               println("sql:"+sql)
              val stmt = conn.createStatement
              stmt.executeUpdate(sql)
            })
          })
          ConnectionPool.returnConnection(conn)
        })

感谢各位的阅读,以上就是“Spark ALS实现的步骤是什么”的内容了,经过本文的学习后,相信大家对Spark ALS实现的步骤是什么这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是创新互联,小编将为大家推送更多相关知识点的文章,欢迎关注!


文章名称:SparkALS实现的步骤是什么
当前网址:http://6mz.cn/article/gioids.html

其他资讯