SDS-2.2, Scalable Data Science

Archived YouTube video of this live unedited lab-lecture:

Power Plant ML Pipeline Application

This is an end-to-end example of using a number of different machine learning algorithms to solve a supervised regression problem.

Table of Contents

• Step 1: Business Understanding
• Step 2: Load Your Data
• Step 3: Explore Your Data
• Step 4: Visualize Your Data
• Step 5: Data Preparation
• Step 6: Data Modeling
• Step 7: Tuning and Evaluation
• Step 8: Deployment

We are trying to predict power output given a set of readings from various sensors in a gas-fired power generation plant. Power generation is a complex process, and understanding and predicting power output is an important element in managing a plant and its connection to the power grid.

More information about Peaker or Peaking Power Plants can be found on Wikipedia https://en.wikipedia.org/wiki/Peaking*power*plant

Given this business problem, we need to translate it to a Machine Learning task. The ML task is regression since the label (or target) we are trying to predict is numeric.

The example data is provided by UCI at UCI Machine Learning Repository Combined Cycle Power Plant Data Set

You can read the background on the UCI page, but in summary we have collected a number of readings from sensors at a Gas Fired Power Plant

(also called a Peaker Plant) and now we want to use those sensor readings to predict how much power the plant will generate.

More information about Machine Learning with Spark can be found in the Spark MLLib Programming Guide

Please note this example only works with Spark version 1.4 or higher

To Rerun Steps 1-4 done in the notebook at:

• Workspace -> scalable-data-science -> sds-2-2 -> 009PowerPlantPipeline01ETLEDA

just run the following command as shown in the cell below:

%run "/scalable-data-science/sds-2-2/009_PowerPlantPipeline_01ETLEDA"

• Note: If you already evaluated the %run ... command above then:

  • first delete the cell by pressing on x on the top-right corner of the cell and
  • revaluate the run command above.

"/scalable-data-science/sds-2-2/009_PowerPlantPipeline_01ETLEDA"

Now we will do the following Steps:

Step 5: Data Preparation,

Step 6: Modeling, and

Step 7: Tuning and Evaluation

We will do Step 8: Deployment later after we get introduced to SparkStreaming.

Step 5: Data Preparation

The next step is to prepare the data. Since all of this data is numeric and consistent, this is a simple task for us today.

We will need to convert the predictor features from columns to Feature Vectors using the org.apache.spark.ml.feature.VectorAssembler

The VectorAssembler will be the first step in building our ML pipeline.

//Let's quickly recall the schema and make sure our table is here now
table("power_plant_table").printSchema

root
 |-- AT: double (nullable = true)
 |-- V: double (nullable = true)
 |-- AP: double (nullable = true)
 |-- RH: double (nullable = true)
 |-- PE: double (nullable = true)

powerPlantDF // make sure we have the DataFrame too

res23: org.apache.spark.sql.DataFrame = [AT: double, V: double ... 3 more fields]

import org.apache.spark.ml.feature.VectorAssembler

// make a DataFrame called dataset from the table
val dataset = sqlContext.table("power_plant_table") 

val vectorizer =  new VectorAssembler()
                      .setInputCols(Array("AT", "V", "AP", "RH"))
                      .setOutputCol("features")

import org.apache.spark.ml.feature.VectorAssembler
dataset: org.apache.spark.sql.DataFrame = [AT: double, V: double ... 3 more fields]
vectorizer: org.apache.spark.ml.feature.VectorAssembler = vecAssembler_bca323dc00ef

Step 6: Data Modeling

Now let's model our data to predict what the power output will be given a set of sensor readings

Our first model will be based on simple linear regression since we saw some linear patterns in our data based on the scatter plots during the exploration stage.

Linear Regression Model

• Linear Regression is one of the most useful work-horses of statistical learning
• See Chapter 7 of Kevin Murphy's Machine Learning froma Probabilistic Perspective for a good mathematical and algorithmic introduction.
• You should have already seen Ameet's treatment of the topic from earlier notebook.

Let's open http://spark.apache.org/docs/latest/mllib-linear-methods.html#regression for some details.

// First let's hold out 20% of our data for testing and leave 80% for training
var Array(split20, split80) = dataset.randomSplit(Array(0.20, 0.80), 1800009193L)

split20: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [AT: double, V: double ... 3 more fields]
split80: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [AT: double, V: double ... 3 more fields]

// Let's cache these datasets for performance
val testSet = split20.cache()
val trainingSet = split80.cache()

testSet: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [AT: double, V: double ... 3 more fields]
trainingSet: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [AT: double, V: double ... 3 more fields]

testSet.count() // action to actually cache

res27: Long = 1966

trainingSet.count() // action to actually cache

res28: Long = 7602

Let's take a few elements of the three DataFrames.

dataset.take(3)

res24: Array[org.apache.spark.sql.Row] = Array([14.96,41.76,1024.07,73.17,463.26], [25.18,62.96,1020.04,59.08,444.37], [5.11,39.4,1012.16,92.14,488.56])

testSet.take(3)

res25: Array[org.apache.spark.sql.Row] = Array([1.81,39.42,1026.92,76.97,490.55], [3.2,41.31,997.67,98.84,489.86], [3.38,41.31,998.79,97.76,489.11])

trainingSet.take(3)

res26: Array[org.apache.spark.sql.Row] = Array([2.34,39.42,1028.47,69.68,490.34], [2.58,39.42,1028.68,69.03,488.69], [2.64,39.64,1011.02,85.24,481.29])

// ***** LINEAR REGRESSION MODEL ****

import org.apache.spark.ml.regression.LinearRegression
import org.apache.spark.ml.regression.LinearRegressionModel
import org.apache.spark.ml.Pipeline

// Let's initialize our linear regression learner
val lr = new LinearRegression()

import org.apache.spark.ml.regression.LinearRegression
import org.apache.spark.ml.regression.LinearRegressionModel
import org.apache.spark.ml.Pipeline
lr: org.apache.spark.ml.regression.LinearRegression = linReg_161e726c1f38

// We use explain params to dump the parameters we can use
lr.explainParams()

res29: String =
aggregationDepth: suggested depth for treeAggregate (>= 2) (default: 2)
elasticNetParam: the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty (default: 0.0)
featuresCol: features column name (default: features)
fitIntercept: whether to fit an intercept term (default: true)
labelCol: label column name (default: label)
maxIter: maximum number of iterations (>= 0) (default: 100)
predictionCol: prediction column name (default: prediction)
regParam: regularization parameter (>= 0) (default: 0.0)
solver: the solver algorithm for optimization. If this is not set or empty, default value is 'auto' (default: auto)
standardization: whether to standardize the training features before fitting the model (default: true)
tol: the convergence tolerance for iterative algorithms (>= 0) (default: 1.0E-6)
weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0 (undefined)

The cell below is based on the Spark ML pipeline API. More information can be found in the Spark ML Programming Guide at https://spark.apache.org/docs/latest/ml-guide.html

// Now we set the parameters for the method
lr.setPredictionCol("Predicted_PE")
  .setLabelCol("PE")
  .setMaxIter(100)
  .setRegParam(0.1)
// We will use the new spark.ml pipeline API. If you have worked with scikit-learn this will be very familiar.
val lrPipeline = new Pipeline()
lrPipeline.setStages(Array(vectorizer, lr))
// Let's first train on the entire dataset to see what we get
val lrModel = lrPipeline.fit(trainingSet)

lrPipeline: org.apache.spark.ml.Pipeline = pipeline_8b0213f4b86e
lrModel: org.apache.spark.ml.PipelineModel = pipeline_8b0213f4b86e

Since Linear Regression is simply a line of best fit over the data that minimizes the square of the error, given multiple input dimensions we can express each predictor as a line function of the form:

$y = b_0 + b_1 x_1 + b_2 x_2 + b_3 x_3 + \ldots + b_i x_i + \ldots + b_k x_k$

where $b_0$ is the intercept and $b_i$'s are coefficients.

To express the coefficients of that line we can retrieve the Estimator stage from the fitted, linear-regression pipeline model named lrModel and express the weights and the intercept for the function.

// The intercept is as follows:
val intercept = lrModel.stages(1).asInstanceOf[LinearRegressionModel].intercept

intercept: Double = 427.9139822165837

// The coefficents (i.e. weights) are as follows:

val weights = lrModel.stages(1).asInstanceOf[LinearRegressionModel].coefficients.toArray

weights: Array[Double] = Array(-1.9083064919040942, -0.25381293007161654, 0.08739350304730673, -0.1474651301033126)

The model has been fit and the intercept and coefficients displayed above.

Now, let us do some work to make a string of the model that is easy to understand for an applied data scientist or data analyst.

val featuresNoLabel = dataset.columns.filter(col => col != "PE")

featuresNoLabel: Array[String] = Array(AT, V, AP, RH)

val coefficentFeaturePairs = sc.parallelize(weights).zip(sc.parallelize(featuresNoLabel))

coefficentFeaturePairs: org.apache.spark.rdd.RDD[(Double, String)] = ZippedPartitionsRDD2[35297] at zip at <console>:42

coefficentFeaturePairs.collect() // this just pairs each coefficient with the name of its corresponding feature

res30: Array[(Double, String)] = Array((-1.9083064919040942,AT), (-0.25381293007161654,V), (0.08739350304730673,AP), (-0.1474651301033126,RH))

// Now let's sort the coefficients from the largest to the smallest

var equation = s"y = $intercept "
//var variables = Array
coefficentFeaturePairs.sortByKey().collect().foreach({
  case (weight, feature) =>
  { 
        val symbol = if (weight > 0) "+" else "-"
        val absWeight = Math.abs(weight)
        equation += (s" $symbol (${absWeight} * ${feature})")
  }
}
)

equation: String = y = 427.9139822165837  - (1.9083064919040942 * AT) - (0.25381293007161654 * V) - (0.1474651301033126 * RH) + (0.08739350304730673 * AP)

// Finally here is our equation
println("Linear Regression Equation: " + equation)

Linear Regression Equation: y = 427.9139822165837  - (1.9083064919040942 * AT) - (0.25381293007161654 * V) - (0.1474651301033126 * RH) + (0.08739350304730673 * AP)

Based on examining the fitted Linear Regression Equation above:

• There is a strong negative correlation between Atmospheric Temperature (AT) and Power Output due to the coefficient being greater than -1.91.
• But our other dimenensions seem to have little to no correlation with Power Output.

Do you remember Step 2: Explore Your Data? When we visualized each predictor against Power Output using a Scatter Plot, only the temperature variable seemed to have a linear correlation with Power Output so our final equation seems logical.

Now let's see what our predictions look like given this model.

val predictionsAndLabels = lrModel.transform(testSet)

display(predictionsAndLabels.select("AT", "V", "AP", "RH", "PE", "Predicted_PE"))

Truncated to 30 rows

Now that we have real predictions we can use an evaluation metric such as Root Mean Squared Error to validate our regression model. The lower the Root Mean Squared Error, the better our model.

//Now let's compute some evaluation metrics against our test dataset

import org.apache.spark.mllib.evaluation.RegressionMetrics 

val metrics = new RegressionMetrics(predictionsAndLabels.select("Predicted_PE", "PE").rdd.map(r => (r(0).asInstanceOf[Double], r(1).asInstanceOf[Double])))

import org.apache.spark.mllib.evaluation.RegressionMetrics
metrics: org.apache.spark.mllib.evaluation.RegressionMetrics = org.apache.spark.mllib.evaluation.RegressionMetrics@32433b70

val rmse = metrics.rootMeanSquaredError

rmse: Double = 4.609375859170583

val explainedVariance = metrics.explainedVariance

explainedVariance: Double = 274.54186073318266

val r2 = metrics.r2

r2: Double = 0.9308377700269259

println (f"Root Mean Squared Error: $rmse")
println (f"Explained Variance: $explainedVariance")  
println (f"R2: $r2")

Root Mean Squared Error: 4.609375859170583
Explained Variance: 274.54186073318266
R2: 0.9308377700269259

Generally a good model will have 68% of predictions within 1 RMSE and 95% within 2 RMSE of the actual value. Let's calculate and see if our RMSE meets this criteria.

display(predictionsAndLabels) // recall the DataFrame predictionsAndLabels

// First we calculate the residual error and divide it by the RMSE from predictionsAndLabels DataFrame and make another DataFrame that is registered as a temporary table Power_Plant_RMSE_Evaluation
predictionsAndLabels.selectExpr("PE", "Predicted_PE", "PE - Predicted_PE AS Residual_Error", s""" (PE - Predicted_PE) / $rmse AS Within_RSME""").createOrReplaceTempView("Power_Plant_RMSE_Evaluation")

SELECT * from Power_Plant_RMSE_Evaluation

Truncated to 30 rows

-- Now we can display the RMSE as a Histogram. Clearly this shows that the RMSE is centered around 0 with the vast majority of the error within 2 RMSEs.
SELECT Within_RSME  from Power_Plant_RMSE_Evaluation

Truncated to 30 rows

We can see this definitively if we count the number of predictions within + or - 1.0 and + or - 2.0 and display this as a pie chart:

SELECT case when Within_RSME <= 1.0 and Within_RSME >= -1.0 then 1  when  Within_RSME <= 2.0 and Within_RSME >= -2.0 then 2 else 3 end RSME_Multiple, COUNT(*) count  from Power_Plant_RMSE_Evaluation
group by case when Within_RSME <= 1.0 and Within_RSME >= -1.0 then 1  when  Within_RSME <= 2.0 and Within_RSME >= -2.0 then 2 else 3 end

So we have about 70% of our training data within 1 RMSE and about 97% (70% + 27%) within 2 RMSE. So the model is pretty decent. Let's see if we can tune the model to improve it further.

NOTE: these numbers will vary across runs due to the seed in random sampling of training and test set, number of iterations, and other stopping rules in optimization, for example.

Step 7: Tuning and Evaluation

Now that we have a model with all of the data let's try to make a better model by tuning over several parameters.

import org.apache.spark.ml.tuning.{ParamGridBuilder, CrossValidator}
import org.apache.spark.ml.evaluation._

import org.apache.spark.ml.tuning.{ParamGridBuilder, CrossValidator}
import org.apache.spark.ml.evaluation._

First let's use a cross validator to split the data into training and validation subsets. See http://spark.apache.org/docs/latest/ml-tuning.html.

//Let's set up our evaluator class to judge the model based on the best root mean squared error
val regEval = new RegressionEvaluator()
regEval.setLabelCol("PE")
  .setPredictionCol("Predicted_PE")
  .setMetricName("rmse")

regEval: org.apache.spark.ml.evaluation.RegressionEvaluator = regEval_c6d2feda2ee0
res37: regEval.type = regEval_c6d2feda2ee0

We now treat the lrPipeline as an Estimator, wrapping it in a CrossValidator instance.

This will allow us to jointly choose parameters for all Pipeline stages.

A CrossValidator requires an Estimator, an Evaluator (which we set next).

//Let's create our crossvalidator with 3 fold cross validation
val crossval = new CrossValidator()
crossval.setEstimator(lrPipeline)
crossval.setNumFolds(3)
crossval.setEvaluator(regEval)

crossval: org.apache.spark.ml.tuning.CrossValidator = cv_c77b278496fa
res38: crossval.type = cv_c77b278496fa

A CrossValidator also requires a set of EstimatorParamMaps which we set next.

For this we need a regularization parameter (more generally a hyper-parameter that is model-specific).

Now, let's tune over our regularization parameter from 0.01 to 0.10.

val regParam = ((1 to 10) toArray).map(x => (x /100.0))

warning: there was one feature warning; re-run with -feature for details
regParam: Array[Double] = Array(0.01, 0.02, 0.03, 0.04, 0.05, 0.06, 0.07, 0.08, 0.09, 0.1)

Check out the scala docs for syntactic details on org.apache.spark.ml.tuning.ParamGridBuilder.

val paramGrid = new ParamGridBuilder()
                    .addGrid(lr.regParam, regParam)
                    .build()
crossval.setEstimatorParamMaps(paramGrid)

paramGrid: Array[org.apache.spark.ml.param.ParamMap] =
Array({
    linReg_161e726c1f38-regParam: 0.01
}, {
    linReg_161e726c1f38-regParam: 0.02
}, {
    linReg_161e726c1f38-regParam: 0.03
}, {
    linReg_161e726c1f38-regParam: 0.04
}, {
    linReg_161e726c1f38-regParam: 0.05
}, {
    linReg_161e726c1f38-regParam: 0.06
}, {
    linReg_161e726c1f38-regParam: 0.07
}, {
    linReg_161e726c1f38-regParam: 0.08
}, {
    linReg_161e726c1f38-regParam: 0.09
}, {
    linReg_161e726c1f38-regParam: 0.1
})
res39: crossval.type = cv_c77b278496fa

//Now let's create our model
val cvModel = crossval.fit(trainingSet)

cvModel: org.apache.spark.ml.tuning.CrossValidatorModel = cv_c77b278496fa

In addition to CrossValidator Spark also offers TrainValidationSplit for hyper-parameter tuning. TrainValidationSplit only evaluates each combination of parameters once as opposed to k times in case of CrossValidator. It is therefore less expensive, but will not produce as reliable results when the training dataset is not sufficiently large.

• http://spark.apache.org/docs/latest/ml-tuning.html#train-validation-split

Now that we have tuned let's see what we got for tuning parameters and what our RMSE was versus our intial model

val predictionsAndLabels = cvModel.transform(testSet)
val metrics = new RegressionMetrics(predictionsAndLabels.select("Predicted_PE", "PE").rdd.map(r => (r(0).asInstanceOf[Double], r(1).asInstanceOf[Double])))

val rmse = metrics.rootMeanSquaredError
val explainedVariance = metrics.explainedVariance
val r2 = metrics.r2

predictionsAndLabels: org.apache.spark.sql.DataFrame = [AT: double, V: double ... 5 more fields]
metrics: org.apache.spark.mllib.evaluation.RegressionMetrics = org.apache.spark.mllib.evaluation.RegressionMetrics@77606693
rmse: Double = 4.599964072968395
explainedVariance: Double = 277.2272873387723
r2: Double = 0.9311199234339246

println (f"Root Mean Squared Error: $rmse")
println (f"Explained Variance: $explainedVariance")  
println (f"R2: $r2")

Root Mean Squared Error: 4.599964072968395
Explained Variance: 277.2272873387723
R2: 0.9311199234339246

Let us explore other models to see if we can predict the power output better

There are several families of models in Spark's scalable machine learning library:

• http://spark.apache.org/docs/latest/ml-classification-regression.html

So our initial untuned and tuned linear regression models are statistically identical.

Given that the only linearly correlated variable is Temperature, it makes sense try another machine learning method such a Decision Tree to handle non-linear data and see if we can improve our model

A Decision Tree creates a model based on splitting variables using a tree structure. We will first start with a single decision tree model.

Reference Decision Trees: https://en.wikipedia.org/wiki/Decision*tree*learning

//Let's build a decision tree pipeline
import org.apache.spark.ml.regression.DecisionTreeRegressor

// we are using a Decision Tree Regressor as opposed to a classifier we used for the hand-written digit classification problem
val dt = new DecisionTreeRegressor()
dt.setLabelCol("PE")
dt.setPredictionCol("Predicted_PE")
dt.setFeaturesCol("features")
dt.setMaxBins(100)

val dtPipeline = new Pipeline()
dtPipeline.setStages(Array(vectorizer, dt))

import org.apache.spark.ml.regression.DecisionTreeRegressor
dt: org.apache.spark.ml.regression.DecisionTreeRegressor = dtr_532b2d8e3739
dtPipeline: org.apache.spark.ml.Pipeline = pipeline_604afd9d7ee3
res41: dtPipeline.type = pipeline_604afd9d7ee3

//Let's just resuse our CrossValidator
crossval.setEstimator(dtPipeline)

res42: crossval.type = cv_c77b278496fa

val paramGrid = new ParamGridBuilder()
                     .addGrid(dt.maxDepth, Array(2, 3))
                     .build()

paramGrid: Array[org.apache.spark.ml.param.ParamMap] =
Array({
    dtr_532b2d8e3739-maxDepth: 2
}, {
    dtr_532b2d8e3739-maxDepth: 3
})

crossval.setEstimatorParamMaps(paramGrid)

res43: crossval.type = cv_c77b278496fa

val dtModel = crossval.fit(trainingSet) // fit decitionTree with cv

dtModel: org.apache.spark.ml.tuning.CrossValidatorModel = cv_c77b278496fa

import org.apache.spark.ml.regression.DecisionTreeRegressionModel
import org.apache.spark.ml.PipelineModel
dtModel.bestModel.asInstanceOf[PipelineModel].stages.last.asInstanceOf[DecisionTreeRegressionModel].toDebugString

import org.apache.spark.ml.regression.DecisionTreeRegressionModel
import org.apache.spark.ml.PipelineModel
res45: String =
"DecisionTreeRegressionModel (uid=dtr_532b2d8e3739) of depth 3 with 15 nodes
  If (feature 0 <= 17.84)
   If (feature 0 <= 11.95)
    If (feature 0 <= 8.75)
     Predict: 483.5412151067323
    Else (feature 0 > 8.75)
     Predict: 475.6305502392345
   Else (feature 0 > 11.95)
    If (feature 0 <= 15.33)
     Predict: 467.63141917293234
    Else (feature 0 > 15.33)
     Predict: 460.74754125412574
  Else (feature 0 > 17.84)
   If (feature 0 <= 23.02)
    If (feature 1 <= 47.83)
     Predict: 457.1077966101695
    Else (feature 1 > 47.83)
     Predict: 448.74750213858016
   Else (feature 0 > 23.02)
    If (feature 1 <= 66.25)
     Predict: 442.88544855967086
    Else (feature 1 > 66.25)
     Predict: 434.7293710691822
"

The line above will pull the Decision Tree model from the Pipeline and display it as an if-then-else string.

Next let's visualize it as a decision tree for regression.

display(dtModel.bestModel.asInstanceOf[PipelineModel].stages.last.asInstanceOf[DecisionTreeRegressionModel])

Now let's see how our DecisionTree model compares to our LinearRegression model

val predictionsAndLabels = dtModel.bestModel.transform(testSet)
val metrics = new RegressionMetrics(predictionsAndLabels.select("Predicted_PE", "PE").map(r => (r(0).asInstanceOf[Double], r(1).asInstanceOf[Double])).rdd)

val rmse = metrics.rootMeanSquaredError
val explainedVariance = metrics.explainedVariance
val r2 = metrics.r2

println (f"Root Mean Squared Error: $rmse")
println (f"Explained Variance: $explainedVariance")  
println (f"R2: $r2")

Root Mean Squared Error: 5.221342219456633
Explained Variance: 269.66550072645475
R2: 0.9112539444165726
predictionsAndLabels: org.apache.spark.sql.DataFrame = [AT: double, V: double ... 5 more fields]
metrics: org.apache.spark.mllib.evaluation.RegressionMetrics = org.apache.spark.mllib.evaluation.RegressionMetrics@2e492425
rmse: Double = 5.221342219456633
explainedVariance: Double = 269.66550072645475
r2: Double = 0.9112539444165726

So our DecisionTree was slightly worse than our LinearRegression model (LR: 4.6 vs DT: 5.2). Maybe we can try an Ensemble method such as Gradient-Boosted Decision Trees to see if we can strengthen our model by using an ensemble of weaker trees with weighting to reduce the error in our model.

Note since this is a complex model, the cell below can take about 16 minutes* or so to run on a small cluster with a couple nodes with about 6GB RAM, go out and grab a coffee and come back :-).*

This GBTRegressor code will be way faster on a larger cluster of course.

A visual explanation of gradient boosted trees:

• http://arogozhnikov.github.io/2016/06/24/gradientboostingexplained.html

Let's see what a boosting algorithm, a type of ensemble method, is all about in more detail.

import org.apache.spark.ml.regression.GBTRegressor

val gbt = new GBTRegressor()
gbt.setLabelCol("PE")
gbt.setPredictionCol("Predicted_PE")
gbt.setFeaturesCol("features")
gbt.setSeed(100088121L)
gbt.setMaxBins(100)
gbt.setMaxIter(120)

val gbtPipeline = new Pipeline()
gbtPipeline.setStages(Array(vectorizer, gbt))
//Let's just resuse our CrossValidator

crossval.setEstimator(gbtPipeline)

val paramGrid = new ParamGridBuilder()
  .addGrid(gbt.maxDepth, Array(2, 3))
  .build()
crossval.setEstimatorParamMaps(paramGrid)

//gbt.explainParams
val gbtModel = crossval.fit(trainingSet)

import org.apache.spark.ml.regression.GBTRegressionModel 

val predictionsAndLabels = gbtModel.bestModel.transform(testSet)
val metrics = new RegressionMetrics(predictionsAndLabels.select("Predicted_PE", "PE").map(r => (r(0).asInstanceOf[Double], r(1).asInstanceOf[Double])).rdd)

val rmse = metrics.rootMeanSquaredError
val explainedVariance = metrics.explainedVariance
val r2 = metrics.r2


println (f"Root Mean Squared Error: $rmse")
println (f"Explained Variance: $explainedVariance")  
println (f"R2: $r2")

We can use the toDebugString method to dump out what our trees and weighting look like:

gbtModel.bestModel.asInstanceOf[PipelineModel].stages.last.asInstanceOf[GBTRegressionModel].toDebugString

Conclusion

Wow! So our best model is in fact our Gradient Boosted Decision tree model which uses an ensemble of 120 Trees with a depth of 3 to construct a better model than the single decision tree.

Step 8: Deployment will be done later

Now that we have a predictive model it is time to deploy the model into an operational environment.

In our example, let's say we have a series of sensors attached to the power plant and a monitoring station.

The monitoring station will need close to real-time information about how much power that their station will generate so they can relay that to the utility.

For this we need to create a Spark Streaming utility that we can use for this purpose. For this you need to be introduced to basic concepts in Spark Streaming first. See http://spark.apache.org/docs/latest/streaming-programming-guide.html if you can't wait!

After deployment you will be able to use the best predictions from gradient boosed regression trees to feed a real-time dashboard or feed the utility with information on how much power the peaker plant will deliver give current conditions.

Datasource References:

• Pinar Tüfekci, Prediction of full load electrical power output of a base load operated combined cycle power plant using machine learning methods, International Journal of Electrical Power & Energy Systems, Volume 60, September 2014, Pages 126-140, ISSN 0142-0615, Web Link
• Heysem Kaya, Pinar Tüfekci , Sadik Fikret Gürgen: Local and Global Learning Methods for Predicting Power of a Combined Gas & Steam Turbine, Proceedings of the International Conference on Emerging Trends in Computer and Electronics Engineering ICETCEE 2012, pp. 13-18 (Mar. 2012, Dubai) Web Link