Usa Dataproc, BigQuery y 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 Dataproc, 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 de 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 de Dataproc para invocar funciones de Spark ML

Costos

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

• Compute Engine
• Dataproc
• 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 califiquen para obtener una prueba gratuita.

Antes de comenzar

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

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

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

   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

3. Enable the Dataproc, BigQuery, Compute Engine APIs.

   Enable the APIs

4. Install the Google Cloud CLI.

5. To initialize the gcloud CLI, run the following command:

   gcloud init

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

   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

7. Enable the Dataproc, BigQuery, Compute Engine APIs.

   Enable the APIs

8. Install the Google Cloud CLI.

9. To initialize the gcloud CLI, run the following command:

   gcloud init

10. Crear un clúster de Dataproc en tu proyecto. Tu clúster debe ejecutar una versión de Dataproc con Spark 2.0 o superior, (incluye las bibliotecas de aprendizaje automático).

Crear un subconjunto de datos de BigQuery natality

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 consola de Google Cloud 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:
         • Clave que es propiedad de Google y está administrada por Google (predeterminada).
         • 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 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 se encuentran en la guía de inicio rápido de Dataproc sobre el uso de bibliotecas cliente. Si deseas obtener más información, consulta la documentación de referencia de la API de Python de Dataproc.

Para autenticarte en Dataproc, configura las credenciales predeterminadas de la aplicación. Si deseas 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 de Google 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 mediante el envío del trabajo al servicio de Dataproc con la consola de Google Cloud o a través del 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ó Dataproc cuando creaste el clúster.

3. Ejecuta la regresión desde la página Enviar un trabajo de Dataproc.

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

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ó Dataproc cuando creaste el clúster.

3. Envía el trabajo de Pyspark al servicio de Dataproc mediante la ejecución del 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|
+--------------------+
solo se muestran las 20 filas principales.

  

Limpia

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 en el proyecto. 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 ID personalizados de proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID del proyecto personalizado que desees utilizar en el futuro. Para conservar las URL que utilizan el ID del proyecto, como una URL appspot.com, borra los recursos seleccionados dentro del proyecto en lugar de borrar todo el proyecto.

Si planeas explorar varias infraestructuras, instructivos y guías de inicio rápido la reutilización de proyectos puede ayudarte a evitar exceder los límites de las cuotas del proyecto.

1. In the Google Cloud console, go to the Manage resources page.

   Go to Manage resources

2. In the project list, select the project that you want to delete, and then click Delete.
3. In the dialog, type the project ID, and then click Shut down to delete the project.

Cómo borrar el clúster de Dataproc

Consulta Borra un clúster.

¿Qué sigue?

• Consulta las sugerencias de ajuste de trabajo de Spark