评估指标 - 基于RDD的API

from pyspark.mllib.classification import LogisticRegressionWithLBFGS
from pyspark.mllib.evaluation import BinaryClassificationMetrics
from pyspark.mllib.util import MLUtils
# 当前在scala中可用的几个方法在pyspark中缺失
# 以LIBSVM格式加载训练数据
data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_binary_classification_data.txt")
# 将数据拆分为训练集（60%）和测试集（40%）
training, test = data.randomSplit([0.6, 0.4], seed=11)
training.cache()
# 运行训练算法以构建模型
model = LogisticRegressionWithLBFGS.train(training)
# 计算测试集上的原始分数
predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
# 实例化指标对象
metrics = BinaryClassificationMetrics(predictionAndLabels)
# 精确率-召回率曲线下面积
print("PR下面积 = %s" % metrics.areaUnderPR)
# ROC曲线下面积
print("ROC下面积 = %s" % metrics.areaUnderROC)

import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.util.MLUtils
// 以 LIBSVM 格式加载训练数据
val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_binary_classification_data.txt")
// 将数据拆分为训练集（60%）和测试集（40%）
val Array(training, test) = data.randomSplit(Array(0.6, 0.4), seed = 11L)
training.cache()
// 运行训练算法来构建模型
val model = new LogisticRegressionWithLBFGS()
.setNumClasses(2)
.run(training)
// 清除预测阈值，以便模型将返回概率
model.clearThreshold
// 在测试集上计算原始分数
val predictionAndLabels = test.map { case LabeledPoint(label, features) =>
val prediction = model.predict(features)
(prediction, label)
}
// 实例化指标对象
val metrics = new BinaryClassificationMetrics(predictionAndLabels)
// 按阈值计算精确度
val precision = metrics.precisionByThreshold
precision.collect.foreach { case (t, p) =>
println(s"阈值: $t, 精确度: $p")
}
// 按阈值计算召回率
val recall = metrics.recallByThreshold
recall.collect.foreach { case (t, r) =>
println(s"阈值: $t, 召回率: $r")
}
// 精确度-召回率曲线
val PRC = metrics.pr
// F-measure
val f1Score = metrics.fMeasureByThreshold
f1Score.collect.foreach { case (t, f) =>
println(s"阈值: $t, F-score: $f, Beta = 1")
}
val beta = 0.5
val fScore = metrics.fMeasureByThreshold(beta)
fScore.collect.foreach { case (t, f) =>
println(s"阈值: $t, F-score: $f, Beta = 0.5")
}
// AUPRC
val auPRC = metrics.areaUnderPR
println(s"精确率-召回率曲线下的面积 = $auPRC")
// 计算 ROC 和 PR 曲线中使用的阈值
val thresholds = precision.map(_._1)
// ROC 曲线
val roc = metrics.roc
// AUROC
val auROC = metrics.areaUnderROC
println(s"ROC 曲线下的面积 = $auROC")

import scala.Tuple2;
import org.apache.spark.api.java.*;
import org.apache.spark.mllib.classification.LogisticRegressionModel;
import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS;
import org.apache.spark.mllib.evaluation.BinaryClassificationMetrics;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;
String path = "data/mllib/sample_binary_classification_data.txt";
JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(sc, path).toJavaRDD();
// 将初始 RDD 分成两部分... [60% 训练数据, 40% 测试数据].
JavaRDD<LabeledPoint>[] splits =
data.randomSplit(new double[]{0.6, 0.4}, 11L);
JavaRDD<LabeledPoint> training = splits[0].cache();
JavaRDD<LabeledPoint> test = splits[1];
// 运行训练算法以构建模型.
LogisticRegressionModel model = new LogisticRegressionWithLBFGS()
.setNumClasses(2)
.run(training.rdd());
// 清除预测阈值，以便模型将返回概率
model.clearThreshold();
// 在测试集上计算原始分数.
JavaPairRDD<Object, Object> predictionAndLabels = test.mapToPair(p ->
new Tuple2<>(model.predict(p.features()), p.label()));
// 获取评估指标.
BinaryClassificationMetrics metrics =
new BinaryClassificationMetrics(predictionAndLabels.rdd());
// 根据阈值的精确度
JavaRDD<Tuple2<Object, Object>> precision = metrics.precisionByThreshold().toJavaRDD();
System.out.println("根据阈值的精确度: " + precision.collect());
// 根据阈值的召回率
JavaRDD recall = metrics.recallByThreshold().toJavaRDD();
System.out.println("根据阈值的召回率: " + recall.collect());
// 根据阈值的 F 分数
JavaRDD f1Score = metrics.fMeasureByThreshold().toJavaRDD();
System.out.println("根据阈值的 F1 分数: " + f1Score.collect());
JavaRDD f2Score = metrics.fMeasureByThreshold(2.0).toJavaRDD();
System.out.println("根据阈值的 F2 分数: " + f2Score.collect());
// 精确度-召回率曲线
JavaRDD prc = metrics.pr().toJavaRDD();
System.out.println("精确度-召回率曲线: " + prc.collect());
// 阈值
JavaRDD<Double> thresholds = precision.map(t -> Double.parseDouble(t._1().toString()));
// ROC 曲线
JavaRDD roc = metrics.roc().toJavaRDD();
System.out.println("ROC 曲线: " + roc.collect());
// AUPRC
System.out.println("精确度-召回率曲线下面积 = " + metrics.areaUnderPR());
// AUROC
System.out.println("ROC 曲线下面积 = " + metrics.areaUnderROC());
// 保存和加载模型
model.save(sc, "target/tmp/LogisticRegressionModel");
LogisticRegressionModel.load(sc, "target/tmp/LogisticRegressionModel");

from pyspark.mllib.classification import LogisticRegressionWithLBFGS
from pyspark.mllib.util import MLUtils
from pyspark.mllib.evaluation import MulticlassMetrics
# 以LIBSVM格式加载训练数据
data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_multiclass_classification_data.txt")
# 将数据分为训练集 (60%) 和测试集 (40%)
training, test = data.randomSplit([0.6, 0.4], seed=11)
training.cache()
# 运行训练算法以构建模型
model = LogisticRegressionWithLBFGS.train(training, numClasses=3)
# 计算测试集上的原始分数
predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
# 实例化指标对象
metrics = MulticlassMetrics(predictionAndLabels)
# 整体统计信息
precision = metrics.precision(1.0)
recall = metrics.recall(1.0)
f1Score = metrics.fMeasure(1.0)
print("Summary Stats")
print("Precision = %s" % precision)
print("Recall = %s" % recall)
print("F1 Score = %s" % f1Score)
# 按类别统计
labels = data.map(lambda lp: lp.label).distinct().collect()
for label in sorted(labels):
print("Class %s precision = %s" % (label, metrics.precision(label)))
print("Class %s recall = %s" % (label, metrics.recall(label)))
print("Class %s F1 Measure = %s" % (label, metrics.fMeasure(label, beta=1.0)))
# 加权统计
print("Weighted recall = %s" % metrics.weightedRecall)
print("Weighted precision = %s" % metrics.weightedPrecision)
print("Weighted F(1) Score = %s" % metrics.weightedFMeasure())
print("Weighted F(0.5) Score = %s" % metrics.weightedFMeasure(beta=0.5))
print("Weighted false positive rate = %s" % metrics.weightedFalsePositiveRate)

import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.util.MLUtils
// 以LIBSVM格式加载训练数据
val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_multiclass_classification_data.txt")
// 将数据分为训练集（60%）和测试集（40%）
val Array(training, test) = data.randomSplit(Array(0.6, 0.4), seed = 11L)
training.cache()
// 运行训练算法以构建模型
val model = new LogisticRegressionWithLBFGS()
.setNumClasses(3)
.run(training)
// 在测试集上计算原始分数
val predictionAndLabels = test.map { case LabeledPoint(label, features) =>
val prediction = model.predict(features)
(prediction, label)
}
// 实例化指标对象
val metrics = new MulticlassMetrics(predictionAndLabels)
// 混淆矩阵
println("混淆矩阵:")
println(metrics.confusionMatrix)
// 总体统计
val accuracy = metrics.accuracy
println("摘要统计")
println(s"准确率 = $accuracy")
// 按标签的精确度
val labels = metrics.labels
labels.foreach { l =>
println(s"精确度($l) = " + metrics.precision(l))
}
// 按标签的召回率
labels.foreach { l =>
println(s"召回率($l) = " + metrics.recall(l))
}
// 按标签的假阳性率
labels.foreach { l =>
println(s"假阳性率($l) = " + metrics.falsePositiveRate(l))
}
// 按标签的F值
labels.foreach { l =>
println(s"F1-分数($l) = " + metrics.fMeasure(l))
}
// 加权统计
println(s"加权精确度: ${metrics.weightedPrecision}")
println(s"加权召回率: ${metrics.weightedRecall}")
println(s"加权F1分数: ${metrics.weightedFMeasure}")
println(s"加权假阳性率: ${metrics.weightedFalsePositiveRate}")

import scala.Tuple2;
import org.apache.spark.api.java.*;
import org.apache.spark.mllib.classification.LogisticRegressionModel;
import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS;
import org.apache.spark.mllib.evaluation.MulticlassMetrics;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;
import org.apache.spark.mllib.linalg.Matrix;
String path = "data/mllib/sample_multiclass_classification_data.txt";
JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(sc, path).toJavaRDD();
// 将初始 RDD 分割成两个... [60% 训练数据，40% 测试数据]。
JavaRDD<LabeledPoint>[] splits = data.randomSplit(new double[]{0.6, 0.4}, 11L);
JavaRDD<LabeledPoint> training = splits[0].cache();
JavaRDD<LabeledPoint> test = splits[1];
// 运行训练算法来构建模型。
LogisticRegressionModel model = new LogisticRegressionWithLBFGS()
.setNumClasses(3)
.run(training.rdd());
// 计算测试集上的原始分数。
JavaPairRDD<Object, Object> predictionAndLabels = test.mapToPair(p ->
new Tuple2<>(model.predict(p.features()), p.label()));
// 获取评估指标。
MulticlassMetrics metrics = new MulticlassMetrics(predictionAndLabels.rdd());
// 混淆矩阵
Matrix confusion = metrics.confusionMatrix();
System.out.println("Confusion matrix: \n" + confusion);
// 总体统计
System.out.println("Accuracy = " + metrics.accuracy());
// 按标签统计
for (int i = 0; i < metrics.labels().length; i++) {
System.out.format("Class %f precision = %f\n", metrics.labels()[i],metrics.precision(
metrics.labels()[i]));
System.out.format("Class %f recall = %f\n", metrics.labels()[i], metrics.recall(
metrics.labels()[i]));
System.out.format("Class %f F1 score = %f\n", metrics.labels()[i], metrics.fMeasure(
metrics.labels()[i]));
}
// 加权统计
System.out.format("Weighted precision = %f\n", metrics.weightedPrecision());
System.out.format("Weighted recall = %f\n", metrics.weightedRecall());
System.out.format("Weighted F1 score = %f\n", metrics.weightedFMeasure());
System.out.format("Weighted false positive rate = %f\n", metrics.weightedFalsePositiveRate());
// 保存和加载模型
model.save(sc, "target/tmp/LogisticRegressionModel");
LogisticRegressionModel sameModel = LogisticRegressionModel.load(sc,
"target/tmp/LogisticRegressionModel");

from pyspark.mllib.evaluation import MultilabelMetrics
scoreAndLabels = sc.parallelize([
([0.0, 1.0], [0.0, 2.0]),
([0.0, 2.0], [0.0, 1.0]),
([], [0.0]),
([2.0], [2.0]),
([2.0, 0.0], [2.0, 0.0]),
([0.0, 1.0, 2.0], [0.0, 1.0]),
([1.0], [1.0, 2.0])])
# 实例化度量对象
metrics = MultilabelMetrics(scoreAndLabels)
# 概要统计
print("召回率 = %s" % metrics.recall())
print("精确度 = %s" % metrics.precision())
print("F1 值 = %s" % metrics.f1Measure())
print("准确率 = %s" % metrics.accuracy)
# 单个标签统计
labels = scoreAndLabels.flatMap(lambda x: x[1]).distinct().collect()
for label in labels:
print("类别 %s 精确度 = %s" % (label, metrics.precision(label)))
print("类别 %s 召回率 = %s" % (label, metrics.recall(label)))
print("类别 %s F1 值 = %s" % (label, metrics.f1Measure(label)))
# 微观统计
print("微观精确度 = %s" % metrics.microPrecision)
print("微观召回率 = %s" % metrics.microRecall)
print("微观 F1 值 = %s" % metrics.microF1Measure)
# 汉明损失
print("汉明损失 = %s" % metrics.hammingLoss)
# 子集准确率
print("子集准确率 = %s" % metrics.subsetAccuracy)

import org.apache.spark.mllib.evaluation.MultilabelMetrics
import org.apache.spark.rdd.RDD
val scoreAndLabels: RDD[(Array[Double], Array[Double])] = sc.parallelize(
Seq((Array(0.0, 1.0), Array(0.0, 2.0)),
(Array(0.0, 2.0), Array(0.0, 1.0)),
(Array.empty[Double], Array(0.0)),
(Array(2.0), Array(2.0)),
(Array(2.0, 0.0), Array(2.0, 0.0)),
(Array(0.0, 1.0, 2.0), Array(0.0, 1.0)),
(Array(1.0), Array(1.0, 2.0))), 2)
// 实例化指标对象
val metrics = new MultilabelMetrics(scoreAndLabels)
// 总结统计
println(s"召回率 = ${metrics.recall}")
println(s"精确率 = ${metrics.precision}")
println(s"F1 评分 = ${metrics.f1Measure}")
println(s"准确率 = ${metrics.accuracy}")
// 各个标签统计
metrics.labels.foreach(label =>
println(s"类 $label 精确率 = ${metrics.precision(label)}"))
metrics.labels.foreach(label => println(s"类 $label 召回率 = ${metrics.recall(label)}"))
metrics.labels.foreach(label => println(s"类 $label F1 分数 = ${metrics.f1Measure(label)}"))
// 微观统计
println(s"微观召回率 = ${metrics.microRecall}")
println(s"微观精确率 = ${metrics.microPrecision}")
println(s"微观 F1 评分 = ${metrics.microF1Measure}")
// 汉明损失
println(s"汉明损失 = ${metrics.hammingLoss}")
// 子集准确率
println(s"子集准确率 = ${metrics.subsetAccuracy}")

import java.util.Arrays;
import java.util.List;
import scala.Tuple2;
import org.apache.spark.api.java.*;
import org.apache.spark.mllib.evaluation.MultilabelMetrics;
import org.apache.spark.SparkConf;
List<Tuple2double[], double[]>> data = Arrays.asList(
new Tuple2<>(new double[]{0.0, 1.0}, new double[]{0.0, 2.0}),
new Tuple2<>(new double[]{0.0, 2.0}, new double[]{0.0, 1.0}),
new Tuple2<>(new double[]{}, new double[]{0.0}),
new Tuple2<>(new double[]{2.0}, new double[]{2.0}),
new Tuple2<>(new double[]{2.0, 0.0}, new double[]{2.0, 0.0}),
new Tuple2<>(new double[]{0.0, 1.0, 2.0}, new double[]{0.0, 1.0}),
new Tuple2<>(new double[]{1.0}, new double[]{1.0, 2.0})
);
JavaRDD<Tuple2<double[], double[]>> scoreAndLabels = sc.parallelize(data);
// 实例化指标对象
MultilabelMetrics metrics = new MultilabelMetrics(scoreAndLabels.rdd());
// 汇总统计
System.out.format("召回率 = %f\n", metrics.recall());
System.out.format("精确度 = %f\n", metrics.precision());
System.out.format("F1 值 = %f\n", metrics.f1Measure());
System.out.format("准确率 = %f\n", metrics.accuracy());
// 按标签统计
for (int i = 0; i < metrics.labels().length - 1; i++) {
System.out.format("类别 %1.1f 精确度 = %f\n", metrics.labels()[i], metrics.precision(
metrics.labels()[i]));
System.out.format("类别 %1.1f 召回率 = %f\n", metrics.labels()[i], metrics.recall(
metrics.labels()[i]));
System.out.format("类别 %1.1f F1 分数 = %f\n", metrics.labels()[i], metrics.f1Measure(
metrics.labels()[i]));
}
// 微观统计
System.out.format("微观召回率 = %f\n", metrics.microRecall());
System.out.format("微观精确度 = %f\n", metrics.microPrecision());
System.out.format("微观 F1 值 = %f\n", metrics.microF1Measure());
// 汉明损失
System.out.format("汉明损失 = %f\n", metrics.hammingLoss());
// 子集准确率
System.out.format("子集准确率 = %f\n", metrics.subsetAccuracy());

from pyspark.mllib.recommendation import ALS, Rating
from pyspark.mllib.evaluation import RegressionMetrics
# 读取评分数据
lines = sc.textFile("data/mllib/sample_movielens_data.txt")
def parseLine(line):
fields = line.split("::")
return Rating(int(fields[0]), int(fields[1]), float(fields[2]) - 2.5)
ratings = lines.map(lambda r: parseLine(r))
# 训练模型以预测用户-产品评分
model = ALS.train(ratings, 10, 10, 0.01)
# 对所有现有的用户-产品对获取预测评分
testData = ratings.map(lambda p: (p.user, p.product))
predictions = model.predictAll(testData).map(lambda r: ((r.user, r.product), r.rating))
ratingsTuple = ratings.map(lambda r: ((r.user, r.product), r.rating))
scoreAndLabels = predictions.join(ratingsTuple).map(lambda tup: tup[1])
# 实例化回归指标以比较预测和实际评分
metrics = RegressionMetrics(scoreAndLabels)
# 均方根误差
print("RMSE = %s" % metrics.rootMeanSquaredError)
# R平方
print("R-squared = %s" % metrics.r2)

import org.apache.spark.mllib.evaluation.{RankingMetrics, RegressionMetrics}
import org.apache.spark.mllib.recommendation.{ALS, Rating}
// 读取评分数据
val ratings = spark.read.textFile("data/mllib/sample_movielens_data.txt").rdd.map { line =>
val fields = line.split("::")
Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble - 2.5)
}.cache()
// 将评分映射为1或0，1表示应推荐的电影
val binarizedRatings = ratings.map(r => Rating(r.user, r.product,
if (r.rating > 0) 1.0 else 0.0)).cache()
// 总结评分
val numRatings = ratings.count()
val numUsers = ratings.map(_.user).distinct().count()
val numMovies = ratings.map(_.product).distinct().count()
println(s"从 $numUsers 用户获取了 $numRatings 条评分，共 $numMovies 部电影.")
// 构建模型
val numIterations = 10
val rank = 10
val lambda = 0.01
val model = ALS.train(ratings, rank, numIterations, lambda)
// 定义一个函数，将评分缩放到0到1之间
def scaledRating(r: Rating): Rating = {
val scaledRating = math.max(math.min(r.rating, 1.0), 0.0)
Rating(r.user, r.product, scaledRating)
}
// 获取每个用户的前十个推荐并将其缩放至[0, 1]
val userRecommended = model.recommendProductsForUsers(10).map { case (user, recs) =>
(user, recs.map(scaledRating))
}
// 假设用户评分3或更高的电影（映射为1）是相关文档
// 和前十个最相关的文档进行比较
val userMovies = binarizedRatings.groupBy(_.user)
val relevantDocuments = userMovies.join(userRecommended).map { case (user, (actual,
predictions)) =>
(predictions.map(_.product), actual.filter(_.rating > 0.0).map(_.product).toArray)
}
// 实例化指标对象
val metrics = new RankingMetrics(relevantDocuments)
// K的精度
Array(1, 3, 5).foreach { k =>
println(s"在 $k 时的精度 = ${metrics.precisionAt(k)}")
}
// 平均精度
println(s"平均精度 = ${metrics.meanAveragePrecision}")
// K的平均精度
println(s"在2时的平均精度 = ${metrics.meanAveragePrecisionAt(2)}")
// 规范化折扣累计增益
Array(1, 3, 5).foreach { k =>
println(s"在 $k 时的NDCG = ${metrics.ndcgAt(k)}")
}
// K的召回率
Array(1, 3, 5).foreach { k =>
println(s"在 $k 时的召回率 = ${metrics.recallAt(k)}")
}
// 获取每个数据点的预测
val allPredictions = model.predict(ratings.map(r => (r.user, r.product))).map(r => ((r.user,
r.product), r.rating))
val allRatings = ratings.map(r => ((r.user, r.product), r.rating))
val predictionsAndLabels = allPredictions.join(allRatings).map { case ((user, product),
(predicted, actual)) =>
(predicted, actual)
}
// 使用回归指标获取RMSE
val regressionMetrics = new RegressionMetrics(predictionsAndLabels)
println(s"RMSE = ${regressionMetrics.rootMeanSquaredError}")
// R平方
println(s"R平方 = ${regressionMetrics.r2}")

import java.util.*;
import scala.Tuple2;
import org.apache.spark.api.java.*;
import org.apache.spark.mllib.evaluation.RegressionMetrics;
import org.apache.spark.mllib.evaluation.RankingMetrics;
import org.apache.spark.mllib.recommendation.ALS;
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
import org.apache.spark.mllib.recommendation.Rating;
String path = "data/mllib/sample_movielens_data.txt";
JavaRDD<String> data = sc.textFile(path);
JavaRDD<Rating> ratings = data.map(line -> {
String[] parts = line.split("::");
return new Rating(Integer.parseInt(parts[0]), Integer.parseInt(parts[1]), Double
.parseDouble(parts[2]) - 2.5);
});
ratings.cache();
// 训练一个 ALS 模型
MatrixFactorizationModel model = ALS.train(JavaRDD.toRDD(ratings), 10, 10, 0.01);
// 为每个用户获取前 10 个推荐并将评分从 0 缩放到 1
JavaRDD<Tuple2<Object, Rating[]>> userRecs = model.recommendProductsForUsers(10).toJavaRDD();
JavaRDD<Tuple2<Object, Rating[]>> userRecsScaled = userRecs.map(t -> {
Rating[] scaledRatings = new Rating[t._2().length];
for (int i = 0; i < scaledRatings.length; i++) {
double newRating = Math.max(Math.min(t._2()[i].rating(), 1.0), 0.0);
scaledRatings[i] = new Rating(t._2()[i].user(), t._2()[i].product(), newRating);
}
return new Tuple2<>(t._1(), scaledRatings);
});
JavaPairRDD<Object, Rating[]> userRecommended = JavaPairRDD.fromJavaRDD(userRecsScaled);
// 将评分映射为 1 或 0，1 表示应该推荐的电影
JavaRDD<Rating> binarizedRatings = ratings.map(r -> {
double binaryRating;
if (r.rating() > 0.0) {
binaryRating = 1.0;
} else {
binaryRating = 0.0;
}
return new Rating(r.user(), r.product(), binaryRating);
});
// 按共同用户分组评分
JavaPairRDD<Object, Iterable<Rating>> userMovies = binarizedRatings.groupBy(Rating::user);
// 从所有用户评分中获取真实相关文档
JavaPairRDD<Object, ListInteger>> userMoviesList = userMovies.mapValues(docs -> {
List<Integer> products = new ArrayList<>();
for (Rating r : docs) {
if (r.rating() > 0.0) {
products.add(r.product());
}
}
return products;
});
// 从每个推荐中提取产品 ID
JavaPairRDD<Object, List<Integer>> userRecommendedList = userRecommended.mapValues(docs -> {
List<Integer> products = new ArrayList<>();
for (Rating r : docs) {
products.add(r.product());
}
return products;
});
JavaRDD<Tuple2<List<Integer>, List<Integer>>> relevantDocs = userMoviesList.join(
userRecommendedList).values();
// 实例化度量对象
RankingMetrics<Integer> metrics = RankingMetrics.of(relevantDocs);
// 精确度，NDCG 和召回率在 k
Integer[] kVector = {1, 3, 5};
for (Integer k : kVector) {
System.out.format("Precision at %d = %f\n", k, metrics.precisionAt(k));
System.out.format("NDCG at %d = %f\n", k, metrics.ndcgAt(k));
System.out.format("Recall at %d = %f\n", k, metrics.recallAt(k));
}
// 平均精度
System.out.format("Mean average precision = %f\n", metrics.meanAveragePrecision());
// k 的平均精度
System.out.format("Mean average precision at 2 = %f\n", metrics.meanAveragePrecisionAt(2));
// 使用数值评分和回归度量评估模型
JavaRDD<Tuple2<Object, Object>> userProducts =
ratings.map(r -> new Tuple2<>(r.user(), r.product()));
JavaPairRDD<Tuple2<Integer, Integer>, Object> predictions = JavaPairRDD.fromJavaRDD(
model.predict(JavaRDD.toRDD(userProducts)).toJavaRDD().map(r ->
new Tuple2<>(new Tuple2<>(r.user(), r.product()), r.rating())));
JavaRDD<Tuple2<Object, Object>> ratesAndPreds =
JavaPairRDD.fromJavaRDD(ratings.map(r ->
new Tuple2<Tuple2<Integer, Integer>, Object>(
new Tuple2<>(r.user(), r.product()),
r.rating())
)).join(predictions).values();
// 创建回归度量对象
RegressionMetrics regressionMetrics = new RegressionMetrics(ratesAndPreds.rdd());
// 均方根误差
System.out.format("RMSE = %f\n", regressionMetrics.rootMeanSquaredError());
// R 平方
System.out.format("R-squared = %f\n", regressionMetrics.r2());