<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 1.3. BigQuery Storage & Spark MLlib - Python
Use the BigQuery storage connector and Spark MLlib to build a Linear Regression model and make predictions
Create Dataproc Cluster with Jupyter
This notebook is designed to be run on Google Cloud Dataproc.
Follow the links below for instructions on how to create a Dataproc Cluster with the Juypter component installed.
Tutorial - Install and run a Jupyter notebook on a Dataproc cluster
Blog post - Apache Spark and Jupyter Notebooks made easy with Dataproc component gateway
Python 3 Kernel
Use a Python 3 kernel (not PySpark) to allow you to configure the SparkSession in the notebook and include the spark-bigquery-connector required to use the BigQuery Storage API.
Scala Version
Check what version of Scala you are running so you can include the correct spark-bigquery-connector jar
Step1:  Create Spark Session
Include the correct version of the spark-bigquery-connector jar
If you are using scala version 2.11 use 'gs
Step2:  Enable repl.eagerEval
This will output the results of DataFrames in each step without the new need to show df.show() and also improves the formatting of the output
Step3:  Read the data from BigQuery as a Spark Dataframe
Step5:  Limit no of rows and cache data
limit the no of rows that will be read for this example to run faster. 
The DataFrame is cached as LinearRegression is iterative and this avoids re-reading the data from BigQuery Storage for each iteration
Step6:  Optional
Run count to check no of rows in DataFrame
Step7:  Create an input DataFrame for Spark MLlib using VectorAssembler
Spark MLlib estimators expect a single vector column for features. Multiple columns can be converted to a single vector column using VectorAssembler
Step8:  Create a training data DataFrame
Create a training data DataFrame with just the features and label column.
Cache the training data table.
Step9:  Split the data into training and test sets
30% held out for testing
Step10:  Construct a new LinearRegression object and fit the training data
Import and use and the LinearRegression model
Step11:  Print the coefficients and intercept for linear regression
Step12:  Summarize the model over the training data and print metrics
Step13:  Make predictions on test data
Step14:  Select (prediction, true label) and compute test error
<ASSISTANT_TASK:>
Python Code:
!scala -version
Explanation: 1.3. BigQuery Storage & Spark MLlib - Python
Use the BigQuery storage connector and Spark MLlib to build a Linear Regression model and make predictions
Create Dataproc Cluster with Jupyter
This notebook is designed to be run on Google Cloud Dataproc.
Follow the links below for instructions on how to create a Dataproc Cluster with the Juypter component installed.
Tutorial - Install and run a Jupyter notebook on a Dataproc cluster
Blog post - Apache Spark and Jupyter Notebooks made easy with Dataproc component gateway
Python 3 Kernel
Use a Python 3 kernel (not PySpark) to allow you to configure the SparkSession in the notebook and include the spark-bigquery-connector required to use the BigQuery Storage API.
Scala Version
Check what version of Scala you are running so you can include the correct spark-bigquery-connector jar
End of explanation
from pyspark.sql import SparkSession
spark = SparkSession.builder \
  .appName('1.3. BigQuery Storage &  Spark MLlib - Python')\
  .config('spark.jars', 'gs://spark-lib/bigquery/spark-bigquery-latest.jar') \
  .getOrCreate()
spark.version
Explanation: Create Spark Session
Include the correct version of the spark-bigquery-connector jar
If you are using scala version 2.11 use 'gs://spark-lib/bigquery/spark-bigquery-latest.jar'.
If you are using scala version 2.12 use 'gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar'.
End of explanation
spark.conf.set("spark.sql.repl.eagerEval.enabled",True)
Explanation: Enable repl.eagerEval
This will output the results of DataFrames in each step without the new need to show df.show() and also improves the formatting of the output
End of explanation
table  = 'bigquery-public-data.samples.natality'
df_natality_table = spark.read \
  .format("bigquery") \
  .option("table", table) \
  .load()
Explanation: Read the data from BigQuery as a Spark Dataframe
End of explanation
limit = 10000
df_natality_select = df_natality_table \
.select("weight_pounds", "mother_age", "father_age", "gestation_weeks", "weight_gain_pounds", "apgar_5min") \
.where(
weight_pounds IS NOT NULL 
AND mother_age IS NOT NULL
AND father_age IS NOT NULL
AND gestation_weeks IS NOT NULL
AND weight_gain_pounds IS NOT NULL
AND apgar_5min IS NOT NULL
) \
.limit(limit) \
.cache()
df_natality_select.printSchema()
Explanation: Limit no of rows and cache data
limit the no of rows that will be read for this example to run faster. 
The DataFrame is cached as LinearRegression is iterative and this avoids re-reading the data from BigQuery Storage for each iteration
End of explanation
df_natality_select.count()
Explanation: Optional
Run count to check no of rows in DataFrame
End of explanation
from pyspark.ml.linalg import Vectors
from pyspark.ml.feature import VectorAssembler
assembler = VectorAssembler(
    inputCols=["mother_age", "father_age", "gestation_weeks", "weight_gain_pounds", "apgar_5min"],
    outputCol="features")
df_assembler_output = assembler.transform(df_natality_select)
df_assembler_output
Explanation: Create an input DataFrame for Spark MLlib using VectorAssembler
Spark MLlib estimators expect a single vector column for features. Multiple columns can be converted to a single vector column using VectorAssembler
End of explanation
df_training_data = df_assembler_output \
.select("features", "weight_pounds") \
.withColumnRenamed("weight_pounds","label")
df_training_data.cache()
df_training_data
Explanation: Create a training data DataFrame
Create a training data DataFrame with just the features and label column.
Cache the training data table.
End of explanation
(df_training, df_test) = df_training_data.randomSplit([0.7, 0.3])
Explanation: Split the data into training and test sets
30% held out for testing
End of explanation
from pyspark.ml.regression import LinearRegression
lr = LinearRegression(maxIter=5, regParam=0.2, solver="normal")
model = lr.fit(df_training)
Explanation: Construct a new LinearRegression object and fit the training data
Import and use and the LinearRegression model
End of explanation
print("Coefficients:" + str(model.coefficients))
print("Intercept:" + str(model.intercept))
Explanation: Print the coefficients and intercept for linear regression
End of explanation
trainingSummary = model.summary
print("numIterations: %d" % trainingSummary.totalIterations)
print("objectiveHistory: %s" % str(trainingSummary.objectiveHistory))
print("RMSE: %f" % trainingSummary.rootMeanSquaredError)
print("r2: %f" % trainingSummary.r2)
trainingSummary.residuals
Explanation: Summarize the model over the training data and print metrics
End of explanation
predictions = model.transform(df_test)
predictions.select("prediction", "label", "features")
Explanation: Make predictions on test data
End of explanation
from pyspark.ml.evaluation import RegressionEvaluator
evaluator = RegressionEvaluator(
    labelCol="label", predictionCol="prediction", metricName="rmse")
rmse = evaluator.evaluate(predictions)
print("Root Mean Squared Error (RMSE) on test data = %g" % rmse)
Explanation: Select (prediction, true label) and compute test error
End of explanation
<END_TASK>