Usar 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's com os recursos de machine learning do Apache Spark’s Machine Learning capabilities. Neste tutorial, mostramos como usar o Managed Service for Apache Spark, 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 gestação
• escala de Apgar

Usar as seguintes ferramentas:

• BigQuery, para preparar a tabela de entrada de regressão linear, que é gravada no seu Google Cloud projeto
• 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 Managed Service for Apache Spark PySpark, para invocar as funções do ML do Spark

Custos

Neste documento, você vai usar os seguintes componentes faturáveis do Google Cloud:

• Compute Engine
• Managed Service for Apache Spark
• BigQuery

Para gerar uma estimativa de custo baseada na projeção de uso, use a calculadora de preços.

Novos usuários do Google Cloud podem estar qualificados para um teste sem custo financeiro.

Antes de começar

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

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

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Enable the Dataproc, BigQuery, Compute Engine APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Instale a Google Cloud CLI.

Ao usar um provedor de identidade (IdP) externo, primeiro faça login na gcloud CLI com sua identidade federada.

Para inicializar a CLI gcloud, execute o seguinte comando:

gcloud init

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Enable the Dataproc, BigQuery, Compute Engine APIs.

Roles required to enable APIs

To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

Enable the APIs

Instale a Google Cloud CLI.

Ao usar um provedor de identidade (IdP) externo, primeiro faça login na gcloud CLI com sua identidade federada.

Para inicializar a CLI gcloud, execute o seguinte comando:

gcloud init

1. Crie um cluster do Managed Service for Apache Spark no seu projeto. Seu cluster precisa executar uma versão do Managed Service for Apache Spark com Spark 2.0 ou superior, incluindo bibliotecas de machine learning.

Criar um subconjunto de dados natality do BigQuery

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 Google Cloud console ou executando um script Python na 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:
      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:
         • Google-owned and Google-managed encryption key (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 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. Depois que a consulta for concluída (em aproximadamente um minuto), os resultados serão salvos como uma tabela do BigQuery "regression_input" no conjunto de dados natality_regression no seu projeto.

      

Python

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

Para autenticar o Managed Service for Apache Spark, configure o Application Default Credentials. Se quiser mais informações, consulte Configurar a autenticação para um ambiente de desenvolvimento local.

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 a tecla <return> no shell para executar o código para criar um conjunto de dados "natality_regression" do BigQuery no seu projeto padrão com uma tabela "regression_input" preenchida com um subconjunto dos dados públicos.Google Cloud natality

   Como configurar 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.
   """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.

   

Executar uma regressão linear

Nesta seção, você executará uma regressão linear do PySpark enviando o job ao Managed Service for Apache Spark por meio do Google Cloud console ou executando o gcloud comando 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 local natality_sparkml.py para um bucket do Cloud Storage no seu projeto.

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

   Em vez de copiar o arquivo para um grupo de usuários no projeto, é possível copiá-lo para o bucket de preparo criado pelo Managed Service for Apache Spark quando você criou seu cluster.

3. Execute a regressão na página Enviar um job do Managed Service for Apache Spark Submit a job.

   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 Job do Managed Service for Apache Spark.

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

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 local natality_sparkml.py para um bucket do Cloud Storage no seu projeto.

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

   Em vez de copiar o arquivo para um grupo de usuários no projeto, é possível copiá-lo para o bucket de preparo criado pelo Managed Service for Apache Spark quando você criou seu cluster.

3. Envie o job Pyspark para o Managed Service for Apache Spark 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
      

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

Exclua o projeto

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

Para excluir o projeto:

Cuidado: excluir um projeto tem os seguintes efeitos:

• Tudo no projeto é excluído. Se você tiver usado um projeto atual para as tarefas neste documento, a exclusão dele incluirá a exclusão de quaisquer outros trabalhos feitos nele.
• Os IDs do projeto personalizados são perdidos. Ao criar o projeto, você pode ter criado um ID 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árias arquiteturas, tutoriais e guias de início rápido, a reutilização de projetos pode evitar que você exceda limites da cota.

1. No Google Cloud console, 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 Desligar para excluir o projeto.

Excluir o cluster do Managed Service for Apache Spark

Consulte Excluir um cluster.

A seguir

• Veja dicas de ajuste de jobs do Spark.