Utiliser Apache Spark ML pour le machine learning

Le connecteur BigQuery pour Apache Spark permet aux data scientists de combiner la puissance du moteur SQL évolutif et fluide avec les fonctionnalités de machine learning Apache Spark. Dans ce tutoriel, nous allons vous montrer comment utiliser Managed Service pour Apache Spark, BigQuery et Apache Spark ML pour effectuer du machine learning sur un ensemble de données.

Objectifs

Utilisez la régression linéaire pour créer un modèle de poids de naissance en fonction des cinq facteurs suivants :

• Semaines de gestation
• l'âge de la mère
• âge du père
• la prise de poids de la mère pendant la grossesse ;
• Score d'Apgar

Utilisez les outils suivants :

• BigQuery, pour préparer la table d'entrée de régression linéaire, qui est écrite dans votre projet Google Cloud
• Python, pour interroger et gérer les données dans BigQuery
• Apache Spark, pour accéder à la table de régression linéaire résultante
• Spark ML, pour créer et évaluer le modèle
• Job PySpark Managed Service pour Apache Spark, pour appeler les fonctions Spark ML

Coûts

Dans ce document, vous utilisez les composants facturables suivants de Google Cloud :

• Compute Engine
• Managed Service for Apache Spark
• BigQuery

Pour obtenir une estimation des coûts en fonction de votre utilisation prévue, utilisez le simulateur de coût.

Les nouveaux utilisateurs de Google Cloud peuvent bénéficier d'un essai sans frais.

Avant de commencer

Un cluster Managed Service pour Apache Spark comporte les composants Spark, y compris Spark ML. Pour configurer un cluster Managed Service pour Apache Spark et exécuter le code de cet exemple, vous devez effectuer (ou avoir effectué) les opérations suivantes :

Connectez-vous à votre compte Google Cloud . Si vous débutez sur Google Cloud, créez un compte pour évaluer les performances de nos produits en conditions réelles. Les nouveaux clients bénéficient également de 300 $ de crédits sans frais pour exécuter, tester et déployer des charges de travail.

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, if any are not already enabled.

Roles required to enable APIs

To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

Enable the APIs

Install the Google Cloud CLI.

If you're using an external identity provider (IdP), you must first sign in to the gcloud CLI with your federated identity.

To initialize the gcloud CLI, run the following command:

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, if any are not already enabled.

Roles required to enable APIs

To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

Enable the APIs

Install the Google Cloud CLI.

If you're using an external identity provider (IdP), you must first sign in to the gcloud CLI with your federated identity.

To initialize the gcloud CLI, run the following command:

gcloud init

1. Créez un cluster Managed Service pour Apache Spark dans votre projet. Votre cluster doit exécuter une version Managed Service pour Apache Spark avec Spark 2.0 ou version ultérieure (y compris les bibliothèques de machine learning).

Créer un sous-ensemble de données BigQuery sur la natalité

Dans cette section, vous créez un ensemble de données dans votre projet, puis une table dans cet ensemble de données. Copiez un sous-ensemble de données dans la table regroupant des données de taux de natalité, issues de l'ensemble de données publiques natality BigQuery. Dans la suite de ce tutoriel, vous utiliserez les données du sous-ensemble de cette table pour prédire le poids de naissance en fonction de l'âge de la mère, de l'âge du père et du nombre de semaines de gestation.

Vous pouvez créer le sous-ensemble de données à l'aide de la console Google Cloud ou en exécutant un script Python sur votre machine locale.

Console

1. Créez un ensemble de données dans votre projet.

   1. Accédez à l'interface utilisateur Web de BigQuery.
   2. Dans le volet de navigation de gauche, cliquez sur le nom de votre projet, puis sur CREATE DATASET (Créer un ensemble de données).
   3. Dans la boîte de dialogue Create Dataset (Créer un ensemble de données) :
      1. Dans le champ Dataset ID (ID de l'ensemble de données), saisissez "natality_regression".
      2. Pour Emplacement des données, vous pouvez choisir un emplacement pour l'ensemble de données. La valeur par défaut est US multi-region. Une fois l'ensemble de données créé, l'emplacement ne peut plus être modifié.
      3. Pour indiquer le délai d'expiration par défaut de la table, choisissez l'une des options suivantes :
         • Jamais (par défaut) : vous devez supprimer la table manuellement.
         • Nombre de jours : la table est supprimée une fois le nombre de jours spécifié écoulé, à compter de sa date de création.
      4. Dans le champ Chiffrement, choisissez l'une des options suivantes :
         • Google-owned and Google-managed encryption key (par défaut)
         • Clé gérée par le client : consultez la page Protéger des données avec des clés Cloud KMS
      5. Cliquez sur Créer un ensemble de données.
         Vous ne pouvez pas ajouter de description ni d'étiquette lorsque vous créez un ensemble de données à l'aide de l'interface utilisateur Web. Une fois l'ensemble de données créé, vous pouvez ajouter une description et ajouter une étiquette.

2. Exécutez une requête sur l'ensemble de données public "natality", puis enregistrez les résultats de la requête dans une nouvelle table de l'ensemble de données.

   1. Copiez et collez la requête suivante dans l'éditeur de requête, puis cliquez sur "Exécuter".

      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. Une fois la requête terminée (en une minute environ), les résultats sont enregistrés en tant que table BigQuery "regression_input" dans l'ensemble de données natality_regression de votre projet.

      

Python

Avant d'essayer cet exemple, suivez les instructions de configuration pour Python décrites dans le guide de démarrage rapide de Managed Service pour Apache Spark à l'aide des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence de l'API Python Managed Service pour Apache Spark.

Pour vous authentifier auprès de Managed Service pour Apache Spark, configurez les Identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer l'authentification pour un environnement de développement local.

1. Pour savoir comment installer Python et la bibliothèque cliente Google Cloud pour Python (nécessaire pour exécuter le code), consultez Configurer un environnement de développement Python. Il est recommandé d'installer et d'utiliser un environnement virtualenv Python.

2. Copiez et collez le code natality_tutorial.py ci-dessous dans une interface système python sur votre machine locale. Appuyez sur la touche <return> dans le shell pour exécuter le code permettant de créer un ensemble de données BigQuery "natality_regression" dans votre projetGoogle Cloud par défaut, avec une table "regression_input" qui est renseignée avec un sous-ensemble des données publiques natality.

   La configuration suivante permet d'exécuter le code d'application localement :

   • Exécutez gcloud config list project pour afficher le nom de votre projet par défaut.
   • Exécutez gcloud config set project project-name pour modifier le projet par défaut.
   • Exécutez gcloud auth application-default login pour définir les identifiants de l'application sur votre compte utilisateur.
   """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. Confirmez la création de l'ensemble de données natality_regression et de la table regression_input.

   

Exécuter une régression linéaire

Dans cette section, vous allez exécuter une régression linéaire PySpark en envoyant le job à Managed Service pour Apache Spark à l'aide de la console Google Cloud ou en exécutant la commande gcloud à partir d'un terminal local.

Console

1. Copiez et collez le code suivant dans un nouveau fichier natality_sparkml.py sur votre machine locale.

   """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. Copiez le fichier natality_sparkml.py local dans un bucket Cloud Storage de votre projet.

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

   Au lieu de copier le fichier dans un bucket utilisateur de votre projet, vous pouvez le copier dans le bucket intermédiaire que Managed Service pour Apache Spark a créé lorsque vous avez créé votre cluster.

3. Exécutez la régression à partir de la page Envoyer un job de Managed Service pour Apache Spark.

   1. Dans le champ Fichier Python principal, insérez l'URI gs:// du bucket Cloud Storage dans lequel se trouve votre copie du fichier natality_sparkml.py.

   2. Sélectionnez PySpark comme type de tâche.

   3. Insérez gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar dans le champ Fichiers JAR. Le connecteur spark-bigquery-connector est ainsi disponible pour l'application PySpark au moment de l'exécution, ce qui lui permet de lire les données BigQuery dans un DataFrame Spark.

      Le fichier jar 2.12 est compatible avec les clusters Dataproc créés avec l'image 1.5 ou ultérieure. Si votre cluster Dataproc a été créé avec l'image 1.3 ou 1.4, spécifiez plutôt le fichier jar 2.11 (gs://spark-lib/bigquery/spark-bigquery-latest_2.11.jar).

   4. Renseignez les champs ID du job, Région et Cluster.

   5. Cliquez sur Envoyer pour exécuter la tâche sur votre cluster.

Une fois le job terminé, le récapitulatif du modèle de sortie de régression linéaire s'affiche dans la fenêtre "Détails du job Managed Service pour Apache Spark".

Terminologie du résumé de modèle :

• R^2 : unité de mesure de la quantité de "signaux" enregistrés par le modèle
• Résiduels : différence entre la prédiction du modèle et la valeur réelle pour chaque point permettant de correspondre au modèle

gcloud

1. Copiez et collez le code suivant dans un nouveau fichier natality_sparkml.py sur votre machine locale.

   """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. Copiez le fichier natality_sparkml.py local dans un bucket Cloud Storage de votre projet.

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

   Au lieu de copier le fichier dans un bucket utilisateur de votre projet, vous pouvez le copier dans le bucket intermédiaire que Managed Service pour Apache Spark a créé lorsque vous avez créé votre cluster.

3. Envoyez le job Pyspark à Managed Service pour Apache Spark en exécutant la commande gcloud ci-dessous à partir d'une fenêtre de terminal sur votre machine locale.

   1. La valeur du flag --jars rend le spark-bigquery-connector disponible pour le job PySpark au moment de l'exécution, ce qui lui permet de lire les données BigQuery dans un DataFrame 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
      

      Le fichier JAR 2.12 est compatible avec les clusters Dataproc créés avec l'image 1.5 ou une version ultérieure. Si votre cluster Dataproc a été créé avec l'image 1.3 ou 1.4, spécifiez plutôt le fichier JAR 2.11. Pour en savoir plus, consultez Rendre le connecteur disponible pour votre application.

Le résultat de la régression linéaire (résumé du modèle) s'affiche dans la fenêtre du terminal une fois le job terminé.

Terminologie du résumé de modèle :

• R ^ 2 : unité de mesure de la quantité de "signaux" enregistrés par le modèle.
• Résiduels : différence entre la prédiction du modèle et la valeur réelle pour chaque point permettant de correspondre au modèle.

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

  

Effectuer un nettoyage

Une fois le tutoriel terminé, vous pouvez procéder au nettoyage des ressources que vous avez créées afin qu'elles ne soient plus comptabilisées dans votre quota et qu'elles ne vous soient plus facturées. Dans les sections suivantes, nous allons voir comment supprimer ou désactiver ces ressources.

Supprimer le projet

Le moyen le plus simple d'empêcher la facturation est de supprimer le projet que vous avez créé pour ce tutoriel.

Pour supprimer le projet :

Attention : La suppression d'un projet entraîne les effets décrits ci-dessous :

• Tout le contenu du projet est supprimé. Si vous avez utilisé un projet existant pour les tâches décrites dans ce document et que vous le supprimez, vous supprimerez également tout autre travail effectué dans le projet.
• Les ID de projets personnalisés sont perdus. Lorsque vous avez créé ce projet, vous avez peut-être créé un ID de projet personnalisé que vous souhaitez utiliser à l'avenir. Pour conserver les URL qui utilisent l'ID de projet, telle qu'une URL appspot.com, supprimez les ressources sélectionnées dans le projet au lieu de supprimer l'ensemble du projet.

Si vous envisagez d'explorer plusieurs architectures, tutoriels et guides de démarrage rapide, réutiliser des projets peut vous aider à ne pas dépasser les limites de quotas des projets.

1. Dans la console Google Cloud , accédez à la page Gérer les ressources.

   Accéder à la page "Gérer les ressources"

2. Dans la liste des projets, sélectionnez le projet que vous souhaitez supprimer, puis cliquez sur Supprimer.
3. Dans la boîte de dialogue, saisissez l'ID du projet, puis cliquez sur Arrêter pour supprimer le projet.

Supprimer le cluster Managed Service pour Apache Spark

Consultez la section Supprimer un cluster.

Étape suivante

• Consultez Conseils pour optimiser les jobs Spark.