Usar o Dataproc, o BigQuery e o Apache Spark ML para machine learning

Console

1. Crie um conjunto de dados no projeto.

   1. Acesse a IU da Web do BigQuery.
   2. No painel de navegação esquerdo, clique no nome do projeto e em CRIAR CONJUNTO DE DADOS.
   3. Na caixa de diálogo Criar conjunto de dados:
      1. Em ID do conjunto de dados, digite "natality_regression".
      2. Em Local de dados, escolha um local para o conjunto de dados. O local do valor padrão é US multi-region. Depois que um conjunto de dados é criado, o local não pode ser alterado.
      3. Em Validade da tabela padrão, escolha uma das seguintes opções:
         • Nunca (padrão): você deve excluir a tabela manualmente.
         • Número de dias: a tabela será excluída após o número de dias especificado a partir da data de criação.
      4. Para Criptografia, escolha uma das seguintes opções:
         • Chave de propriedade e gerenciada pelo Google (padrão).
         • Chave gerenciada pelo cliente: consulte Como proteger dados com chaves do Cloud KMS.
      5. Clique em Criar conjunto de dados.

2. Execute uma consulta no conjunto de dados de natalidade pública e salve os resultados da consulta em uma nova tabela no conjunto de dados.

   1. Copie e cole a seguinte consulta no Editor de Consultas e clique em Executar.

      CREATE OR REPLACE TABLE natality_regression.regression_input as
      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
      

   2. Após a conclusão da consulta (em aproximadamente um minuto), os resultados são salvas como "regression_input" Tabela do BigQuery no conjunto de dados natality_regression do projeto.

      

Python

Antes de testar este exemplo, siga as instruções de configuração do Python no Guia de início rápido do Dataproc: como usar bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Python do Dataproc.

Para autenticar no Dataproc, configure o Application Default Credentials. Para mais informações, consulte Configurar a autenticação para um ambiente de desenvolvimento local.

1. Consulte Como configurar um ambiente de desenvolvimento em Python para instruções sobre como instalar o Python e a biblioteca de cliente do Google Cloud para Python (necessário para executar o código). É recomendado instalar e usar um virtualenv do Python.

2. Copie e cole o código natality_tutorial.py abaixo em um shell python na máquina local. Pressione o botão <return> no shell para executar o código para criar um conjunto de dados "natality_regression" do BigQuery no seu projeto padrão do Google Cloud com uma tabela "regression_input" preenchida com um subconjunto dos dados natality públicos.

   """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.
   """
   
   from google.cloud import bigquery
   
   # 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. Confirme a criação do conjunto de dados natality_regression e da tabela regression_input.

   

Console

1. Copie e cole o seguinte código em um novo arquivo natality_sparkml.py na máquina local.

   """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
   
   """
   from pyspark.context import SparkContext
   from pyspark.ml.linalg import Vectors
   from pyspark.ml.regression import LinearRegression
   from pyspark.sql.session import SparkSession
   # 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.
   def vector_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 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
   """
   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. Copie o arquivo natality_sparkml.py local para um bucket do Cloud Storage no seu projeto.

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

3. Execute a regressão na página Enviar um job do Dataproc.

   1. No campo Arquivo principal python, insira o URI gs:// do bucket do Cloud Storage no qual sua cópia do arquivo natality_sparkml.py está localizada.

   2. Selecione PySpark como o Tipo de job.

   3. Insira gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar no campo Arquivos Jar. Isso torna o conector spark-bigquery disponível para o aplicativo PySpark no ambiente de execução para permitir a leitura de dados do BigQuery em um Spark DataFrame.

   4. Preencha os campos ID do job, Região e Cluster.

   5. Clique em Enviar para executar o job no cluster.

Quando o job é concluído, o resumo do modelo de saída de regressão linear é exibido na janela de detalhes Job do Dataproc.

gcloud

1. Copie e cole o seguinte código em um novo arquivo natality_sparkml.py na máquina local.

   """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
   
   """
   from pyspark.context import SparkContext
   from pyspark.ml.linalg import Vectors
   from pyspark.ml.regression import LinearRegression
   from pyspark.sql.session import SparkSession
   # 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.
   def vector_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 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
   """
   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. Copie o arquivo natality_sparkml.py local para um bucket do Cloud Storage no seu projeto.

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

3. Envie o job Pyspark para o serviço Dataproc executando o comando gcloud, mostrado abaixo, a partir de uma janela de terminal na máquina local.

   1. O valor da sinalização --jars torna o conector spark-bigquery disponível para o jobv do PySpark no ambiente de execução para permitir a leitura de dados do BigQuery em um 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
      

A saída de regressão linear (resumo do modelo) é exibida na janela do terminal quando o job é concluído.

<<< # Imprime o resumo do modelo.
... imprimir "Coeficientes:" + str(model.coeficientes)
Coefficients:[0.0166657454602,-0.00296751984046,0.235714392936,0.00213002070133,-0.00048577251587]
&lt;&lt;&lt; print "Intercept:" + str(model.intercept)
Interceptação:-2.26130330748
&lt;&lt;&lt; imprimir "R^2:" + str(model.summary.r2)
R^2:0,295200579035
&lt;&lt;&lt; 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|
+--------------------+
mostrando apenas as 20 primeiras linhas.