協(xié)同過濾

算法介紹：

協(xié)同過濾常被用于推薦系統(tǒng)。這類技術(shù)目標(biāo)在于填充“用戶－商品”聯(lián)系矩陣中的缺失項。Spark.ml目前支持基于模型的協(xié)同過濾，其中用戶和商品以少量的潛在因子來描述，用以預(yù)測缺失項。Spark.ml使用交替最小二乘(ALS)算法來學(xué)習(xí)這些潛在因子。

＊注意基于DataFrame的ALS接口目前僅支持整數(shù)型的用戶和商品編號。

顯式與隱式反饋

基于矩陣分解的協(xié)同過濾的標(biāo)準(zhǔn)方法中，“用戶－商品”矩陣中的條目是用戶給予商品的顯式偏好，例如，用戶給電影評級。然而在現(xiàn)實世界中使用時，我們常常只能訪問隱式反饋(如意見、點擊、購買、喜歡以及分享等)，在spark.ml中我們使用“隱式反饋數(shù)據(jù)集的協(xié)同過濾“來處理這類數(shù)據(jù)。本質(zhì)上來說它不是直接對評分矩陣進(jìn)行建模，而是將數(shù)據(jù)當(dāng)作數(shù)值來看待，這些數(shù)值代表用戶行為的觀察值(如點擊次數(shù)，用戶觀看一部電影的持續(xù)時間)。這些數(shù)值被用來衡量用戶偏好觀察值的置信水平，而不是顯式地給商品一個評分。然后，模型用來尋找可以用來預(yù)測用戶對商品預(yù)期偏好的潛在因子。

正則化參數(shù)

我們調(diào)整正則化參數(shù)regParam來解決用戶在更新用戶因子時產(chǎn)生新評分或者商品更新商品因子時收到的新評分帶來的最小二乘問題。這個方法叫做“ALS-WR”它降低regParam對數(shù)據(jù)集規(guī)模的依賴，所以我們可以將從部分子集中學(xué)習(xí)到的最佳參數(shù)應(yīng)用到整個數(shù)據(jù)集中時獲得同樣的性能。

參數(shù)：

alpha:

類型：雙精度型。

含義：隱式偏好中的alpha參數(shù)(非負(fù))。

checkpointInterval:

類型：整數(shù)型。

含義：設(shè)置檢查點間隔(>=1)，或不設(shè)置檢查點(-1)。

implicitPrefs:

類型：布爾型。

含義：特征列名。

itemCol:

類型：字符串型。

含義：商品編號列名。

maxIter:

類型：整數(shù)型。

含義：迭代次數(shù)(>=0)。

nonnegative:

類型：布爾型。

含義：是否需要非負(fù)約束。

numItemBlocks:

類型：整數(shù)型。

含義：商品數(shù)目(正數(shù))。

numUserBlocks:

類型：整數(shù)型。

含義：用戶數(shù)目(正數(shù))。

predictionCol:

類型：字符串型。

含義：預(yù)測結(jié)果列名。

rank:

類型：整數(shù)型。

含義：分解矩陣的排名(正數(shù))。

ratingCol:

類型：字符串型。

含義：評分列名。

regParam:

類型：雙精度型。

含義：正則化參數(shù)(>=0)。

seed:

類型：長整型。

含義：隨機(jī)種子。

userCol:

類型：字符串型。

含義：用戶列名。

調(diào)用示例：

下面的例子中，我們從MovieLens dataset讀入評分?jǐn)?shù)據(jù)，每一行包括用戶、電影、評分以及時間戳。我們默認(rèn)其排序是顯式的來訓(xùn)練ALS模型。我們通過預(yù)測評分的均方根誤差來評價推薦模型。如果評分矩陣來自其他信息來源，也可將implicitPrefs設(shè)置為true來獲得更好的結(jié)果。

Scala:

import org.apache.spark.ml.evaluation.RegressionEvaluator

import org.apache.spark.ml.recommendation.ALS

case class Rating(userId: Int, movieId: Int, rating: Float, timestamp: Long)

def parseRating(str: String): Rating = {

val fields = str.split("::")

assert(fields.size == 4)

Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat, fields(3).toLong)

}

val ratings = spark.read.textFile("data/mllib/als/sample_movielens_ratings.txt")

.map(parseRating)

.toDF()

val Array(training, test) = ratings.randomSplit(Array(0.8, 0.2))

// Build the recommendation model using ALS on the training dataval als = new ALS()

.setMaxIter(5)

.setRegParam(0.01)

.setUserCol("userId")

.setItemCol("movieId")

.setRatingCol("rating")

val model = als.fit(training)

// Evaluate the model by computing the RMSE on the test dataval predictions = model.transform(test)

val evaluator = new RegressionEvaluator()

.setMetricName("rmse")

.setLabelCol("rating")

.setPredictionCol("prediction")

val rmse = evaluator.evaluate(predictions)

println(s"Root-mean-square error =$rmse")

Java:

import java.io.Serializable;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.function.Function;

import org.apache.spark.ml.evaluation.RegressionEvaluator;

import org.apache.spark.ml.recommendation.ALS;

import org.apache.spark.ml.recommendation.ALSModel;

public static class Rating implements Serializable {

private int userId;

private int movieId;

private float rating;

private long timestamp;

public Rating() {}

public Rating(int userId, int movieId, float rating, long timestamp) {

this.userId = userId;

this.movieId = movieId;

this.rating = rating;

this.timestamp = timestamp;

}

public int getUserId() {

return userId;

}

public int getMovieId() {

return movieId;

}

public float getRating() {

return rating;

}

public long getTimestamp() {

return timestamp;

}

public static Rating parseRating(String str) {

String[] fields = str.split("::");

if (fields.length != 4) {

throw new IllegalArgumentException("Each line must contain 4 fields");

}

int userId = Integer.parseInt(fields[0]);

int movieId = Integer.parseInt(fields[1]);

float rating = Float.parseFloat(fields[2]);

long timestamp = Long.parseLong(fields[3]);

return new Rating(userId, movieId, rating, timestamp);

}

}

JavaRDD ratingsRDD = spark

.read().textFile("data/mllib/als/sample_movielens_ratings.txt").javaRDD()

.map(new Function() {

public Rating call(String str) {

return Rating.parseRating(str);

}

});

Dataset ratings = spark.createDataFrame(ratingsRDD, Rating.class);

Dataset[] splits = ratings.randomSplit(new double[]{0.8, 0.2});

Dataset training = splits[0];

Dataset test = splits[1];

// Build the recommendation model using ALS on the training dataALS als = new ALS()

.setMaxIter(5)

.setRegParam(0.01)

.setUserCol("userId")

.setItemCol("movieId")

.setRatingCol("rating");

ALSModel model = als.fit(training);

// Evaluate the model by computing the RMSE on the test dataDataset predictions = model.transform(test);

RegressionEvaluator evaluator = new RegressionEvaluator()

.setMetricName("rmse")

.setLabelCol("rating")

.setPredictionCol("prediction");

Double rmse = evaluator.evaluate(predictions);

System.out.println("Root-mean-square error = " + rmse);

Python：

from pyspark.ml.evaluation import RegressionEvaluator

from pyspark.ml.recommendation import ALS

from pyspark.sql import Row

lines = spark.read.text("data/mllib/als/sample_movielens_ratings.txt").rdd

parts = lines.map(lambda row: row.value.split("::"))

ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),

rating=float(p[2]), timestamp=long(p[3])))

ratings = spark.createDataFrame(ratingsRDD)

(training, test) = ratings.randomSplit([0.8, 0.2])

# Build the recommendation model using ALS on the training data

als = ALS(maxIter=5, regParam=0.01, userCol="userId", itemCol="movieId", ratingCol="rating")

model = als.fit(training)

# Evaluate the model by computing the RMSE on the test data

predictions = model.transform(test)

evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",

predictionCol="prediction")

rmse = evaluator.evaluate(predictions)

print("Root-mean-square error = " + str(rmse))

總結(jié)

以上是生活随笔為你收集整理的mllib协同过滤 java实现_协同过滤(ALS)算法介绍及Spark MLlib调用实例(Scala/Java/Python)...的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。