回归 - 基于 RDD 的 API

保序回归

保序回归属于回归算法家族。形式上，保序回归是一个问题，给定一个有限的实数集 $Y = {y_1, y_2, ..., y_n}$ 表示观察到的响应， $X = {x_1, x_2, ..., x_n}$ 表示要拟合的未知响应值，找到一个函数，使

\begin{equation} f(x) = \sum_{i=1}^n w_i (y_i - x_i)^2 \end{equation}

在完全排序约束下最小化 $x_1\le x_2\le ...\le x_n$ ，其中 $w_i$ 是正权重。得到的函数称为保序回归，它是唯一的。它可以看作是顺序约束下的最小二乘问题。本质上，保序回归是一个最适合原始数据点的单调函数。

spark.mllib 支持相邻违规者池算法，该算法使用一种并行化保序回归的方法。训练输入是一个由三个双精度值组成的元组的 RDD，分别表示标签、特征和权重。如果有多个元组具有相同的特征，则这些元组将聚合为一个元组，如下所示

聚合标签是所有标签的加权平均值。
聚合特征是唯一的特征值。
聚合权重是所有权重的总和。

此外，IsotonicRegression 算法还有一个可选参数 $isotonic$，默认为 true。该参数指定保序回归是保序的（单调递增）还是反序的（单调递减）。

训练返回一个 IsotonicRegressionModel，可用于预测已知和未知特征的标签。保序回归的结果被视为分段线性函数。因此，预测规则如下

如果预测输入与训练特征完全匹配，则返回关联的预测。
如果预测输入低于或高于所有训练特征，则分别返回具有最低或最高特征的预测。
如果预测输入介于两个训练特征之间，则将预测视为分段线性函数，并根据两个最接近特征的预测计算插值。

示例

从文件中读取数据，每行的格式为标签、特征，例如 4710.28,500.00。数据被分为训练集和测试集。使用训练集创建模型，并根据测试集中预测的标签和真实标签计算均方误差。

有关 API 的更多详细信息，请参阅 IsotonicRegression Python 文档和 IsotonicRegressionModel Python 文档。

import math
from pyspark.mllib.regression import IsotonicRegression, IsotonicRegressionModel
from pyspark.mllib.util import MLUtils

# Load and parse the data
def parsePoint(labeledData):
    return (labeledData.label, labeledData.features[0], 1.0)

data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_isotonic_regression_libsvm_data.txt")

# Create label, feature, weight tuples from input data with weight set to default value 1.0.
parsedData = data.map(parsePoint)

# Split data into training (60%) and test (40%) sets.
training, test = parsedData.randomSplit([0.6, 0.4], 11)

# Create isotonic regression model from training data.
# Isotonic parameter defaults to true so it is only shown for demonstration
model = IsotonicRegression.train(training)

# Create tuples of predicted and real labels.
predictionAndLabel = test.map(lambda p: (model.predict(p[1]), p[0]))

# Calculate mean squared error between predicted and real labels.
meanSquaredError = predictionAndLabel.map(lambda pl: math.pow((pl[0] - pl[1]), 2)).mean()
print("Mean Squared Error = " + str(meanSquaredError))

# Save and load model
model.save(sc, "target/tmp/myIsotonicRegressionModel")
sameModel = IsotonicRegressionModel.load(sc, "target/tmp/myIsotonicRegressionModel")

在 Spark 存储库的 "examples/src/main/python/mllib/isotonic_regression_example.py" 中查找完整的示例代码。

有关 API 的详细信息，请参阅 IsotonicRegression Scala 文档和 IsotonicRegressionModel Scala 文档。

import org.apache.spark.mllib.regression.{IsotonicRegression, IsotonicRegressionModel}
import org.apache.spark.mllib.util.MLUtils

val data = MLUtils.loadLibSVMFile(sc,
  "data/mllib/sample_isotonic_regression_libsvm_data.txt").cache()

// Create label, feature, weight tuples from input data with weight set to default value 1.0.
val parsedData = data.map { labeledPoint =>
  (labeledPoint.label, labeledPoint.features(0), 1.0)
}

// Split data into training (60%) and test (40%) sets.
val splits = parsedData.randomSplit(Array(0.6, 0.4), seed = 11L)
val training = splits(0)
val test = splits(1)

// Create isotonic regression model from training data.
// Isotonic parameter defaults to true so it is only shown for demonstration
val model = new IsotonicRegression().setIsotonic(true).run(training)

// Create tuples of predicted and real labels.
val predictionAndLabel = test.map { point =>
  val predictedLabel = model.predict(point._2)
  (predictedLabel, point._1)
}

// Calculate mean squared error between predicted and real labels.
val meanSquaredError = predictionAndLabel.map { case (p, l) => math.pow((p - l), 2) }.mean()
println(s"Mean Squared Error = $meanSquaredError")

// Save and load model
model.save(sc, "target/tmp/myIsotonicRegressionModel")
val sameModel = IsotonicRegressionModel.load(sc, "target/tmp/myIsotonicRegressionModel")

在 Spark 存储库的 "examples/src/main/scala/org/apache/spark/examples/mllib/IsotonicRegressionExample.scala" 中查找完整的示例代码。

有关 API 的详细信息，请参阅 IsotonicRegression Java 文档和 IsotonicRegressionModel Java 文档。

import scala.Tuple2;
import scala.Tuple3;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.regression.IsotonicRegression;
import org.apache.spark.mllib.regression.IsotonicRegressionModel;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;

JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(
  jsc.sc(), "data/mllib/sample_isotonic_regression_libsvm_data.txt").toJavaRDD();

// Create label, feature, weight tuples from input data with weight set to default value 1.0.
JavaRDD<Tuple3<Double, Double, Double>> parsedData = data.map(point ->
  new Tuple3<>(point.label(), point.features().apply(0), 1.0));

// Split data into training (60%) and test (40%) sets.
JavaRDD<Tuple3<Double, Double, Double>>[] splits =
  parsedData.randomSplit(new double[]{0.6, 0.4}, 11L);
JavaRDD<Tuple3<Double, Double, Double>> training = splits[0];
JavaRDD<Tuple3<Double, Double, Double>> test = splits[1];

// Create isotonic regression model from training data.
// Isotonic parameter defaults to true so it is only shown for demonstration
IsotonicRegressionModel model = new IsotonicRegression().setIsotonic(true).run(training);

// Create tuples of predicted and real labels.
JavaPairRDD<Double, Double> predictionAndLabel = test.mapToPair(point ->
  new Tuple2<>(model.predict(point._2()), point._1()));

// Calculate mean squared error between predicted and real labels.
double meanSquaredError = predictionAndLabel.mapToDouble(pl -> {
  double diff = pl._1() - pl._2();
  return diff * diff;
}).mean();
System.out.println("Mean Squared Error = " + meanSquaredError);

// Save and load model
model.save(jsc.sc(), "target/tmp/myIsotonicRegressionModel");
IsotonicRegressionModel sameModel =
  IsotonicRegressionModel.load(jsc.sc(), "target/tmp/myIsotonicRegressionModel");

在 Spark 存储库的 "examples/src/main/java/org/apache/spark/examples/mllib/JavaIsotonicRegressionExample.java" 中查找完整的示例代码。

MLlib：主指南

MLlib：基于 RDD 的 API 指南

回归 - 基于 RDD 的 API

保序回归

示例