//设置package包名称以及导入依赖的类
package org.sparksamples.regression.bikesharing

import org.apache.log4j.Logger
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{VectorAssembler, VectorIndexer}
import org.apache.spark.ml.regression.LinearRegression
import org.apache.spark.mllib.evaluation.RegressionMetrics
import org.apache.spark.sql.{DataFrame, SparkSession}


object LinearRegressionPipeline {
  @transient lazy val logger = Logger.getLogger(getClass.getName)

  def linearRegressionWithVectorFormat(vectorAssembler: VectorAssembler, vectorIndexer: VectorIndexer, dataFrame: DataFrame) = {
    val lr = new LinearRegression()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setRegParam(0.1)
      .setElasticNetParam(1.0)
      .setMaxIter(10)

    val pipeline = new Pipeline().setStages(Array(vectorAssembler, vectorIndexer, lr))

    val Array(training, test) = dataFrame.randomSplit(Array(0.8, 0.2), seed = 12345)

    val model = pipeline.fit(training)

    val fullPredictions = model.transform(test).cache()
    val predictions = fullPredictions.select("prediction").rdd.map(_.getDouble(0))
    val labels = fullPredictions.select("label").rdd.map(_.getDouble(0))
    val RMSE = new RegressionMetrics(predictions.zip(labels)).rootMeanSquaredError
    println(s"  Root mean squared error (RMSE): $RMSE")
  }

  def linearRegressionWithSVMFormat(spark: SparkSession) = {
    // Load training data
    val training = spark.read.format("libsvm")
      .load("./src/main/scala/org/sparksamples/regression/dataset/BikeSharing/lsvmHours.txt")

    val lr = new LinearRegression()
      .setMaxIter(10)
      .setRegParam(0.3)
      .setElasticNetParam(0.8)

    // Fit the model
    val lrModel = lr.fit(training)

    // Print the coefficients and intercept for linear regression
    println(s"Coefficients: ${lrModel.coefficients} Intercept: ${lrModel.intercept}")

    // Summarize the model over the training set and print out some metrics
    val trainingSummary = lrModel.summary
    println(s"numIterations: ${trainingSummary.totalIterations}")
    println(s"objectiveHistory: ${trainingSummary.objectiveHistory.toList}")
    trainingSummary.residuals.show()
    println(s"RMSE: ${trainingSummary.rootMeanSquaredError}")

    println(s"r2: ${trainingSummary.r2}")
  }
} 

//设置package包名称以及导入依赖的类
package songs

import org.apache.spark.ml.PipelineModel
import org.apache.spark.ml.regression.LinearRegressionModel
import org.apache.spark.mllib.evaluation.RegressionMetrics
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.slf4j.LoggerFactory

object TrainModel {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName(Config.appName)
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val logger = LoggerFactory.getLogger(getClass.getName)

    logger.info("Loading datasets from parquet format")
    val data = SongML.loadModelData(sqlContext = sqlContext)

    logger.info("Showing summary stats for training data")
    val summary = data.training.describe(SongML.allColumns:_*)
    summary.show(1000)

    logger.info("Training Linear Regression Model")
    val startTime = System.nanoTime()

    val pipeline = SongML.trainingPipeline.fit(data.training)

    val elapsedTime = (System.nanoTime() - startTime) / 1e9
    logger.info(s"Training time: $elapsedTime seconds")

    logger.info("Calculating Regression Metrics")
    val bestModel = pipeline.bestModel.asInstanceOf[PipelineModel]
    val testPredictions: RDD[(Double,Double)] = bestModel.transform(data.training)
      .select(SongML.predictionColumn, SongML.labelColumn)
      .map(r => (r.getAs[Double](SongML.predictionColumn), r.getAs[Double](SongML.labelColumn)))

    val rm = new RegressionMetrics(testPredictions)

    val model = bestModel.stages(SongML.lrStages.indices.last).asInstanceOf[LinearRegressionModel]

    logger.info(s"Saving model to ${Config.modelOut}")
    model.write.overwrite().save(Config.modelOut)

    logger.info(SongML.printStats(model,rm,"Training"))

    logger.info("Exiting")
    sc.stop()
  }
} 

//设置package包名称以及导入依赖的类
package songs

import org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel}
import org.apache.spark.mllib.evaluation.RegressionMetrics
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.slf4j.LoggerFactory

object EvaluateModel {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName(Config.appName)
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val logger = LoggerFactory.getLogger(getClass.getName)

    logger.info(s"Loading Linear Regression Model from ${Config.modelOut}")
    val model = LinearRegressionModel.load(Config.modelOut)

    logger.info("Loading datasets")
    val datasets = SongML.loadModelData(sqlContext = sqlContext)
    val pipelineModel = SongML.transformPipeline.fit(datasets.training)
    val testData = pipelineModel.transform(datasets.test).select(SongML.labelColumn,SongML.featuresColumn)

    logger.info("Calculating Regression Metrics")
    val testPredictions = model.transform(testData)
      .select(SongML.labelColumn,SongML.predictionColumn)
      .map(r => (r.getAs[Double](SongML.predictionColumn), r.getAs[Double](SongML.labelColumn)))
    val rm = new RegressionMetrics(testPredictions)

    logger.info(SongML.printStats(model,rm,"Testing"))

    logger.info("Exiting")
    sc.stop()
  }

}