MLlib--保序回归

 1 import org.apache.log4j.{Level, Logger}
 2 import org.apache.spark.rdd.RDD
 3 import org.apache.spark.{SparkConf, SparkContext}
 4 import org.apache.spark.mllib.regression.{IsotonicRegressionModel, IsotonicRegression}
 5 
 6 /**
 7   * Created by hzf
 8   */
 9 object IsotonicRegression_new {
10     //  F:\额外项目\pensionRisk\data\IsR\train\sample_isotonic_regression_data.txt F:\额外项目\pensionRisk\data\IsR\model true local
11     def main(args: Array[String]) {
12         Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
13         if (args.length < 4) {
14             System.err.println("Usage: LRwithLGD <inputPath> <modelPath> Isotonic <master> [<AppName>]")
15             System.err.println("eg: hdfs://192.168.57.104:8020/user/000000_0 hdfs://192.168.57.104:8020/user/model true  spark://192.168.57.104:7077  IsotonicRegression")
16             System.exit(1)
17         }
18         val appName = if (args.length > 4) args(4) else "IsotonicRegression"
19         val conf = new SparkConf().setAppName(appName).setMaster(args(3))
20         val sc = new SparkContext(conf)
21         var isotonic = true
22         isotonic = args(2) match {
23             case "true" => true
24             case "false" => false
25         }
26         val data = sc.textFile(args(0))
27         val parsedData: RDD[(Double, Double, Double)] = data.map { line =>
28             val parts = line.split(‘,‘).map(_.toDouble)
29             (parts(0), parts(1), 1.0)
30         }
31 
32         val splitRdd: Array[RDD[(Double, Double, Double)]] = parsedData.randomSplit(Array(1.0, 9.0))
33         val testData = splitRdd(0)
34         val realTrainData: RDD[(Double, Double, Double)] = splitRdd(1)
35 
36         val model: IsotonicRegressionModel = new IsotonicRegression().setIsotonic(isotonic).run(realTrainData)
37         val predictionAndLabel = testData.map { point =>
38             val predictedLabel = model.predict(point._2)
39             (predictedLabel, point._1)
40         }
41 
42         val meanSquaredError = predictionAndLabel.map { case p => math.pow((p._1 - p._2), 2) }.mean()
43         println("meanSquaredError = " + meanSquaredError)
44         model.boundaries.zip(model.predictions).foreach(println(_))
45         model.save(sc, args(1))
46 
47     }
48 }