<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Data Source
Step1:  Load Data
Step2:  Convert Spark Dataframe to Pandas Dataframe
Step3:  Verctorize the features
Step4:  Fit Linear Regression Model
Step5:  View model summary
Step6:  Predict
Step7:  Evaluate
Step8:  Build a pipeline
Step9:  Save the pipeline to disk to persist the model
Step10:  Load the persisted model from the disk
Step11:  Tune the model
<ASSISTANT_TASK:>
Python Code:
!ls -ltr /data
spark
Explanation: Data Source: https://archive.ics.uci.edu/ml/datasets/Combined+Cycle+Power+Plant
Features consist of hourly average ambient variables
Temperature (T) in the range 1.81°C and 37.11°C,
Ambient Pressure (AP) in the range 992.89-1033.30 milibar,
Relative Humidity (RH) in the range 25.56% to 100.16%
Exhaust Vacuum (V) in teh range 25.36-81.56 cm Hg
Net hourly electrical energy output (EP) 420.26-495.76 MW
The averages are taken from various sensors located around the plant that record the ambient variables every second. The variables are given without normalization.
Dataset Information:
The dataset contains 9568 data points collected from a Combined Cycle Power Plant over 6 years (2006-2011), when the power plant was set to work with full load. Features consist of hourly average ambient variables Temperature (T), Ambient Pressure (AP), Relative Humidity (RH) and Exhaust Vacuum (V) to predict the net hourly electrical energy output (EP) of the plant. 
A combined cycle power plant (CCPP) is composed of gas turbines (GT), steam turbines (ST) and heat recovery steam generators. In a CCPP, the electricity is generated by gas and steam turbines, which are combined in one cycle, and is transferred from one turbine to another. While the Vacuum is colected from and has effect on the Steam Turbine, he other three of the ambient variables effect the GT performance.
End of explanation
df = spark.read.format("csv").option("header","true")\
.option("inferSchema","true").load("/data/Combined_Cycle_Power_Plant.csv")
df.show()
df.cache()
Explanation: Load Data
End of explanation
df.limit(10).toPandas().head() 
Explanation: Convert Spark Dataframe to Pandas Dataframe
End of explanation
from pyspark.ml.feature import *
vectorizer = VectorAssembler()
vectorizer.setInputCols(["AT", "V", "AP", "RH"])
vectorizer.setOutputCol("features")
df_vect = vectorizer.transform(df)
df_vect.show(10, False)
print(vectorizer.explainParams())
Explanation: Verctorize the features
End of explanation
from pyspark.ml.regression import LinearRegression
lr = LinearRegression()
print(lr.explainParams())
lr.setLabelCol("EP")
lr.setFeaturesCol("features")
model = lr.fit(df_vect)
type(model)
Explanation: Fit Linear Regression Model
End of explanation
print("R2:", model.summary.r2)
print("Intercept: ", model.intercept, "Coefficients", model.coefficients)
Explanation: View model summary
End of explanation
df_pred = model.transform(df_vect)
df_pred.show()
Explanation: Predict
End of explanation
from pyspark.ml.evaluation import RegressionEvaluator
evaluator = RegressionEvaluator()
print(evaluator.explainParams())
evaluator = RegressionEvaluator(labelCol = "EP", 
                                predictionCol = "prediction", 
                                metricName = "rmse")
evaluator.evaluate(df_pred)
Explanation: Evaluate
End of explanation
from pyspark.ml.pipeline import Pipeline, PipelineModel
pipeline = Pipeline()
print(pipeline.explainParams())
pipeline.setStages([vectorizer, lr])
pipelineModel = pipeline.fit(df)
pipeline.getStages()
lr_model = pipelineModel.stages[1]
lr_model .coefficients
pipelineModel.transform(df).show()
evaluator.evaluate(pipelineModel.transform(df))
Explanation: Build a pipeline
End of explanation
pipelineModel.save("/tmp/lr-pipeline")
!tree /tmp/lr-pipeline
Explanation: Save the pipeline to disk to persist the model
End of explanation
saved_model = PipelineModel.load("/tmp/lr-pipeline")
saved_model.stages[1].coefficients
saved_model.transform(df).show()
df_train, df_test = df.randomSplit(weights=[0.7, 0.3], seed = 200)
pipelineModel = pipeline.fit(df_train)
evaluator.evaluate(pipelineModel.transform(df_test))
Explanation: Load the persisted model from the disk
End of explanation
from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit
paramGrid = ParamGridBuilder()\
    .addGrid(lr.regParam, [0.1, 0.01]) \
    .addGrid(lr.fitIntercept, [False, True])\
    .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0])\
    .build()
# In this case the estimator is simply the linear regression.
# A TrainValidationSplit requires an Estimator, a set of Estimator ParamMaps, and an Evaluator.
tvs = TrainValidationSplit(estimator=lr,
                           estimatorParamMaps=paramGrid,
                           evaluator=evaluator,
                           trainRatio=0.8)
tuned_model = tvs.fit(vectorizer.transform(df_train))
tuned_model.bestModel, tuned_model.validationMetrics
df_test_pred = tuned_model.transform(vectorizer.transform(df_test))
df_test_pred.show()
evaluator.evaluate(df_test_pred)
Explanation: Tune the model
End of explanation
<END_TASK>