協同過濾

算法介紹：

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

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

顯式與隱式反饋

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

正則化參數

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

參數：

alpha:

類型：雙精度型。

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

checkpointInterval:

類型：整數型。

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

implicitPrefs:

類型：布爾型。

含義：特征列名。

itemCol:

類型：字符串型。

含義：商品編號列名。

maxIter:

類型：整數型。

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

nonnegative:

類型：布爾型。

含義：是否需要非負約束。

numItemBlocks:

類型：整數型。

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

numUserBlocks:

類型：整數型。

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

predictionCol:

類型：字符串型。

含義：預測結果列名。

rank:

類型：整數型。

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

ratingCol:

類型：字符串型。

含義：評分列名。

regParam:

類型：雙精度型。

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

seed:

類型：長整型。

含義：隨機種子。

userCol:

類型：字符串型。

含義：用戶列名。

調用示例：

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

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))

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。