Usa Apache Spark ML para el aprendizaje automático

El Conector de BigQuery para Apache Spark permite que los científicos de datos combinen el poder del motor SQL perfectamente escalable de BigQuery con las capacidades del aprendizaje automático de Apache Spark. En este instructivo, mostramos cómo usar el servicio administrado para Apache Spark, BigQuery y Apache Spark ML para realizar el aprendizaje automático en un conjunto de datos.

Objetivos

Usa la regresión lineal para compilar un modelo de peso de un recién nacido como una función de cinco factores:

• semanas de gestación
• Edad de la madre
• edad del padre
• aumento de peso de la madre durante el embarazo
• puntuación de Apgar

Usa las siguientes herramientas:

• BigQuery, para preparar la tabla de entrada de regresión lineal, que se escribe en tu proyecto Google Cloud
• Python, para consultar y administrar datos en BigQuery
• Apache Spark, para acceder a la tabla de regresión lineal resultante
• Spark ML, para compilar y evaluar el modelo
• Trabajo de PySpark del servicio administrado para Apache Spark, para invocar funciones de Spark ML

Costos

En este documento, usarás los siguientes componentes facturables de Google Cloud:

• Compute Engine
• Managed Service for Apache Spark
• BigQuery

Para generar una estimación de costos en función del uso previsto, usa la calculadora de precios.

Es posible que los usuarios nuevos de Google Cloud cumplan con los requisitos para acceder a una prueba gratuita.

Antes de comenzar

Un clúster del servicio administrado para Apache Spark tiene los componentes de Spark, incluido Spark ML, instalados. Para configurar un clúster de Managed Service for Apache Spark y ejecutar el código en este ejemplo, necesitarás hacer (o haber hecho) lo siguiente:

Accede a tu cuenta de Google Cloud . Si eres nuevo en Google Cloud, crea una cuenta para evaluar el rendimiento de nuestros productos en situaciones reales. Los clientes nuevos también obtienen $300 en créditos gratuitos para ejecutar, probar y, además, implementar cargas de trabajo.

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

Instala Google Cloud CLI.

Si usas un proveedor de identidad externo (IdP), primero debes acceder a la gcloud CLI con tu identidad federada.

Para inicializar gcloud CLI, ejecuta el siguiente 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

Instala Google Cloud CLI.

Si usas un proveedor de identidad externo (IdP), primero debes acceder a la gcloud CLI con tu identidad federada.

Para inicializar gcloud CLI, ejecuta el siguiente comando:

gcloud init

1. Crea un clúster de Managed Service para Apache Spark en tu proyecto. Tu clúster debe ejecutar una versión del servicio administrado para Apache Spark con Spark 2.0 o superior (incluye las bibliotecas de aprendizaje automático).

Crea un subconjunto de datos de natalidad de BigQuery

En esta sección, creas un conjunto de datos en tu proyecto y luego creas una tabla en el conjunto de datos al que copias el subconjunto de datos de tasa de natalidad del conjunto de datos de BigQuery de natalidad disponible públicamente. Más adelante en este instructivo, usarás los datos del subconjunto en esta tabla para predecir en peso del recién nacido en función de la edad de la madre, del padre y las semanas de gestación.

Puedes crear el subconjunto de datos con la Google Cloud consola o ejecutar una secuencia de comandos de Python en tu máquina local.

Console

1. Crea un conjunto de datos en tu proyecto.

   1. Ve a la IU web de BigQuery.
   2. En el panel de navegación izquierdo, haz clic en el nombre del proyecto y, luego, en CREAR CONJUNTO DE DATOS.
   3. En el diálogo Crear conjunto de datos, realice lo siguiente:
      1. Para el ID de conjunto de datos (Dataset ID), ingresa "natality_regression".
      2. En Ubicación de los datos, puedes seleccionar una ubicación para el conjunto de datos. La ubicación predeterminada es US multi-region. Una vez que se crea un conjunto de datos, la ubicación no se puede cambiar.
      3. En Vencimiento predeterminado de la tabla, selecciona una de las siguientes opciones:
         • Nunca (predeterminado): Debes borrar la tabla manualmente.
         • Cantidad de días: La tabla se borrará después de la cantidad de días especificada a partir del momento de su creación.
      4. En Encriptación, elige una de las siguientes opciones:
         • Google-owned and Google-managed encryption key (predeterminado)
         • Clave administrada por el cliente: Consulta la documentación sobre cómo proteger datos con claves de Cloud KMS.
      5. Haga clic en Crear conjunto de datos.
         No puedes agregar una descripción o una etiqueta cuando creas un conjunto de datos con la IU web. Una vez que creas el conjunto de datos, puedes agregar una descripción y agregar una etiqueta.

2. Ejecuta una consulta en el conjunto de datos público de natalidad y, luego, guarda los resultados de la consulta en una tabla nueva en tu conjunto de datos.

   1. Copia y pega la siguiente consulta en el Editor de consultas y, luego, haz clic en Ejecutar.

      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. Cuando la consulta finalice (en aproximadamente un minuto), los resultados se guardarán como una tabla "regression_input" de BigQuery en el conjunto de datos natality_regression de tu proyecto.

      

Python

Antes de probar este ejemplo, sigue las instrucciones de configuración para Python que encontrarás en la guía de inicio rápido de Managed Service for Apache Spark sobre cómo usar bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Managed Service for Apache Spark Python.

Para autenticarte en el servicio administrado para Apache Spark, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura la autenticación para un entorno de desarrollo local.

1. Consulta la documentación sobre cómo configurar el entorno de desarrollo de Python para obtener instrucciones sobre cómo instalar Python y la Biblioteca cliente de Google Cloud para Python (necesaria para ejecutar el código). Se recomienda instalar y usar un virtualenv de Python.

2. Copia y pega el código natality_tutorial.py que aparece a continuación en un shell python en tu máquina local. Presiona la tecla <return> en el shell para ejecutar el código y crear un conjunto de datos de BigQuery "natality_regression" en tu proyecto predeterminado deGoogle Cloud con una tabla "regression_input" que se propaga con un subconjunto de los datos natality públicos.

   Configuración para ejecutar el código de la aplicación de forma local:

   • Ejecuta gcloud config list project para ver el nombre de tu proyecto predeterminado.
   • Ejecuta gcloud config set project project-name para cambiar el proyecto predeterminado.
   • Ejecuta gcloud auth application-default login para establecer las credenciales de la aplicación en tu cuenta de usuario.
   """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. Confirma la creación del conjunto de datos natality_regression y la tabla regression_input.

   

Ejecuta una regresión lineal

En esta sección, ejecutarás una regresión lineal de PySpark enviando el trabajo a Managed Service for Apache Spark con la consola de Google Cloud o ejecutando el comando gcloud desde una terminal local.

Console

1. Copia y pega el siguiente código en un archivo natality_sparkml.py nuevo en tu 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. Copia el archivo natality_sparkml.py local a un bucket de Cloud Storage en tu proyecto.

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

   En lugar de copiar el archivo en un bucket de usuario de tu proyecto, puedes copiarlo en el bucket de etapa de pruebas que creó el servicio administrado para Apache Spark cuando creaste tu clúster.

3. Ejecuta la regresión desde la página Enviar un trabajo del servicio administrado para Apache Spark.

   1. En el campo Archivo principal de Python, inserta la URI gs:// del depósito de Cloud Storage donde se encuentra la copia del archivo natality_sparkml.py.

   2. Selecciona PySpark como el Tipo de trabajo.

   3. Inserta gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar en el campo Archivos jar. Esto hace que spark-bigquery-connector esté disponible para la aplicación PySpark en el tiempo de ejecución a fin de permitirle leer datos de BigQuery en un DataFrame de Spark.

      El jar 2.12 es compatible con los clústeres de Dataproc creados con la imagen 1.5 o posterior. Si tu clúster de Dataproc se creó con la imagen 1.3 o 1.4, especifica el jar 2.11 en su lugar (gs://spark-lib/bigquery/spark-bigquery-latest_2.11.jar).

   4. Completa los campos ID de trabajo, Región y Clúster.

   5. Haz clic en Enviar para ejecutar el trabajo en tu clúster.

Cuando el trabajo finalice, el resumen del modelo del resultado de la regresión lineal aparecerá en la ventana de detalles del trabajo del servicio administrado de Apache Spark.

Terminología del resumen del modelo:

• R^2: una medida para la cantidad de "señal" que captura el modelo
• Residuos: la diferencia entre la predicción del modelo y el valor real para cada punto que se usó para ajustar el modelo

gcloud

1. Copia y pega el siguiente código en un archivo natality_sparkml.py nuevo en tu 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. Copia el archivo natality_sparkml.py local a un bucket de Cloud Storage en tu proyecto.

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

   En lugar de copiar el archivo en un bucket de usuario de tu proyecto, puedes copiarlo en el bucket de etapa de pruebas que creó el Servicio administrado para Apache Spark cuando creaste tu clúster.

3. Envía el trabajo de Pyspark al servicio administrado para Apache Spark ejecutando el comando gcloud, que se muestra a continuación, desde una ventana de la terminal en tu máquina local.

   1. El valor de la marca --jars hace que spark-bigquery-connector esté disponible para el trabajo de PySpark en el entorno de ejecución a fin de que pueda leer datos de BigQuery en un DataFrame de 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
      

      El jar 2.12 es compatible con los clústeres de Dataproc creados con la imagen 1.5 o posterior. Si tu clúster de Dataproc se creó con la imagen 1.3 o 1.4, especifica el jar 2.11 en su lugar. Consulta Cómo hacer que el conector esté disponible para tu aplicación para obtener más información.

Cuando finaliza el trabajo, el resultado de la regresión lineal (resumen del modelo) aparece en la ventana de la terminal.

Terminología del resumen del modelo:

• R^2: una medida para la cantidad de "señal" que captura el modelo
• Residuos: la diferencia entre la predicción del modelo y el valor real para cada punto que se usó para ajustar el modelo

<<< # Imprime el resumen del modelo.
… 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.

  

Realiza una limpieza

Una vez que completes el instructivo, puedes limpiar los recursos que creaste para que dejen de usar la cuota y generar cargos. En las siguientes secciones, se describe cómo borrar o desactivar estos recursos.

Borra el proyecto

La manera más fácil de eliminar la facturación es borrar el proyecto que creaste para el instructivo.

Para borrar el proyecto, sigue estos pasos:

Precaución: Borrar un proyecto tiene las siguientes consecuencias:

• Se borra todo su contenido. Si usaste un proyecto existente para las tareas de este documento, cuando lo borres, también se borrará cualquier otro trabajo que hayas realizado en el proyecto.
• Se pierden los IDs personalizados de los proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID personalizado del proyecto que desees utilizar en el futuro. Para conservar las URLs que usan el ID del proyecto, como una URL appspot.com, borra los recursos seleccionados dentro del proyecto, en lugar de todo el proyecto.

Si planeas explorar varios instructivos, arquitecturas o guías de inicio rápido, reutilizar proyectos puede ser útil para no exceder los límites de cuota de los proyectos.

1. En la Google Cloud consola, ve a la página Administrar recursos.

   Ir a Administrar recursos

2. En la lista de proyectos, elige el proyecto que quieres borrar y haz clic en Borrar.
3. En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrar el proyecto.

Borra el clúster de Managed Service for Apache Spark

Consulta Borra un clúster.

¿Qué sigue?

• Consulta las sugerencias de ajuste de trabajo de Spark