Console

1. Create a dataset in your project.

   1. Go to the BigQuery Web UI.
   2. In the left navigation panel, click your project name, then clickCREATE DATASET.
   3. In theCreate dataset dialog:
      1. ForDataset ID, enter "natality_regression".
      2. ForData location, you can choose alocationfor the dataset. The default value location isUS multi-region.After a dataset is created, the location cannot be changed.
      3. ForDefault table expiration, choose one of the following options:
         • Never (default): You must delete the table manually.
         • Number of days: The table will be deletedafter the specified number days from its creation time.
      4. ForEncryption, choose one of the following options:
         • Google-owned and Google-managed encryption key (default).
         • Customer-managed key: SeeProtecting data with Cloud KMS keys.
      5. ClickCreate dataset.You cannot add a description or a label when you create a dataset usingthe Web UI. After the dataset is created, you canadd a description,andadd a label.

2. Run a query against the public natality dataset, then save the query results in a new table in your dataset.

   1. Copy and paste the following query into the Query Editor, thenclick Run.

      CREATE OR REPLACE TABLE natality_regression.regression_input asSELECTweight_pounds,mother_age,father_age,gestation_weeks,weight_gain_pounds,apgar_5minFROM`bigquery-public-data.samples.natality`WHEREweight_pounds IS NOT NULLAND mother_age IS NOT NULLAND father_age IS NOT NULLAND gestation_weeks IS NOT NULLAND weight_gain_pounds IS NOT NULLAND apgar_5min IS NOT NULL

   2. After the query completes (in approximately one minute), the resultsare saved as "regression_input" BigQuery tablein thenatality_regression dataset in your project.

      

Python

Before trying this sample, follow thePython setup instructions in theDataproc quickstart using client libraries. For more information, see theDataprocPython API reference documentation.

To authenticate to Dataproc, set up Application Default Credentials. For more information, seeSet up authentication for a local development environment.

1. SeeSetting Up a Python Development Environment for instructions on installing Python and the Google Cloud Client Library forPython (needed to run the code). Installing and using a Pythonvirtualenv is recommended.

2. Copy and paste thenatality_tutorial.py code, below,into apython shell on your local machine.Press the<return> key in the shell to run the code to create a"natality_regression" BigQuery dataset in your defaultGoogle Cloud project with a"regression_input" table that is populated with a subset of the publicnatality data.

   Setting up to run the application code locally:
   • Rungcloud config list project to see the name of yourdefault project.
   • Rungcloud config set projectproject-name to change thedefault project.
   • Rungcloud auth application-default login to setapplication credentials to your user account.
   """Create a Google BigQuery linear regression input table.In the code below, the following actions are taken:* A new dataset is created "natality_regression."* A query is run against the public dataset,    bigquery-public-data.samples.natality, selecting only the data of    interest to the regression, the output of which is stored in a new    "regression_input" table.* The output table is moved over the wire to the user's default project via    the built-in BigQuery Connector for Spark that bridges BigQuery and    Cloud Dataproc."""fromgoogle.cloudimportbigquery# Create a new Google BigQuery client using Google Cloud Platform project# defaults.client=bigquery.Client()# Prepare a reference to a new dataset for storing the query results.dataset_id="natality_regression"dataset_id_full=f"{client.project}.{dataset_id}"dataset=bigquery.Dataset(dataset_id_full)# Create the new BigQuery dataset.dataset=client.create_dataset(dataset)# Configure the query job.job_config=bigquery.QueryJobConfig()# Set the destination table to where you want to store query results.# As of google-cloud-bigquery 1.11.0, a fully qualified table ID can be# used in place of a TableReference.job_config.destination=f"{dataset_id_full}.regression_input"# Set up a query in Standard SQL, which is the default for the BigQuery# Python client library.# The query selects the fields of interest.query="""    SELECT        weight_pounds, mother_age, father_age, gestation_weeks,        weight_gain_pounds, apgar_5min    FROM        `bigquery-public-data.samples.natality`    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"""# Run the query.client.query_and_wait(query,job_config=job_config)# Waits for the query to finish

3. Confirm the creation of thenatality_regression dataset andtheregression_input table.

   

Console

1. Copy and paste the following code into a newnatality_sparkml.pyfile on your local machine.

   """Run a linear regression using Apache Spark ML.In the following PySpark (Spark Python API) code, we take the following actions:  * Load a previously created linear regression (BigQuery) input table    into our Cloud Dataproc Spark cluster as an RDD (Resilient    Distributed Dataset)  * Transform the RDD into a Spark Dataframe  * Vectorize the features on which the model will be trained  * Compute a linear regression using Spark ML"""frompyspark.contextimportSparkContextfrompyspark.ml.linalgimportVectorsfrompyspark.ml.regressionimportLinearRegressionfrompyspark.sql.sessionimportSparkSession# The imports, above, allow us to access SparkML features specific to linear# regression as well as the Vectors types.# Define a function that collects the features of interest# (mother_age, father_age, and gestation_weeks) into a vector.# Package the vector in a tuple containing the label (`weight_pounds`) for that# row.defvector_from_inputs(r):return(r["weight_pounds"],Vectors.dense(float(r["mother_age"]),float(r["father_age"]),float(r["gestation_weeks"]),float(r["weight_gain_pounds"]),float(r["apgar_5min"])))sc=SparkContext()spark=SparkSession(sc)# Read the data from BigQuery as a Spark Dataframe.natality_data=spark.read.format("bigquery").option("table","natality_regression.regression_input").load()# Create a view so that Spark SQL queries can be run against the data.natality_data.createOrReplaceTempView("natality")# As a precaution, run a query in Spark SQL to ensure no NULL values exist.sql_query="""SELECT *from natalitywhere weight_pounds is not nulland mother_age is not nulland father_age is not nulland gestation_weeks is not null"""clean_data=spark.sql(sql_query)# Create an input DataFrame for Spark ML using the above function.training_data=clean_data.rdd.map(vector_from_inputs).toDF(["label","features"])training_data.cache()# Construct a new LinearRegression object and fit the training data.lr=LinearRegression(maxIter=5,regParam=0.2,solver="normal")model=lr.fit(training_data)# Print the model summary.print("Coefficients:"+str(model.coefficients))print("Intercept:"+str(model.intercept))print("R^2:"+str(model.summary.r2))model.summary.residuals.show()

2. Copy the localnatality_sparkml.py file to a Cloud Storagebucket in your project.

   gcloud storage cp natality_sparkml.py gs://bucket-name

   Instead of copying the file to a user bucket in your project, you can copy it to thestaging bucket that Dataproc created when you created your cluster.

3. Run the regression from the DataprocSubmit a job page.

   1. In theMain python file field, insert thegs:// URI ofthe Cloud Storage bucket where your copy of thenatality_sparkml.pyfile is located.

   2. SelectPySpark as theJob type.

   3. Insertgs://spark-lib/bigquery/spark-bigquery-latest_2.12.jarin theJar files field. This makes the spark-bigquery-connector availableto the PySpark application at runtime to allow it to read BigQuerydata into a Spark DataFrame. The 2.12 jar is compatible with Dataproc clusters created withthe 1.5 or later image. If your Dataproc cluster was created with the1.3 or 1.4 image, specify the 2.11 jar instead(gs://spark-lib/bigquery/spark-bigquery-latest_2.11.jar).

   4. Fill in theJob ID,Region, andCluster fields.

   5. ClickSubmit to run the job on your cluster.

When the job completes, the linear regression output model summary appears inthe Dataproc Job details window.Model Summary Terminology:

• R^2: a measure for how much of the "signal" the model captures.
• Residuals: the difference between the prediction of the model and the actual value for each point that was used to fit the model.

gcloud

1. Copy and paste the following code into a newnatality_sparkml.pyfile on your local machine.

   """Run a linear regression using Apache Spark ML.In the following PySpark (Spark Python API) code, we take the following actions:  * Load a previously created linear regression (BigQuery) input table    into our Cloud Dataproc Spark cluster as an RDD (Resilient    Distributed Dataset)  * Transform the RDD into a Spark Dataframe  * Vectorize the features on which the model will be trained  * Compute a linear regression using Spark ML"""frompyspark.contextimportSparkContextfrompyspark.ml.linalgimportVectorsfrompyspark.ml.regressionimportLinearRegressionfrompyspark.sql.sessionimportSparkSession# The imports, above, allow us to access SparkML features specific to linear# regression as well as the Vectors types.# Define a function that collects the features of interest# (mother_age, father_age, and gestation_weeks) into a vector.# Package the vector in a tuple containing the label (`weight_pounds`) for that# row.defvector_from_inputs(r):return(r["weight_pounds"],Vectors.dense(float(r["mother_age"]),float(r["father_age"]),float(r["gestation_weeks"]),float(r["weight_gain_pounds"]),float(r["apgar_5min"])))sc=SparkContext()spark=SparkSession(sc)# Read the data from BigQuery as a Spark Dataframe.natality_data=spark.read.format("bigquery").option("table","natality_regression.regression_input").load()# Create a view so that Spark SQL queries can be run against the data.natality_data.createOrReplaceTempView("natality")# As a precaution, run a query in Spark SQL to ensure no NULL values exist.sql_query="""SELECT *from natalitywhere weight_pounds is not nulland mother_age is not nulland father_age is not nulland gestation_weeks is not null"""clean_data=spark.sql(sql_query)# Create an input DataFrame for Spark ML using the above function.training_data=clean_data.rdd.map(vector_from_inputs).toDF(["label","features"])training_data.cache()# Construct a new LinearRegression object and fit the training data.lr=LinearRegression(maxIter=5,regParam=0.2,solver="normal")model=lr.fit(training_data)# Print the model summary.print("Coefficients:"+str(model.coefficients))print("Intercept:"+str(model.intercept))print("R^2:"+str(model.summary.r2))model.summary.residuals.show()

2. Copy the localnatality_sparkml.py file to a Cloud Storagebucket in your project.

   gcloud storage cp natality_sparkml.py gs://bucket-name

   Instead of copying the file to a user bucket in your project, you can copy it to thestaging bucket that Dataproc created when you created your cluster.

3. Submit the Pyspark job to the Dataproc service by running thegcloud command, shown below, from a terminal window on your local machine.

   1. The--jars flag value makes the spark-bigquery-connector availableto the PySpark jobv at runtime to allow it to readBigQuery data into a Spark DataFrame.

      gcloud dataproc jobs submit pyspark \    gs://your-bucket/natality_sparkml.py \    --cluster=cluster-name \    --region=region \    --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_SCALA_VERSION-CONNECTOR_VERSION.jar

      The 2.12 jar is compatible with Dataproc clusters created withthe 1.5 or later image. If your Dataproc cluster was created with the1.3 or 1.4 image, specify the 2.11 jar instead. SeeMake the connector available to your applicationfor more information.

The linear regression output (model summary) appears inthe terminal window when the job completes.Model Summary Terminology:

• R^2: a measure for how much of the"signal" the model captures.
• Residuals: the differencebetween the prediction of the model and the actual value for each point that wasused to fit the model.

<<< # Print the model summary.... print "Coefficients:" + str(model.coefficients)Coefficients:[0.0166657454602,-0.00296751984046,0.235714392936,0.00213002070133,-0.00048577251587]<<< print "Intercept:" + str(model.intercept)Intercept:-2.26130330748<<< print "R^2:" + str(model.summary.r2)R^2:0.295200579035<<< model.summary.residuals.show()+--------------------+|           residuals|+--------------------+| -0.7234737533344147||  -0.985466980630501|| -0.6669710598385468||  1.4162434829714794||-0.09373154375186754||-0.15461747949235072|| 0.32659061654192545||  1.5053877697929803||  -0.640142797263989||   1.229530260294963||-0.03776160295256...|| -0.5160734239126814|| -1.5165972740062887||  1.3269085258245008||  1.7604670124710626||  1.2348130901905972||   2.318660276655887||  1.0936947030883175||  1.0169768511417363|| -1.7744915698181583|+--------------------+only showing top 20 rows.