import org.apache.log4j.{Level, Logger}

import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS

import org.apache.spark.mllib.evaluation.MulticlassMetrics

import org.apache.spark.{SparkConf, SparkContext}

import org.apache.spark.mllib.util.MLUtils

import org.apache.spark.mllib.regression.LabeledPoint

object LogisticRegression {

  def main(args: Array[String]) {

    // 设置运行环境

    val conf = new SparkConf().setAppName("Logistic Regression Test")

      .setMaster("spark://master:7077").setJars(Seq("E:\\Intellij\\Projects\\MachineLearning\\MachineLearning.jar"))

    val sc = new SparkContext(conf)

    Logger.getRootLogger.setLevel(Level.WARN)

    // 读取样本数据,格式化为LIBSVM的RDD

    val dataRDD = MLUtils.loadLibSVMFile(sc, "hdfs://master:9000/ml/data/sample_libsvm_data.txt")

    // 样本数据划分,训练样本占0.7,测试样本占0.3

    val dataParts = dataRDD.randomSplit(Array(0.7, 0.3), seed = 25L)

    val trainRDD = dataParts(0).cache()

    val testRDD = dataParts(1)

    // 建立逻辑回归模型并训练

    val LRModel = new LogisticRegressionWithLBFGS().setNumClasses(10).run(trainRDD)

    // 对测试样本进行测试

    val prediction = testRDD.map {

      case LabeledPoint(label, features) =>

        val prediction = LRModel.predict(features)

        (prediction, label)

    }

    val showPrediction = prediction.take(10)

    // 输出测试结果

    println("Prediction" + "\t" + "Label")

    for (i <- 0 to showPrediction.length - 1) {

      println(showPrediction(i)._1 + "\t" + showPrediction(i)._2)

    }

    // 计算误差并输出

    val metrics = new MulticlassMetrics(prediction)

    val precision = metrics.precision

    println("Precision = " + precision)

  }

}