保序回归算法原理及Spark MLlib调用实例(Scala/Java/python)

保序回归

算法介绍：

       保序回归是回归算法的一种。保序回归给定一个有限的实数集合

 代表观察到的响应，以及

 代表未知的响应值，训练一个模型来最小化下列方程：


 

       其中

 ，

 为权重是正值。其结果方程称为保序回归，而且其解是唯一的。它可以被视为有顺序约束下的最小二乘法问题。实际上保序回归在拟合原始数据点时是一个单调函数。我们实现池旁者算法，它使用并行保序回归。训练数据是DataFrame格式，包含标签、特征值以及权重三列。另外保序算法还有一个参数名为isotonic，其默认值为真，它指定保序回归为保序（单调递增）或者反序（单调递减）。 

       训练返回一个保序回归模型，可以被用于来预测已知或者未知特征值的标签。保序回归的结果是分段线性函数，预测规则如下：
1.如果预测输入与训练中的特征值完全匹配，则返回相应标签。如果一个特征值对应多个预测标签值，则返回其中一个，具体是哪一个未指定。
2.如果预测输入比训练中的特征值都高（或者都低），则相应返回最高特征值或者最低特征值对应标签。如果一个特征值对应多个预测标签值，则相应返回最高值或者最低值。
3.如果预测输入落入两个特征值之间，则预测将会是一个分段线性函数，其值由两个最近的特征值的预测值计算得到。如果一个特征值对应多个预测标签值，则使用上述两种情况中的处理方式解决。

参数：

featuresIndex:

类型：整数型。

含义：当特征列维向量时提供索引值，否则不进行处理。

featuresCol:

类型：字符串型。

含义：特征列名。

isotonic:

类型：布尔型。

含义：输出序列为保序/增序（真）或者反序/降序（假）。

labelCol:

类型：字符串型。

含义：标签列名。

predictionCol:

类型：字符串型。

含义：预测结果列名。

weightCol:

类型：字符串型。

含义：列权重。

调用示例：

Scala:

import org.apache.spark.ml.regression.IsotonicRegression

// Loads data.
val dataset = spark.read.format("libsvm")
.load("data/mllib/sample_isotonic_regression_libsvm_data.txt")

// Trains an isotonic regression model.
val ir = new IsotonicRegression()
val model = ir.fit(dataset)

println(s"Boundaries in increasing order: ${model.boundaries}")
println(s"Predictions associated with the boundaries: ${model.predictions}")

// Makes predictions.
model.transform(dataset).show()

Java:

import org.apache.spark.ml.regression.IsotonicRegression;
import org.apache.spark.ml.regression.IsotonicRegressionModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

// Loads data.
Dataset<Row> dataset = spark.read().format("libsvm")
.load("data/mllib/sample_isotonic_regression_libsvm_data.txt");

// Trains an isotonic regression model.
IsotonicRegression ir = new IsotonicRegression();
IsotonicRegressionModel model = ir.fit(dataset);

System.out.println("Boundaries in increasing order: " + model.boundaries());
System.out.println("Predictions associated with the boundaries: " + model.predictions());

// Makes predictions.
model.transform(dataset).show();

Python:

from pyspark.ml.regression import IsotonicRegression, IsotonicRegressionModel

# Loads data.
dataset = spark.read.format("libsvm")\
.load("data/mllib/sample_isotonic_regression_libsvm_data.txt")

# Trains an isotonic regression model.
model = IsotonicRegression().fit(dataset)
print("Boundaries in increasing order: " + str(model.boundaries))
print("Predictions associated with the boundaries: " + str(model.predictions))

# Makes predictions.
model.transform(dataset).show()