保序回归（sotonic regression）

Question

问题描述：给定一个无序数字序列 x，通过修改每个元素的值得到一个非递减序列 x ，问如何使 y 和 y' 误差（该处取平方差）最小？

保序回归法：从该序列 x 的首元素往后观察，一旦出现乱序现象停止该轮观察，从该乱序元素开始逐个吸收紧跟在后面的元素组成一个子序列 sub_x，直到子序列 sub_x ​ 所有元素的平均值小于或等于下一个待吸收的元素。

举例：

原始序列：<9, 10, 14>
结果序列：<9, 10, 14>

分析：从 9 往后观察，到最后的元素 14 都未发现乱序情况，不用处理。

原始序列：<9, 14, 10>
结果序列：<9, 12, 12>

分析：从 9 往后观察，观察到 14 时发生乱序（14>10），停止该轮观察转入吸收元素处理，吸收元素 10 后子序列为<14, 10>，取该序列所有元素的平均值得 12，故用序列<12, 12>替代<14, 10>。吸收 10 后已经到了最后的元素，处理操作完成。

原始序列：<14, 9, 10, 15>
结果序列：<11, 11, 11, 15>

分析：从 14 往后观察，观察到 9 时发生乱序（14>9），停止该轮观察转入吸收元素处理，吸收元素 9 后子序列为 <14, 9>。求该序列所有元素的平均值得 12.5，由于 12.5 大于下个带吸收的元素 10，所以再吸收 10，得序列 <14, 9, 10>。求该序列所有元素的平均值得 11，由于 11 小于下个带吸收的元素 15，所以停止吸收操作，用序列 <11, 11, 11> 替代 <14, 9, 10>。

上面的求解方法是 保序回归 的一种特殊的可行解但不是最优解，下面我们给出一般性定义： 对一个有限的实数集合 Y 表示观测响应，X 集合表示未知的响应值，进行拟合找到一个最小化函数：

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

保序的意思就是 x1≤x2≤…≤xn ​，现在问题就转换成最小二乘问题了。

MLlib 支持的算法平行化保序回归，保序回归有一个参数 isotonic，默认值是 true，此参数指定保序回归是保序的（单调增加）还是不保序的（单调减少）。

保序回归的结果被视为分段线性函数，预测的规则是：

1. 如果预测输入能准确匹配训练特征，那么返回相关预测，如果有多个预测匹配训练特征，那么就返回其中之一。
2. 如果预测输入比所有的训练特征低或者高，那么最低和最高的训练特征各自返回。如果有多个预测比所有的训练特征低或者高，那么都会返回。
3. 如果预测输入介于两个训练特征，那么预测会被视为分段线性函数和从最接近的训练特征中计算得到的插值。

例子

从文件中读取数据后对每一行格式化成 label,feature i.e. 4710.28,500.00. ​ 数据分割成训练集和测试集两个部分。通过训练集创建模型然后在利用测试集计算平方差误差

val PATH = "file:///Users/lzz/work/SparkML/"

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

val data = sc.textFile(PATH+"data/mllib/sample_isotonic_regression_data.txt")

// Create label, feature, weight tuples from input data with weight set to default value 1.0.
val parsedData = data.map { line =>
  val parts = line.split(',').map(_.toDouble)
  (parts(0), parts(1), 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("Mean Squared Error = " + meanSquaredError)

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

Mean Squared Error = 0.004883368896285485