What are the steps for Spark ALS implementation

This article mainly explains "what are the steps for the implementation of Spark ALS". The content of the explanation is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn what are the steps for the realization of Spark ALS.

The spark ALS algorithm is used for personalized recommendations, and the data sets it requires are data sets such as the user's scoring table for goods. The main implementation steps are as follows:

1. Define input data

2. Convert the input data to the rating data format, such as case class Rating (user: Int, movie: Int, rating: Float)

3. Design ALS model training data.

4. Calculate the recommendation data and store it for direct use by the business system.

Let's look at the specific code:

Package recommendimport org.apache.spark.sql.SparkSessionimport java.util.Propertiesimport org.apache.spark.rdd.RDDimport org.apache.spark.ml.evaluation.RegressionEvaluatorimport org.apache.spark.ml.recommendation.ALSimport org.apache.spark.ml.feature.StringIndexerimport org.apache.spark.sql.Datasetimport org.apache.spark.sql.Rowimport org.apache.spark.ml.feature.IndexToStringimport scala.collection.mutable.ArrayBufferimport org.apache.spark.TaskContextimport org.apache.spark.ml.Pipelineimport org.apache.spark.sql.SaveMode/ * Personalized recommendation ALS algorithm * the user's click rate on the resource is used as a score * * / 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") / / the date type is not parsed. GetOrCreate () trainModel (spark) spark.close ()} def trainModel (spark: SparkSession): Unit = {import spark.implicits._ val MAX = 3 / / maximum recommended number val rank = 10 / / Vector size Default 10 val iterations = 10 / / iterations Default 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) / / set username props.setProperty ("password", pass) / / set password 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). X.getLong (2) .toFloat) ratings.show () val Array (training, test) = ratings.randomSplit (Array (0.9) Println ("training:") training.show () println ("test:") test.show () / / implicit feedback and display feedback val als = new ALS () .setMaxIter (iterations) .setRegParam (0.01) .setImplications Prefs (false) .setUserC ol ("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) ) for (I {@ 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)}) Thank you for reading The above is the content of "what are the steps for the implementation of Spark ALS". After the study of this article, I believe you have a deeper understanding of what are the steps for the implementation of Spark ALS, and the specific use needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!