spark.locality.wait |
3s |
diff --git a/docs/graphx-programming-guide.md b/docs/graphx-programming-guide.md
index 3f10cb2dc3d2a..99f8c827f767f 100644
--- a/docs/graphx-programming-guide.md
+++ b/docs/graphx-programming-guide.md
@@ -800,7 +800,7 @@ import org.apache.spark.graphx._
// Import random graph generation library
import org.apache.spark.graphx.util.GraphGenerators
// A graph with edge attributes containing distances
-val graph: Graph[Int, Double] =
+val graph: Graph[Long, Double] =
GraphGenerators.logNormalGraph(sc, numVertices = 100).mapEdges(e => e.attr.toDouble)
val sourceId: VertexId = 42 // The ultimate source
// Initialize the graph such that all vertices except the root have distance infinity.
diff --git a/docs/ml-guide.md b/docs/ml-guide.md
index 8c46adf256a9a..b6ca50e98db02 100644
--- a/docs/ml-guide.md
+++ b/docs/ml-guide.md
@@ -561,7 +561,7 @@ test = sc.parallelize([(4L, "spark i j k"),
prediction = model.transform(test)
selected = prediction.select("id", "text", "prediction")
for row in selected.collect():
- print row
+ print(row)
sc.stop()
{% endhighlight %}
diff --git a/docs/mllib-evaluation-metrics.md b/docs/mllib-evaluation-metrics.md
new file mode 100644
index 0000000000000..7066d5c97418c
--- /dev/null
+++ b/docs/mllib-evaluation-metrics.md
@@ -0,0 +1,1497 @@
+---
+layout: global
+title: Evaluation Metrics - MLlib
+displayTitle: MLlib - Evaluation Metrics
+---
+
+* Table of contents
+{:toc}
+
+Spark's MLlib comes with a number of machine learning algorithms that can be used to learn from and make predictions
+on data. When these algorithms are applied to build machine learning models, there is a need to evaluate the performance
+of the model on some criteria, which depends on the application and its requirements. Spark's MLlib also provides a
+suite of metrics for the purpose of evaluating the performance of machine learning models.
+
+Specific machine learning algorithms fall under broader types of machine learning applications like classification,
+regression, clustering, etc. Each of these types have well established metrics for performance evaluation and those
+metrics that are currently available in Spark's MLlib are detailed in this section.
+
+## Classification model evaluation
+
+While there are many different types of classification algorithms, the evaluation of classification models all share
+similar principles. In a [supervised classification problem](https://en.wikipedia.org/wiki/Statistical_classification),
+there exists a true output and a model-generated predicted output for each data point. For this reason, the results for
+each data point can be assigned to one of four categories:
+
+* True Positive (TP) - label is positive and prediction is also positive
+* True Negative (TN) - label is negative and prediction is also negative
+* False Positive (FP) - label is negative but prediction is positive
+* False Negative (FN) - label is positive but prediction is negative
+
+These four numbers are the building blocks for most classifier evaluation metrics. A fundamental point when considering
+classifier evaluation is that pure accuracy (i.e. was the prediction correct or incorrect) is not generally a good metric. The
+reason for this is because a dataset may be highly unbalanced. For example, if a model is designed to predict fraud from
+a dataset where 95% of the data points are _not fraud_ and 5% of the data points are _fraud_, then a naive classifier
+that predicts _not fraud_, regardless of input, will be 95% accurate. For this reason, metrics like
+[precision and recall](https://en.wikipedia.org/wiki/Precision_and_recall) are typically used because they take into
+account the *type* of error. In most applications there is some desired balance between precision and recall, which can
+be captured by combining the two into a single metric, called the [F-measure](https://en.wikipedia.org/wiki/F1_score).
+
+### Binary classification
+
+[Binary classifiers](https://en.wikipedia.org/wiki/Binary_classification) are used to separate the elements of a given
+dataset into one of two possible groups (e.g. fraud or not fraud) and is a special case of multiclass classification.
+Most binary classification metrics can be generalized to multiclass classification metrics.
+
+#### Threshold tuning
+
+It is import to understand that many classification models actually output a "score" (often times a probability) for
+each class, where a higher score indicates higher likelihood. In the binary case, the model may output a probability for
+each class: $P(Y=1|X)$ and $P(Y=0|X)$. Instead of simply taking the higher probability, there may be some cases where
+the model might need to be tuned so that it only predicts a class when the probability is very high (e.g. only block a
+credit card transaction if the model predicts fraud with >90% probability). Therefore, there is a prediction *threshold*
+which determines what the predicted class will be based on the probabilities that the model outputs.
+
+Tuning the prediction threshold will change the precision and recall of the model and is an important part of model
+optimization. In order to visualize how precision, recall, and other metrics change as a function of the threshold it is
+common practice to plot competing metrics against one another, parameterized by threshold. A P-R curve plots (precision,
+recall) points for different threshold values, while a
+[receiver operating characteristic](https://en.wikipedia.org/wiki/Receiver_operating_characteristic), or ROC, curve
+plots (recall, false positive rate) points.
+
+**Available metrics**
+
+
+
+ | Metric | Definition |
|---|
+
+
+
+ | Precision (Postive Predictive Value) |
+ $PPV=\frac{TP}{TP + FP}$ |
+
+
+ | Recall (True Positive Rate) |
+ $TPR=\frac{TP}{P}=\frac{TP}{TP + FN}$ |
+
+
+ | F-measure |
+ $F(\beta) = \left(1 + \beta^2\right) \cdot \left(\frac{PPV \cdot TPR}
+ {\beta^2 \cdot PPV + TPR}\right)$ |
+
+
+ | Receiver Operating Characteristic (ROC) |
+ $FPR(T)=\int^\infty_{T} P_0(T)\,dT \\ TPR(T)=\int^\infty_{T} P_1(T)\,dT$ |
+
+
+ | Area Under ROC Curve |
+ $AUROC=\int^1_{0} \frac{TP}{P} d\left(\frac{FP}{N}\right)$ |
+
+
+ | Area Under Precision-Recall Curve |
+ $AUPRC=\int^1_{0} \frac{TP}{TP+FP} d\left(\frac{TP}{P}\right)$ |
+
+
+
+
+
+**Examples**
+
+
+The following code snippets illustrate how to load a sample dataset, train a binary classification algorithm on the
+data, and evaluate the performance of the algorithm by several binary evaluation metrics.
+
+
+
+{% highlight scala %}
+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
+
+// Load training data in LIBSVM format
+val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_binary_classification_data.txt")
+
+// Split data into training (60%) and test (40%)
+val Array(training, test) = data.randomSplit(Array(0.6, 0.4), seed = 11L)
+training.cache()
+
+// Run training algorithm to build the model
+val model = new LogisticRegressionWithLBFGS()
+ .setNumClasses(2)
+ .run(training)
+
+// Clear the prediction threshold so the model will return probabilities
+model.clearThreshold
+
+// Compute raw scores on the test set
+val predictionAndLabels = test.map { case LabeledPoint(label, features) =>
+ val prediction = model.predict(features)
+ (prediction, label)
+}
+
+// Instantiate metrics object
+val metrics = new BinaryClassificationMetrics(predictionAndLabels)
+
+// Precision by threshold
+val precision = metrics.precisionByThreshold
+precision.foreach { case (t, p) =>
+ println(s"Threshold: $t, Precision: $p")
+}
+
+// Recall by threshold
+val recall = metrics.precisionByThreshold
+recall.foreach { case (t, r) =>
+ println(s"Threshold: $t, Recall: $r")
+}
+
+// Precision-Recall Curve
+val PRC = metrics.pr
+
+// F-measure
+val f1Score = metrics.fMeasureByThreshold
+f1Score.foreach { case (t, f) =>
+ println(s"Threshold: $t, F-score: $f, Beta = 1")
+}
+
+val beta = 0.5
+val fScore = metrics.fMeasureByThreshold(beta)
+f1Score.foreach { case (t, f) =>
+ println(s"Threshold: $t, F-score: $f, Beta = 0.5")
+}
+
+// AUPRC
+val auPRC = metrics.areaUnderPR
+println("Area under precision-recall curve = " + auPRC)
+
+// Compute thresholds used in ROC and PR curves
+val thresholds = precision.map(_._1)
+
+// ROC Curve
+val roc = metrics.roc
+
+// AUROC
+val auROC = metrics.areaUnderROC
+println("Area under ROC = " + auROC)
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight java %}
+import scala.Tuple2;
+
+import org.apache.spark.api.java.*;
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.api.java.function.Function;
+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;
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkContext;
+
+public class BinaryClassification {
+ public static void main(String[] args) {
+ SparkConf conf = new SparkConf().setAppName("Binary Classification Metrics");
+ SparkContext sc = new SparkContext(conf);
+ String path = "data/mllib/sample_binary_classification_data.txt";
+ JavaRDD data = MLUtils.loadLibSVMFile(sc, path).toJavaRDD();
+
+ // Split initial RDD into two... [60% training data, 40% testing data].
+ JavaRDD[] splits = data.randomSplit(new double[] {0.6, 0.4}, 11L);
+ JavaRDD training = splits[0].cache();
+ JavaRDD test = splits[1];
+
+ // Run training algorithm to build the model.
+ final LogisticRegressionModel model = new LogisticRegressionWithLBFGS()
+ .setNumClasses(2)
+ .run(training.rdd());
+
+ // Clear the prediction threshold so the model will return probabilities
+ model.clearThreshold();
+
+ // Compute raw scores on the test set.
+ JavaRDD> predictionAndLabels = test.map(
+ new Function>() {
+ public Tuple2 call(LabeledPoint p) {
+ Double prediction = model.predict(p.features());
+ return new Tuple2(prediction, p.label());
+ }
+ }
+ );
+
+ // Get evaluation metrics.
+ BinaryClassificationMetrics metrics = new BinaryClassificationMetrics(predictionAndLabels.rdd());
+
+ // Precision by threshold
+ JavaRDD> precision = metrics.precisionByThreshold().toJavaRDD();
+ System.out.println("Precision by threshold: " + precision.toArray());
+
+ // Recall by threshold
+ JavaRDD> recall = metrics.recallByThreshold().toJavaRDD();
+ System.out.println("Recall by threshold: " + recall.toArray());
+
+ // F Score by threshold
+ JavaRDD> f1Score = metrics.fMeasureByThreshold().toJavaRDD();
+ System.out.println("F1 Score by threshold: " + f1Score.toArray());
+
+ JavaRDD> f2Score = metrics.fMeasureByThreshold(2.0).toJavaRDD();
+ System.out.println("F2 Score by threshold: " + f2Score.toArray());
+
+ // Precision-recall curve
+ JavaRDD> prc = metrics.pr().toJavaRDD();
+ System.out.println("Precision-recall curve: " + prc.toArray());
+
+ // Thresholds
+ JavaRDD thresholds = precision.map(
+ new Function, Double>() {
+ public Double call (Tuple2 t) {
+ return new Double(t._1().toString());
+ }
+ }
+ );
+
+ // ROC Curve
+ JavaRDD> roc = metrics.roc().toJavaRDD();
+ System.out.println("ROC curve: " + roc.toArray());
+
+ // AUPRC
+ System.out.println("Area under precision-recall curve = " + metrics.areaUnderPR());
+
+ // AUROC
+ System.out.println("Area under ROC = " + metrics.areaUnderROC());
+
+ // Save and load model
+ model.save(sc, "myModelPath");
+ LogisticRegressionModel sameModel = LogisticRegressionModel.load(sc, "myModelPath");
+ }
+}
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight python %}
+from pyspark.mllib.classification import LogisticRegressionWithLBFGS
+from pyspark.mllib.evaluation import BinaryClassificationMetrics
+from pyspark.mllib.regression import LabeledPoint
+from pyspark.mllib.util import MLUtils
+
+# Several of the methods available in scala are currently missing from pyspark
+
+# Load training data in LIBSVM format
+data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_binary_classification_data.txt")
+
+# Split data into training (60%) and test (40%)
+training, test = data.randomSplit([0.6, 0.4], seed = 11L)
+training.cache()
+
+# Run training algorithm to build the model
+model = LogisticRegressionWithLBFGS.train(training)
+
+# Compute raw scores on the test set
+predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
+
+# Instantiate metrics object
+metrics = BinaryClassificationMetrics(predictionAndLabels)
+
+# Area under precision-recall curve
+print("Area under PR = %s" % metrics.areaUnderPR)
+
+# Area under ROC curve
+print("Area under ROC = %s" % metrics.areaUnderROC)
+
+{% endhighlight %}
+
+
+
+
+ | Metric | Definition |
|---|
+
+
+
+ | Confusion Matrix |
+
+ $C_{ij} = \sum_{k=0}^{N-1} \hat{\delta}(\mathbf{y}_k-\ell_i) \cdot \hat{\delta}(\hat{\mathbf{y}}_k - \ell_j)\\ \\
+ \left( \begin{array}{ccc}
+ \sum_{k=0}^{N-1} \hat{\delta}(\mathbf{y}_k-\ell_1) \cdot \hat{\delta}(\hat{\mathbf{y}}_k - \ell_1) & \ldots &
+ \sum_{k=0}^{N-1} \hat{\delta}(\mathbf{y}_k-\ell_1) \cdot \hat{\delta}(\hat{\mathbf{y}}_k - \ell_N) \\
+ \vdots & \ddots & \vdots \\
+ \sum_{k=0}^{N-1} \hat{\delta}(\mathbf{y}_k-\ell_N) \cdot \hat{\delta}(\hat{\mathbf{y}}_k - \ell_1) & \ldots &
+ \sum_{k=0}^{N-1} \hat{\delta}(\mathbf{y}_k-\ell_N) \cdot \hat{\delta}(\hat{\mathbf{y}}_k - \ell_N)
+ \end{array} \right)$
+ |
+
+
+ | Overall Precision |
+ $PPV = \frac{TP}{TP + FP} = \frac{1}{N}\sum_{i=0}^{N-1} \hat{\delta}\left(\hat{\mathbf{y}}_i -
+ \mathbf{y}_i\right)$ |
+
+
+ | Overall Recall |
+ $TPR = \frac{TP}{TP + FN} = \frac{1}{N}\sum_{i=0}^{N-1} \hat{\delta}\left(\hat{\mathbf{y}}_i -
+ \mathbf{y}_i\right)$ |
+
+
+ | Overall F1-measure |
+ $F1 = 2 \cdot \left(\frac{PPV \cdot TPR}
+ {PPV + TPR}\right)$ |
+
+
+ | Precision by label |
+ $PPV(\ell) = \frac{TP}{TP + FP} =
+ \frac{\sum_{i=0}^{N-1} \hat{\delta}(\hat{\mathbf{y}}_i - \ell) \cdot \hat{\delta}(\mathbf{y}_i - \ell)}
+ {\sum_{i=0}^{N-1} \hat{\delta}(\hat{\mathbf{y}}_i - \ell)}$ |
+
+
+ | Recall by label |
+ $TPR(\ell)=\frac{TP}{P} =
+ \frac{\sum_{i=0}^{N-1} \hat{\delta}(\hat{\mathbf{y}}_i - \ell) \cdot \hat{\delta}(\mathbf{y}_i - \ell)}
+ {\sum_{i=0}^{N-1} \hat{\delta}(\mathbf{y}_i - \ell)}$ |
+
+
+ | F-measure by label |
+ $F(\beta, \ell) = \left(1 + \beta^2\right) \cdot \left(\frac{PPV(\ell) \cdot TPR(\ell)}
+ {\beta^2 \cdot PPV(\ell) + TPR(\ell)}\right)$ |
+
+
+ | Weighted precision |
+ $PPV_{w}= \frac{1}{N} \sum\nolimits_{\ell \in L} PPV(\ell)
+ \cdot \sum_{i=0}^{N-1} \hat{\delta}(\mathbf{y}_i-\ell)$ |
+
+
+ | Weighted recall |
+ $TPR_{w}= \frac{1}{N} \sum\nolimits_{\ell \in L} TPR(\ell)
+ \cdot \sum_{i=0}^{N-1} \hat{\delta}(\mathbf{y}_i-\ell)$ |
+
+
+ | Weighted F-measure |
+ $F_{w}(\beta)= \frac{1}{N} \sum\nolimits_{\ell \in L} F(\beta, \ell)
+ \cdot \sum_{i=0}^{N-1} \hat{\delta}(\mathbf{y}_i-\ell)$ |
+
+
+
+
+**Examples**
+
+
+The following code snippets illustrate how to load a sample dataset, train a multiclass classification algorithm on
+the data, and evaluate the performance of the algorithm by several multiclass classification evaluation metrics.
+
+
+
+{% highlight scala %}
+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
+
+// Load training data in LIBSVM format
+val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_multiclass_classification_data.txt")
+
+// Split data into training (60%) and test (40%)
+val Array(training, test) = data.randomSplit(Array(0.6, 0.4), seed = 11L)
+training.cache()
+
+// Run training algorithm to build the model
+val model = new LogisticRegressionWithLBFGS()
+ .setNumClasses(3)
+ .run(training)
+
+// Compute raw scores on the test set
+val predictionAndLabels = test.map { case LabeledPoint(label, features) =>
+ val prediction = model.predict(features)
+ (prediction, label)
+}
+
+// Instantiate metrics object
+val metrics = new MulticlassMetrics(predictionAndLabels)
+
+// Confusion matrix
+println("Confusion matrix:")
+println(metrics.confusionMatrix)
+
+// Overall Statistics
+val precision = metrics.precision
+val recall = metrics.recall // same as true positive rate
+val f1Score = metrics.fMeasure
+println("Summary Statistics")
+println(s"Precision = $precision")
+println(s"Recall = $recall")
+println(s"F1 Score = $f1Score")
+
+// Precision by label
+val labels = metrics.labels
+labels.foreach { l =>
+ println(s"Precision($l) = " + metrics.precision(l))
+}
+
+// Recall by label
+labels.foreach { l =>
+ println(s"Recall($l) = " + metrics.recall(l))
+}
+
+// False positive rate by label
+labels.foreach { l =>
+ println(s"FPR($l) = " + metrics.falsePositiveRate(l))
+}
+
+// F-measure by label
+labels.foreach { l =>
+ println(s"F1-Score($l) = " + metrics.fMeasure(l))
+}
+
+// Weighted stats
+println(s"Weighted precision: ${metrics.weightedPrecision}")
+println(s"Weighted recall: ${metrics.weightedRecall}")
+println(s"Weighted F1 score: ${metrics.weightedFMeasure}")
+println(s"Weighted false positive rate: ${metrics.weightedFalsePositiveRate}")
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight java %}
+import scala.Tuple2;
+
+import org.apache.spark.api.java.*;
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.api.java.function.Function;
+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;
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkContext;
+
+public class MulticlassClassification {
+ public static void main(String[] args) {
+ SparkConf conf = new SparkConf().setAppName("Multiclass Classification Metrics");
+ SparkContext sc = new SparkContext(conf);
+ String path = "data/mllib/sample_multiclass_classification_data.txt";
+ JavaRDD data = MLUtils.loadLibSVMFile(sc, path).toJavaRDD();
+
+ // Split initial RDD into two... [60% training data, 40% testing data].
+ JavaRDD[] splits = data.randomSplit(new double[] {0.6, 0.4}, 11L);
+ JavaRDD training = splits[0].cache();
+ JavaRDD test = splits[1];
+
+ // Run training algorithm to build the model.
+ final LogisticRegressionModel model = new LogisticRegressionWithLBFGS()
+ .setNumClasses(3)
+ .run(training.rdd());
+
+ // Compute raw scores on the test set.
+ JavaRDD> predictionAndLabels = test.map(
+ new Function>() {
+ public Tuple2 call(LabeledPoint p) {
+ Double prediction = model.predict(p.features());
+ return new Tuple2(prediction, p.label());
+ }
+ }
+ );
+
+ // Get evaluation metrics.
+ MulticlassMetrics metrics = new MulticlassMetrics(predictionAndLabels.rdd());
+
+ // Confusion matrix
+ Matrix confusion = metrics.confusionMatrix();
+ System.out.println("Confusion matrix: \n" + confusion);
+
+ // Overall statistics
+ System.out.println("Precision = " + metrics.precision());
+ System.out.println("Recall = " + metrics.recall());
+ System.out.println("F1 Score = " + metrics.fMeasure());
+
+ // Stats by labels
+ 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]));
+ }
+
+ //Weighted stats
+ 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());
+
+ // Save and load model
+ model.save(sc, "myModelPath");
+ LogisticRegressionModel sameModel = LogisticRegressionModel.load(sc, "myModelPath");
+ }
+}
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight python %}
+from pyspark.mllib.classification import LogisticRegressionWithLBFGS
+from pyspark.mllib.util import MLUtils
+from pyspark.mllib.evaluation import MulticlassMetrics
+
+# Load training data in LIBSVM format
+data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_multiclass_classification_data.txt")
+
+# Split data into training (60%) and test (40%)
+training, test = data.randomSplit([0.6, 0.4], seed = 11L)
+training.cache()
+
+# Run training algorithm to build the model
+model = LogisticRegressionWithLBFGS.train(training, numClasses=3)
+
+# Compute raw scores on the test set
+predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
+
+# Instantiate metrics object
+metrics = MulticlassMetrics(predictionAndLabels)
+
+# Overall statistics
+precision = metrics.precision()
+recall = metrics.recall()
+f1Score = metrics.fMeasure()
+print("Summary Stats")
+print("Precision = %s" % precision)
+print("Recall = %s" % recall)
+print("F1 Score = %s" % f1Score)
+
+# Statistics by class
+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)))
+
+# Weighted stats
+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)
+{% endhighlight %}
+
+
+
+
+ | Metric | Definition |
|---|
+
+
+
+ | Precision | $\frac{1}{N} \sum_{i=0}^{N-1} \frac{\left|P_i \cap L_i\right|}{\left|P_i\right|}$ |
+
+
+ | Recall | $\frac{1}{N} \sum_{i=0}^{N-1} \frac{\left|L_i \cap P_i\right|}{\left|L_i\right|}$ |
+
+
+ | Accuracy |
+
+ $\frac{1}{N} \sum_{i=0}^{N - 1} \frac{\left|L_i \cap P_i \right|}
+ {\left|L_i\right| + \left|P_i\right| - \left|L_i \cap P_i \right|}$
+ |
+
+
+ | Precision by label | $PPV(\ell)=\frac{TP}{TP + FP}=
+ \frac{\sum_{i=0}^{N-1} I_{P_i}(\ell) \cdot I_{L_i}(\ell)}
+ {\sum_{i=0}^{N-1} I_{P_i}(\ell)}$ |
+
+
+ | Recall by label | $TPR(\ell)=\frac{TP}{P}=
+ \frac{\sum_{i=0}^{N-1} I_{P_i}(\ell) \cdot I_{L_i}(\ell)}
+ {\sum_{i=0}^{N-1} I_{L_i}(\ell)}$ |
+
+
+ | F1-measure by label | $F1(\ell) = 2
+ \cdot \left(\frac{PPV(\ell) \cdot TPR(\ell)}
+ {PPV(\ell) + TPR(\ell)}\right)$ |
+
+
+ | Hamming Loss |
+
+ $\frac{1}{N \cdot \left|L\right|} \sum_{i=0}^{N - 1} \left|L_i\right| + \left|P_i\right| - 2\left|L_i
+ \cap P_i\right|$
+ |
+
+
+ | Subset Accuracy |
+ $\frac{1}{N} \sum_{i=0}^{N-1} I_{\{L_i\}}(P_i)$ |
+
+
+ | F1 Measure |
+ $\frac{1}{N} \sum_{i=0}^{N-1} 2 \frac{\left|P_i \cap L_i\right|}{\left|P_i\right| \cdot \left|L_i\right|}$ |
+
+
+ | Micro precision |
+ $\frac{TP}{TP + FP}=\frac{\sum_{i=0}^{N-1} \left|P_i \cap L_i\right|}
+ {\sum_{i=0}^{N-1} \left|P_i \cap L_i\right| + \sum_{i=0}^{N-1} \left|P_i - L_i\right|}$ |
+
+
+ | Micro recall |
+ $\frac{TP}{TP + FN}=\frac{\sum_{i=0}^{N-1} \left|P_i \cap L_i\right|}
+ {\sum_{i=0}^{N-1} \left|P_i \cap L_i\right| + \sum_{i=0}^{N-1} \left|L_i - P_i\right|}$ |
+
+
+ | Micro F1 Measure |
+
+ $2 \cdot \frac{TP}{2 \cdot TP + FP + FN}=2 \cdot \frac{\sum_{i=0}^{N-1} \left|P_i \cap L_i\right|}{2 \cdot
+ \sum_{i=0}^{N-1} \left|P_i \cap L_i\right| + \sum_{i=0}^{N-1} \left|L_i - P_i\right| + \sum_{i=0}^{N-1}
+ \left|P_i - L_i\right|}$
+ |
+
+
+
+
+**Examples**
+
+The following code snippets illustrate how to evaluate the performance of a multilabel classifer. The examples
+use the fake prediction and label data for multilabel classification that is shown below.
+
+Document predictions:
+
+* doc 0 - predict 0, 1 - class 0, 2
+* doc 1 - predict 0, 2 - class 0, 1
+* doc 2 - predict none - class 0
+* doc 3 - predict 2 - class 2
+* doc 4 - predict 2, 0 - class 2, 0
+* doc 5 - predict 0, 1, 2 - class 0, 1
+* doc 6 - predict 1 - class 1, 2
+
+Predicted classes:
+
+* class 0 - doc 0, 1, 4, 5 (total 4)
+* class 1 - doc 0, 5, 6 (total 3)
+* class 2 - doc 1, 3, 4, 5 (total 4)
+
+True classes:
+
+* class 0 - doc 0, 1, 2, 4, 5 (total 5)
+* class 1 - doc 1, 5, 6 (total 3)
+* class 2 - doc 0, 3, 4, 6 (total 4)
+
+
+
+
+
+{% highlight scala %}
+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(), 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)
+
+// Instantiate metrics object
+val metrics = new MultilabelMetrics(scoreAndLabels)
+
+// Summary stats
+println(s"Recall = ${metrics.recall}")
+println(s"Precision = ${metrics.precision}")
+println(s"F1 measure = ${metrics.f1Measure}")
+println(s"Accuracy = ${metrics.accuracy}")
+
+// Individual label stats
+metrics.labels.foreach(label => println(s"Class $label precision = ${metrics.precision(label)}"))
+metrics.labels.foreach(label => println(s"Class $label recall = ${metrics.recall(label)}"))
+metrics.labels.foreach(label => println(s"Class $label F1-score = ${metrics.f1Measure(label)}"))
+
+// Micro stats
+println(s"Micro recall = ${metrics.microRecall}")
+println(s"Micro precision = ${metrics.microPrecision}")
+println(s"Micro F1 measure = ${metrics.microF1Measure}")
+
+// Hamming loss
+println(s"Hamming loss = ${metrics.hammingLoss}")
+
+// Subset accuracy
+println(s"Subset accuracy = ${metrics.subsetAccuracy}")
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight java %}
+import scala.Tuple2;
+
+import org.apache.spark.api.java.*;
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.mllib.evaluation.MultilabelMetrics;
+import org.apache.spark.SparkConf;
+import java.util.Arrays;
+import java.util.List;
+
+public class MultilabelClassification {
+ public static void main(String[] args) {
+ SparkConf conf = new SparkConf().setAppName("Multilabel Classification Metrics");
+ JavaSparkContext sc = new JavaSparkContext(conf);
+
+ List> 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> scoreAndLabels = sc.parallelize(data);
+
+ // Instantiate metrics object
+ MultilabelMetrics metrics = new MultilabelMetrics(scoreAndLabels.rdd());
+
+ // Summary stats
+ System.out.format("Recall = %f\n", metrics.recall());
+ System.out.format("Precision = %f\n", metrics.precision());
+ System.out.format("F1 measure = %f\n", metrics.f1Measure());
+ System.out.format("Accuracy = %f\n", metrics.accuracy());
+
+ // Stats by labels
+ for (int i = 0; i < metrics.labels().length - 1; i++) {
+ System.out.format("Class %1.1f precision = %f\n", metrics.labels()[i], metrics.precision(metrics.labels()[i]));
+ System.out.format("Class %1.1f recall = %f\n", metrics.labels()[i], metrics.recall(metrics.labels()[i]));
+ System.out.format("Class %1.1f F1 score = %f\n", metrics.labels()[i], metrics.f1Measure(metrics.labels()[i]));
+ }
+
+ // Micro stats
+ System.out.format("Micro recall = %f\n", metrics.microRecall());
+ System.out.format("Micro precision = %f\n", metrics.microPrecision());
+ System.out.format("Micro F1 measure = %f\n", metrics.microF1Measure());
+
+ // Hamming loss
+ System.out.format("Hamming loss = %f\n", metrics.hammingLoss());
+
+ // Subset accuracy
+ System.out.format("Subset accuracy = %f\n", metrics.subsetAccuracy());
+
+ }
+}
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight python %}
+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])])
+
+# Instantiate metrics object
+metrics = MultilabelMetrics(scoreAndLabels)
+
+# Summary stats
+print("Recall = %s" % metrics.recall())
+print("Precision = %s" % metrics.precision())
+print("F1 measure = %s" % metrics.f1Measure())
+print("Accuracy = %s" % metrics.accuracy)
+
+# Individual label stats
+labels = scoreAndLabels.flatMap(lambda x: x[1]).distinct().collect()
+for label in 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.f1Measure(label)))
+
+# Micro stats
+print("Micro precision = %s" % metrics.microPrecision)
+print("Micro recall = %s" % metrics.microRecall)
+print("Micro F1 measure = %s" % metrics.microF1Measure)
+
+# Hamming loss
+print("Hamming loss = %s" % metrics.hammingLoss)
+
+# Subset accuracy
+print("Subset accuracy = %s" % metrics.subsetAccuracy)
+
+{% endhighlight %}
+
+
+
+
+ | Metric | Definition | Notes |
|---|
+
+
+
+ |
+ Precision at k
+ |
+
+ $p(k)=\frac{1}{M} \sum_{i=0}^{M-1} {\frac{1}{k} \sum_{j=0}^{\text{min}(\left|D\right|, k) - 1} rel_{D_i}(R_i(j))}$
+ |
+
+ Precision at k is a measure of
+ how many of the first k recommended documents are in the set of true relevant documents averaged across all
+ users. In this metric, the order of the recommendations is not taken into account.
+ |
+
+
+ | Mean Average Precision |
+
+ $MAP=\frac{1}{M} \sum_{i=0}^{M-1} {\frac{1}{\left|D_i\right|} \sum_{j=0}^{Q-1} \frac{rel_{D_i}(R_i(j))}{j + 1}}$
+ |
+
+ MAP is a measure of how
+ many of the recommended documents are in the set of true relevant documents, where the
+ order of the recommendations is taken into account (i.e. penalty for highly relevant documents is higher).
+ |
+
+
+ | Normalized Discounted Cumulative Gain |
+
+ $NDCG(k)=\frac{1}{M} \sum_{i=0}^{M-1} {\frac{1}{IDCG(D_i, k)}\sum_{j=0}^{n-1}
+ \frac{rel_{D_i}(R_i(j))}{\text{ln}(j+1)}} \\
+ \text{Where} \\
+ \hspace{5 mm} n = \text{min}\left(\text{max}\left(|R_i|,|D_i|\right),k\right) \\
+ \hspace{5 mm} IDCG(D, k) = \sum_{j=0}^{\text{min}(\left|D\right|, k) - 1} \frac{1}{\text{ln}(j+1)}$
+ |
+
+ NDCG at k is a
+ measure of how many of the first k recommended documents are in the set of true relevant documents averaged
+ across all users. In contrast to precision at k, this metric takes into account the order of the recommendations
+ (documents are assumed to be in order of decreasing relevance).
+ |
+
+
+
+
+**Examples**
+
+The following code snippets illustrate how to load a sample dataset, train an alternating least squares recommendation
+model on the data, and evaluate the performance of the recommender by several ranking metrics. A brief summary of the
+methodology is provided below.
+
+MovieLens ratings are on a scale of 1-5:
+
+ * 5: Must see
+ * 4: Will enjoy
+ * 3: It's okay
+ * 2: Fairly bad
+ * 1: Awful
+
+So we should not recommend a movie if the predicted rating is less than 3.
+To map ratings to confidence scores, we use:
+
+ * 5 -> 2.5
+ * 4 -> 1.5
+ * 3 -> 0.5
+ * 2 -> -0.5
+ * 1 -> -1.5.
+
+This mappings means unobserved entries are generally between It's okay and Fairly bad. The semantics of 0 in this
+expanded world of non-positive weights are "the same as never having interacted at all."
+
+
+
+
+
+{% highlight scala %}
+import org.apache.spark.mllib.evaluation.{RegressionMetrics, RankingMetrics}
+import org.apache.spark.mllib.recommendation.{ALS, Rating}
+
+// Read in the ratings data
+val ratings = sc.textFile("data/mllib/sample_movielens_data.txt").map { line =>
+ val fields = line.split("::")
+ Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble - 2.5)
+}.cache()
+
+// Map ratings to 1 or 0, 1 indicating a movie that should be recommended
+val binarizedRatings = ratings.map(r => Rating(r.user, r.product, if (r.rating > 0) 1.0 else 0.0)).cache()
+
+// Summarize ratings
+val numRatings = ratings.count()
+val numUsers = ratings.map(_.user).distinct().count()
+val numMovies = ratings.map(_.product).distinct().count()
+println(s"Got $numRatings ratings from $numUsers users on $numMovies movies.")
+
+// Build the model
+val numIterations = 10
+val rank = 10
+val lambda = 0.01
+val model = ALS.train(ratings, rank, numIterations, lambda)
+
+// Define a function to scale ratings from 0 to 1
+def scaledRating(r: Rating): Rating = {
+ val scaledRating = math.max(math.min(r.rating, 1.0), 0.0)
+ Rating(r.user, r.product, scaledRating)
+}
+
+// Get sorted top ten predictions for each user and then scale from [0, 1]
+val userRecommended = model.recommendProductsForUsers(10).map{ case (user, recs) =>
+ (user, recs.map(scaledRating))
+}
+
+// Assume that any movie a user rated 3 or higher (which maps to a 1) is a relevant document
+// Compare with top ten most relevant documents
+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)
+}
+
+// Instantiate metrics object
+val metrics = new RankingMetrics(relevantDocuments)
+
+// Precision at K
+Array(1, 3, 5).foreach{ k =>
+ println(s"Precision at $k = ${metrics.precisionAt(k)}")
+}
+
+// Mean average precision
+println(s"Mean average precision = ${metrics.meanAveragePrecision}")
+
+// Normalized discounted cumulative gain
+Array(1, 3, 5).foreach{ k =>
+ println(s"NDCG at $k = ${metrics.ndcgAt(k)}")
+}
+
+// Get predictions for each data point
+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)
+}
+
+// Get the RMSE using regression metrics
+val regressionMetrics = new RegressionMetrics(predictionsAndLabels)
+println(s"RMSE = ${regressionMetrics.rootMeanSquaredError}")
+
+// R-squared
+println(s"R-squared = ${regressionMetrics.r2}")
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight java %}
+import scala.Tuple2;
+
+import org.apache.spark.api.java.*;
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.Function;
+import java.util.*;
+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.Rating;
+
+// Read in the ratings data
+public class Ranking {
+ public static void main(String[] args) {
+ SparkConf conf = new SparkConf().setAppName("Ranking Metrics");
+ JavaSparkContext sc = new JavaSparkContext(conf);
+ String path = "data/mllib/sample_movielens_data.txt";
+ JavaRDD data = sc.textFile(path);
+ JavaRDD ratings = data.map(
+ new Function() {
+ public Rating call(String line) {
+ String[] parts = line.split("::");
+ return new Rating(Integer.parseInt(parts[0]), Integer.parseInt(parts[1]), Double.parseDouble(parts[2]) - 2.5);
+ }
+ }
+ );
+ ratings.cache();
+
+ // Train an ALS model
+ final MatrixFactorizationModel model = ALS.train(JavaRDD.toRDD(ratings), 10, 10, 0.01);
+
+ // Get top 10 recommendations for every user and scale ratings from 0 to 1
+ JavaRDD> userRecs = model.recommendProductsForUsers(10).toJavaRDD();
+ JavaRDD> userRecsScaled = userRecs.map(
+ new Function, Tuple2>() {
+ public Tuple2 call(Tuple2 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 userRecommended = JavaPairRDD.fromJavaRDD(userRecsScaled);
+
+ // Map ratings to 1 or 0, 1 indicating a movie that should be recommended
+ JavaRDD binarizedRatings = ratings.map(
+ new Function() {
+ public Rating call(Rating r) {
+ double binaryRating;
+ if (r.rating() > 0.0) {
+ binaryRating = 1.0;
+ }
+ else {
+ binaryRating = 0.0;
+ }
+ return new Rating(r.user(), r.product(), binaryRating);
+ }
+ }
+ );
+
+ // Group ratings by common user
+ JavaPairRDD> userMovies = binarizedRatings.groupBy(
+ new Function() {
+ public Object call(Rating r) {
+ return r.user();
+ }
+ }
+ );
+
+ // Get true relevant documents from all user ratings
+ JavaPairRDD> userMoviesList = userMovies.mapValues(
+ new Function, List>() {
+ public List call(Iterable docs) {
+ List products = new ArrayList();
+ for (Rating r : docs) {
+ if (r.rating() > 0.0) {
+ products.add(r.product());
+ }
+ }
+ return products;
+ }
+ }
+ );
+
+ // Extract the product id from each recommendation
+ JavaPairRDD> userRecommendedList = userRecommended.mapValues(
+ new Function>() {
+ public List call(Rating[] docs) {
+ List products = new ArrayList();
+ for (Rating r : docs) {
+ products.add(r.product());
+ }
+ return products;
+ }
+ }
+ );
+ JavaRDD, List>> relevantDocs = userMoviesList.join(userRecommendedList).values();
+
+ // Instantiate the metrics object
+ RankingMetrics metrics = RankingMetrics.of(relevantDocs);
+
+ // Precision and NDCG at 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));
+ }
+
+ // Mean average precision
+ System.out.format("Mean average precision = %f\n", metrics.meanAveragePrecision());
+
+ // Evaluate the model using numerical ratings and regression metrics
+ JavaRDD> userProducts = ratings.map(
+ new Function>() {
+ public Tuple2 call(Rating r) {
+ return new Tuple2(r.user(), r.product());
+ }
+ }
+ );
+ JavaPairRDD, Object> predictions = JavaPairRDD.fromJavaRDD(
+ model.predict(JavaRDD.toRDD(userProducts)).toJavaRDD().map(
+ new Function, Object>>() {
+ public Tuple2, Object> call(Rating r){
+ return new Tuple2, Object>(
+ new Tuple2(r.user(), r.product()), r.rating());
+ }
+ }
+ ));
+ JavaRDD> ratesAndPreds =
+ JavaPairRDD.fromJavaRDD(ratings.map(
+ new Function, Object>>() {
+ public Tuple2, Object> call(Rating r){
+ return new Tuple2, Object>(
+ new Tuple2(r.user(), r.product()), r.rating());
+ }
+ }
+ )).join(predictions).values();
+
+ // Create regression metrics object
+ RegressionMetrics regressionMetrics = new RegressionMetrics(ratesAndPreds.rdd());
+
+ // Root mean squared error
+ System.out.format("RMSE = %f\n", regressionMetrics.rootMeanSquaredError());
+
+ // R-squared
+ System.out.format("R-squared = %f\n", regressionMetrics.r2());
+ }
+}
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight python %}
+from pyspark.mllib.recommendation import ALS, Rating
+from pyspark.mllib.evaluation import RegressionMetrics, RankingMetrics
+
+# Read in the ratings data
+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))
+
+# Train a model on to predict user-product ratings
+model = ALS.train(ratings, 10, 10, 0.01)
+
+# Get predicted ratings on all existing user-product pairs
+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])
+
+# Instantiate regression metrics to compare predicted and actual ratings
+metrics = RegressionMetrics(scoreAndLabels)
+
+# Root mean sqaured error
+print("RMSE = %s" % metrics.rootMeanSquaredError)
+
+# R-squared
+print("R-squared = %s" % metrics.r2)
+
+{% endhighlight %}
+
+
+
+
+ | Metric | Definition |
|---|
+
+
+
+ | Mean Squared Error (MSE) |
+ $MSE = \frac{\sum_{i=0}^{N-1} (\mathbf{y}_i - \hat{\mathbf{y}}_i)^2}{N}$ |
+
+
+ | Root Mean Squared Error (RMSE) |
+ $RMSE = \sqrt{\frac{\sum_{i=0}^{N-1} (\mathbf{y}_i - \hat{\mathbf{y}}_i)^2}{N}}$ |
+
+
+ | Mean Absoloute Error (MAE) |
+ $MAE=\sum_{i=0}^{N-1} \left|\mathbf{y}_i - \hat{\mathbf{y}}_i\right|$ |
+
+
+ | Coefficient of Determination $(R^2)$ |
+ $R^2=1 - \frac{MSE}{\text{VAR}(\mathbf{y}) \cdot (N-1)}=1-\frac{\sum_{i=0}^{N-1}
+ (\mathbf{y}_i - \hat{\mathbf{y}}_i)^2}{\sum_{i=0}^{N-1}(\mathbf{y}_i-\bar{\mathbf{y}})^2}$ |
+
+
+ | Explained Variance |
+ $1 - \frac{\text{VAR}(\mathbf{y} - \mathbf{\hat{y}})}{\text{VAR}(\mathbf{y})}$ |
+
+
+
+
+**Examples**
+
+
+The following code snippets illustrate how to load a sample dataset, train a linear regression algorithm on the data,
+and evaluate the performance of the algorithm by several regression metrics.
+
+
+
+{% highlight scala %}
+import org.apache.spark.mllib.regression.LabeledPoint
+import org.apache.spark.mllib.regression.LinearRegressionModel
+import org.apache.spark.mllib.regression.LinearRegressionWithSGD
+import org.apache.spark.mllib.linalg.Vectors
+import org.apache.spark.mllib.evaluation.RegressionMetrics
+import org.apache.spark.mllib.util.MLUtils
+
+// Load the data
+val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_linear_regression_data.txt").cache()
+
+// Build the model
+val numIterations = 100
+val model = LinearRegressionWithSGD.train(data, numIterations)
+
+// Get predictions
+val valuesAndPreds = data.map{ point =>
+ val prediction = model.predict(point.features)
+ (prediction, point.label)
+}
+
+// Instantiate metrics object
+val metrics = new RegressionMetrics(valuesAndPreds)
+
+// Squared error
+println(s"MSE = ${metrics.meanSquaredError}")
+println(s"RMSE = ${metrics.rootMeanSquaredError}")
+
+// R-squared
+println(s"R-squared = ${metrics.r2}")
+
+// Mean absolute error
+println(s"MAE = ${metrics.meanAbsoluteError}")
+
+// Explained variance
+println(s"Explained variance = ${metrics.explainedVariance}")
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight java %}
+import scala.Tuple2;
+
+import org.apache.spark.api.java.*;
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.mllib.linalg.Vectors;
+import org.apache.spark.mllib.regression.LabeledPoint;
+import org.apache.spark.mllib.regression.LinearRegressionModel;
+import org.apache.spark.mllib.regression.LinearRegressionWithSGD;
+import org.apache.spark.mllib.evaluation.RegressionMetrics;
+import org.apache.spark.SparkConf;
+
+public class LinearRegression {
+ public static void main(String[] args) {
+ SparkConf conf = new SparkConf().setAppName("Linear Regression Example");
+ JavaSparkContext sc = new JavaSparkContext(conf);
+
+ // Load and parse the data
+ String path = "data/mllib/sample_linear_regression_data.txt";
+ JavaRDD data = sc.textFile(path);
+ JavaRDD parsedData = data.map(
+ new Function() {
+ public LabeledPoint call(String line) {
+ String[] parts = line.split(" ");
+ double[] v = new double[parts.length - 1];
+ for (int i = 1; i < parts.length - 1; i++)
+ v[i - 1] = Double.parseDouble(parts[i].split(":")[1]);
+ return new LabeledPoint(Double.parseDouble(parts[0]), Vectors.dense(v));
+ }
+ }
+ );
+ parsedData.cache();
+
+ // Building the model
+ int numIterations = 100;
+ final LinearRegressionModel model =
+ LinearRegressionWithSGD.train(JavaRDD.toRDD(parsedData), numIterations);
+
+ // Evaluate model on training examples and compute training error
+ JavaRDD> valuesAndPreds = parsedData.map(
+ new Function>() {
+ public Tuple2 call(LabeledPoint point) {
+ double prediction = model.predict(point.features());
+ return new Tuple2(prediction, point.label());
+ }
+ }
+ );
+
+ // Instantiate metrics object
+ RegressionMetrics metrics = new RegressionMetrics(valuesAndPreds.rdd());
+
+ // Squared error
+ System.out.format("MSE = %f\n", metrics.meanSquaredError());
+ System.out.format("RMSE = %f\n", metrics.rootMeanSquaredError());
+
+ // R-squared
+ System.out.format("R Squared = %f\n", metrics.r2());
+
+ // Mean absolute error
+ System.out.format("MAE = %f\n", metrics.meanAbsoluteError());
+
+ // Explained variance
+ System.out.format("Explained Variance = %f\n", metrics.explainedVariance());
+
+ // Save and load model
+ model.save(sc.sc(), "myModelPath");
+ LinearRegressionModel sameModel = LinearRegressionModel.load(sc.sc(), "myModelPath");
+ }
+}
+
+{% endhighlight %}
+
+
+
+
+
+{% highlight python %}
+from pyspark.mllib.regression import LabeledPoint, LinearRegressionWithSGD
+from pyspark.mllib.evaluation import RegressionMetrics
+from pyspark.mllib.linalg import DenseVector
+
+# Load and parse the data
+def parsePoint(line):
+ values = line.split()
+ return LabeledPoint(float(values[0]), DenseVector([float(x.split(':')[1]) for x in values[1:]]))
+
+data = sc.textFile("data/mllib/sample_linear_regression_data.txt")
+parsedData = data.map(parsePoint)
+
+# Build the model
+model = LinearRegressionWithSGD.train(parsedData)
+
+# Get predictions
+valuesAndPreds = parsedData.map(lambda p: (float(model.predict(p.features)), p.label))
+
+# Instantiate metrics object
+metrics = RegressionMetrics(valuesAndPreds)
+
+# Squared Error
+print("MSE = %s" % metrics.meanSquaredError)
+print("RMSE = %s" % metrics.rootMeanSquaredError)
+
+# R-squared
+print("R-squared = %s" % metrics.r2)
+
+# Mean absolute error
+print("MAE = %s" % metrics.meanAbsoluteError)
+
+# Explained variance
+print("Explained variance = %s" % metrics.explainedVariance)
+
+{% endhighlight %}
+
+
+