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

---

O BigQuery Connector para Apache Spark permite que cientistas de dados mesclem a força do mecanismo SQL escalonável e confiável do BigQuery com os recursos de machine learning do Apache Spark. Neste tutorial, mostramos como usar o Dataproc, o BigQuery e o Apache Spark ML para realizar machine learning em um conjunto de dados.

Objetivos

Usar a regressão linear para criar um modelo de peso de nascimento como uma função de cinco fatores:

• semanas de gestação
• idade da mãe
• idade do pai
• ganho de peso da mãe durante a gravidez
• Pontuação do Apgar

Use as seguintes ferramentas:

• O BigQuery prepara a tabela de entrada de regressão linear, que é gravada no projeto do Google Cloud.
• Python, para consultar e gerenciar dados no BigQuery
• Apache Spark, para acessar a tabela de regressão linear resultante
• Spark ML para criar e avaliar o modelo
• Job do PySpark do Dataproc para invocar funções do Spark ML

Custos

Neste tutorial, usamos os seguintes componentes faturáveis do Google Cloud:

• Compute Engine
• Dataproc
• BigQuery

Para gerar uma estimativa de custo baseada na projeção de uso deste tutorial, use a calculadora de preços. Novos usuários do Google Cloud podem estar qualificados para uma avaliação gratuita.

Antes de começar

Um cluster do Dataproc tem os componentes do Spark, inclusive o Spark ML, instalados. Para configurar um cluster do Dataproc e executar o código deste exemplo, você precisará fazer (ou ter feito) o seguinte:

1. Faça login na sua conta do Google Cloud. Se você começou a usar o Google Cloud agora, crie uma conta para avaliar o desempenho de nossos produtos em situações reais. Clientes novos também recebem US$ 300 em créditos para executar, testar e implantar cargas de trabalho.

2. No console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto do Google Cloud.

   Observação: se você não pretende manter os recursos criados neste procedimento, crie um projeto novo em vez de selecionar um que já existe. Depois de concluir essas etapas, é possível excluir o projeto. Para fazer isso, basta remover todos os recursos associados a ele.

   Acessar o seletor de projetos

3. Ative as APIs Dataproc, BigQuery, Compute Engine.

   Ative as APIs

4. Instale a CLI do Google Cloud.

5. Para inicializar a CLI gcloud, execute o seguinte comando:

   gcloud init

6. No console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto do Google Cloud.

   Observação: se você não pretende manter os recursos criados neste procedimento, crie um projeto novo em vez de selecionar um que já existe. Depois de concluir essas etapas, é possível excluir o projeto. Para fazer isso, basta remover todos os recursos associados a ele.

   Acessar o seletor de projetos

7. Ative as APIs Dataproc, BigQuery, Compute Engine.

   Ative as APIs

8. Instale a CLI do Google Cloud.

9. Para inicializar a CLI gcloud, execute o seguinte comando:

   gcloud init

10. Crie um cluster do Dataproc no projeto. Seu cluster precisa executar uma versão do Dataproc com Spark 2.0 ou superior, incluindo bibliotecas de machine learning.

Criar um subconjunto de dados do BigQuery natality

Nesta seção, você cria um conjunto de dados no projeto e uma tabela no conjunto de dados para que queira copiar um subconjunto de dados da taxa de natalidade do conjunto de dados natality do BigQuery disponível publicamente. Posteriormente, neste tutorial, você usará os dados do subconjunto nesta tabela para prever o peso de nascimento como uma função da idade materna, da idade paterna e das semanas de gestação.

É possível criar o subconjunto de dados usando o console do Google Cloud ou executando um script Python na sua máquina local.

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, faça o seguinte:
      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. Em Criptografia, escolha uma das seguintes opções:
         • Chave 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.
         Não é possível adicionar uma descrição ou um rótulo ao criar um conjunto de dados usando a IU da Web. Após a criação do conjunto de dados, você pode adicionar uma descrição e adicionar um marcador.

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 consulta a seguir no Editor de consultas e clique em "Executar".

      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. Depois que a consulta for concluída (após aproximadamente um minuto), clique em SALVAR RESULTADOS e selecione opções de salvamento para salvar os resultados como uma tabela "regression_input" do BigQuery no conjunto de dados natality_regression no seu projeto.

      

Python

1. Consulte Como configurar um ambiente de desenvolvimento do Python para ver instruções sobre como instalar o Python e a biblioteca de cliente do Google Cloud para Python (necessária 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.

   Configuração para executar o código do aplicativo localmente:

   • Execute gcloud config list project para ver o nome do seu projeto padrão.
   • Execute gcloud config set project project-name para alterar o projeto padrão.
   • Execute gcloud auth application-default login para definir as credenciais do aplicativo para sua conta de usuário.
   samples/snippets/natality_tutorial.py

   Ver no GitHub

   """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.
   query_job = client.query(query, job_config=job_config)
   query_job.result()  # Waits for the query to finish

3. Confirme a criação do conjunto de dados natality_regression e da tabela regression_input.

   

Executar uma regressão linear

Nesta seção, você executará uma regressão linear do PySpark enviando o job ao serviço do Dataproc usando o console do Google Cloud ou executando o comando gcloud a partir de um terminal local.

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 projeto.

   gsutil cp natality_sparkml.py gs://bucket-name
   

   Em vez de copiar o arquivo para um bucket do usuário no projeto, é possível copiá-lo para o bucket de preparo criado pelo Dataproc quando você criou o cluster.

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.

      O jar 2.12 é compatível com clusters do Dataproc criados com a imagem de 1.5 ou posterior. Se o cluster do Dataproc foi criado com a imagem 1.3 ou 1.4, especifique o jar 2.11 (gs://spark-lib/bigquery/spark-bigquery-latest_2.11.jar).

   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 do job do Dataproc.

Terminologia do resumo do modelo:

• R^2: medida para a quantidade de "sinal" que o modelo captura.
• Residuals:a diferença entre a previsão do modelo e o valor real de cada ponto que foi usado para ajustar o modelo.

gcloud CLI

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 projeto.

   gsutil cp natality_sparkml.py gs://bucket-name
   

   Em vez de copiar o arquivo para um bucket do usuário no projeto, é possível copiá-lo para o bucket de preparo criado pelo Dataproc quando você criou o cluster.

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 disponibiliza o job spark-bigquery-connector para o jobv do PySpark no ambiente de execução para permitir a leitura de dados do BigQuery em um DataFrame do Spark.

      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
      

      O jar 2.12 é compatível com clusters do Dataproc criados com a imagem de 1.5 ou posterior. Se o cluster do Dataproc tiver sido criado com a imagem 1.3 ou 1.4, especifique o jar 2.11. Consulte Disponibilizar o conector para seu aplicativo para mais informações.

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

Terminologia do resumo do modelo:

• R^2: medida para a quantidade de "sinal" que o modelo captura.
• Residuals: é a diferença entre a previsão do modelo e o valor real de cada ponto que foi usado para ajustar o modelo.

<<< # 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.

Limpar

Depois de concluir o tutorial, você pode limpar os recursos que criou para que eles parem de usar a cota e gerar cobranças. Nas seções a seguir, você aprenderá a excluir e desativar esses recursos.

Excluir o projeto

O jeito mais fácil de evitar cobranças é excluindo o projeto que você criou para o tutorial.

Para excluir o projeto, faça o seguinte:

Cuidado: excluir um projeto tem os seguintes efeitos:

• Tudo no projeto é excluído. Se você tiver usado um projeto atual para seguir este tutorial, a exclusão dele incluirá quaisquer outras ações feitas no projeto.
• Os IDs do projeto personalizados são perdidos. Ao criar o projeto, você pode ter criado um código do projeto personalizado para ser usado no futuro. Para preservar os URLs que usam o ID do projeto, como um URL appspot.com, exclua recursos específicos do projeto, em vez de excluir o projeto inteiro.

Se você planeja explorar vários tutoriais e guias de início rápido, a reutilização de projetos pode evitar que você exceda os limites da cota do projeto.

1. No Console do Google Cloud, acesse a página Gerenciar recursos.

   Acessar "Gerenciar recursos"

2. Na lista de projetos, selecione o projeto que você quer excluir e clique em Excluir .
3. Na caixa de diálogo, digite o ID do projeto e clique em Encerrar para excluí-lo.

Como excluir o cluster do Dataproc

Consulte Excluir um cluster.

A seguir

• Veja dicas de ajuste de jobs do Spark.