Prédiction avec un pipeline scikit-learn personnalisé

Logo Colab Exécuter ce tutoriel sous forme de notebook dans Colab Logo GitHub Afficher le notebook sur GitHub

Ce tutoriel explique comment utiliser AI Platform Prediction pour déployer un pipeline scikit-learn qui utilise des transformateurs personnalisés.

Les pipelines scikit-learn vous permettent de composer plusieurs Estimators. Par exemple, vous pouvez utiliser des transformateurs pour prétraiter les données et les transmettre après transformation à un classificateur. La plate-forme scikit-learn fournit de nombreux transformateurs dans le package sklearn.

Vous pouvez également utiliser les classes scikit-learn FunctionTransformer ou TransformerMixin pour créer votre propre transformateur personnalisé. Si vous souhaitez déployer un pipeline utilisant des transformateurs personnalisés pour AI Platform Prediction, vous devez fournir ce code à AI Platform Prediction comme package de distribution source.

Ce tutoriel présente un exemple de problème lié aux données de recensement pour vous guider lors des étapes suivantes :

  • Entraînement d'un pipeline scikit-learn avec des transformateurs personnalisés sur le service d'entraînement AI Platform
  • Déploiement du pipeline entraîné et de votre code personnalisé sur le service de prédiction AI Platform
  • Envoi des requêtes de prédiction issues de ce déploiement

Ensemble de données

Ce tutoriel utilise un ensemble de données sur le revenu obtenues par recensement aux États-Unis, fournies par l'UCI Machine Learning Repository (dépôt de machine learning de l'Université de Californie à Irvine). Cet ensemble de données contient des informations sur les personnes issues d'une base de données du recensement 1994, comprenant des caractéristiques d'âge, de niveau d'études, d'état civil, de profession et indiquant si le revenu de la personne est supérieur à 50 000 dollars par an.

Les données utilisées dans ce tutoriel sont disponibles dans un bucket Cloud Storage public : gs://cloud-samples-data/ai-platform/sklearn/census_data/

Objectif

L'objectif est d'entraîner un pipeline scikit-learn, qui permet de prédire si un individu gagne plus de 50 000 $ par an (étiquette cible) à partir d'autres informations issues du recensement sur la personne (caractéristiques).

Ce tutoriel porte davantage sur l'utilisation de ce modèle avec AI Platform Prediction que sur la conception du modèle. Toutefois, lors de la création de systèmes de machine learning, il est important de considérer les problèmes potentiels et les conséquences inattendues qu'impliquent de tels modèles. Consultez l'exercice du cours d'initiation au machine learning relatif au principe d'équité pour en savoir plus sur les sources de biais dans l'ensemble de données du recensement, ainsi que sur le principe d'équité dans le domaine du machine learning en général.

Coûts

Ce tutoriel utilise des composants facturables de Google Cloud :

  • AI Platform Training
  • AI Platform Prediction
  • Cloud Storage

Découvrez les tarifs d'AI Platform Training, les tarifs d'AI Platform Prediction et les tarifs de Cloud Storage, puis utilisez le simulateur de coûts pour générer une estimation des coûts en fonction de votre utilisation prévue.

Avant de commencer

Vous devez effectuer plusieurs opérations avant de pouvoir entraîner et déployer un modèle sur AI Platform Prediction :

  • Configurer l'environnement de développement local
  • Configurer un projet Google Cloud avec la facturation et les API nécessaires activées
  • Créer un bucket Cloud Storage pour stocker le package d'entraînement et le modèle entraîné

Configurer l'environnement de développement local

Pour suivre ce tutoriel, vous avez besoin des éléments suivants :

  • Python 3
  • virtualenv
  • Le SDK Google Cloud

Le guide Google Cloud Configurer un environnement de développement Python fournit des instructions détaillées permettant de répondre à ces exigences. Les étapes suivantes présentent un condensé de ces instructions :

  1. Installez Python 3.

  2. Installez virtualenv et créez un environnement virtuel utilisant Python 3.

  3. Activez cet environnement.

  4. Effectuez les étapes de la section suivante pour installer le SDK Cloud.

Configurer un projet Google Cloud

  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.

    Go to project selector

  3. Make sure that billing is enabled for your Google Cloud project.

  4. Enable the AI Platform Training & Prediction and Compute Engine APIs.

    Enable the APIs

  5. Install the Google Cloud CLI.

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

    gcloud init

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

    Go to project selector

  8. Make sure that billing is enabled for your Google Cloud project.

  9. Enable the AI Platform Training & Prediction and Compute Engine APIs.

    Enable the APIs

  10. Install the Google Cloud CLI.

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

    gcloud init

Authentifier le compte GCP

Pour configurer l'authentification, vous devez créer une clé de compte de service et définir une variable d'environnement pour le chemin d'accès à la clé de compte de service.

  1. Créez un compte de service :

    1. Dans la console Google Cloud, accédez à la page Créer un compte de service.

      Accéder à la page "Créer un compte de service"

    2. Dans le champ Nom du compte de service, saisissez un nom.
    3. Facultatif : Dans le champ Description du compte de service, saisissez une description.
    4. Cliquez sur Créer.
    5. Cliquez sur le champ Sélectionner un rôle. Sous Tous les rôles, sélectionnez AI Platform > Administrateur AI Platform.
    6. Cliquez sur Ajouter un autre rôle.

    7. Cliquez sur le champ Sélectionner un rôle. Sous Tous les rôles, sélectionnez Stockage > Administrateur des objets de l'espace de stockage.

    8. Cliquez sur Terminé pour créer le compte de service.

      Ne fermez pas la fenêtre de votre navigateur. Vous en aurez besoin lors de la tâche suivante.

  2. Créez une clé de compte de service pour l'authentification :

    1. Dans la console Google Cloud, cliquez sur l'adresse e-mail du compte de service que vous avez créé.
    2. Cliquez sur Keys (Clés).
    3. Cliquez sur AJOUTER UNE CLÉ -> Créer une clé.
    4. Cliquez sur Créer. Un fichier de clé JSON est téléchargé sur votre ordinateur.
    5. Cliquez sur Close (Fermer).

  3. Définissez la variable d'environnement GOOGLE_APPLICATION_CREDENTIALS de façon à pointer vers le chemin du fichier JSON contenant la clé de votre compte de service. Cette variable ne s'applique qu'à la session d'interface système actuelle. Par conséquent, si vous ouvrez une nouvelle session, vous devez la définir à nouveau.

Créer un bucket Cloud Storage

Ce tutoriel utilise Cloud Storage de plusieurs manières :

  • Lorsque vous envoyez une tâche d'entraînement à l'aide du SDK Cloud, vous importez un package Python contenant votre code d'entraînement dans un bucket Cloud Storage. AI Platform Training exécute le code de ce package.

  • Dans ce tutoriel, le modèle entraîné résultant de la tâche est enregistré par AI Platform Training dans le même bucket.

  • Pour déployer votre pipeline scikit-learn qui utilise le code personnalisé pour AI Platform Prediction, vous devez importer les transformateurs personnalisés que votre pipeline utilise dans Cloud Storage.

Lors de la création de la ressource de version AI Platform Prediction qui diffuse les prédictions, vous fournissez le pipeline scikit-learn entraîné, ainsi que votre code personnalisé en tant qu'URI Cloud Storage.

Définissez le nom du bucket Cloud Storage en tant que variable d'environnement. Ce nom doit être unique dans tous les buckets Cloud Storage :

BUCKET_NAME="your-bucket-name"

Sélectionnez une région dans laquelle AI Platform Training et AI Platform Prediction sont disponibles, et créez une autre variable d'environnement. Exemple :

REGION="us-central1"

Créez le bucket Cloud Storage dans cette région, puis utilisez cette même région pour l'entraînement et les prédictions. Exécutez la commande suivante pour créer le bucket s'il n'existe pas déjà :

gcloud storage buckets create gs://$BUCKET_NAME --location=$REGION

Créer une application d'entraînement et un code de pipeline personnalisé

Créez une application pour entraîner un pipeline scikit-learn avec les données de recensement. Dans ce tutoriel, le package d'entraînement contient également le code personnalisé utilisé par le pipeline entraîné lors de la prédiction. Ce modèle est utile, car les pipelines sont généralement conçus pour utiliser les mêmes transformateurs lors des opérations d'entraînement et de prédiction.

Procédez comme suit pour créer un répertoire contenant trois fichiers correspondant à la structure suivante :

census_package/
    __init__.py
    my_pipeline.py
    train.py

Commencez par créer le répertoire vide census_package/ :

mkdir census_package

Dans le répertoire census_package/, créez un fichier vierge nommé __init__.py :

touch ./census_package/__init__.py

Cette opération permet d'importer census_package/ en tant que package dans Python.

Créer des transformateurs personnalisés

La plate-forme scikit-learn fournit de nombreux transformateurs que vous pouvez utiliser dans le cadre d'un pipeline, mais elle vous permet également de définir vos propres transformateurs personnalisés. Ces transformateurs peuvent même apprendre un état enregistré pendant l'entraînement, qui sera utilisé plus tard au cours de la prédiction.

Étendez sklearn.base.TransformerMixin pour définir trois transformateurs:

  • PositionalSelector : à partir d'une liste d'indices C et d'une matrice M, ce transformateur renvoie une matrice contenant le sous-ensemble de colonnes de M correspondant à C.

  • StripString : à partir d'une matrice de chaînes, ce transformateur supprime les espaces de chaque chaîne.

  • SimpleOneHotEncoder: encodeur one-hot simple, qui peut être appliqué à une matrice de chaînes.

Pour ce faire, copiez le code suivant dans un fichier nommé census_package/my_pipeline.py.

import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin


class PositionalSelector(BaseEstimator, TransformerMixin):
    def __init__(self, positions):
        self.positions = positions

    def fit(self, X, y=None):
        return self

    def transform(self, X):
        return np.array(X)[:, self.positions]


class StripString(BaseEstimator, TransformerMixin):
    def fit(self, X, y=None):
        return self

    def transform(self, X):
        strip = np.vectorize(str.strip)
        return strip(np.array(X))


class SimpleOneHotEncoder(BaseEstimator, TransformerMixin):
    def fit(self, X, y=None):
        self.values = []
        for c in range(X.shape[1]):
            Y = X[:, c]
            values = {v: i for i, v in enumerate(np.unique(Y))}
            self.values.append(values)
        return self

    def transform(self, X):
        X = np.array(X)
        matrices = []
        for c in range(X.shape[1]):
            Y = X[:, c]
            matrix = np.zeros(shape=(len(Y), len(self.values[c])), dtype=np.int8)
            for i, x in enumerate(Y):
                if x in self.values[c]:
                    matrix[i][self.values[c][x]] = 1
            matrices.append(matrix)
        res = np.concatenate(matrices, axis=1)
        return res

Définir un pipeline et créer un module d'entraînement

Créez ensuite un module d'entraînement pour entraîner votre pipeline scikit-learn sur les données de recensement. Une partie de ce code implique la définition du pipeline.

Ce module d'entraînement effectue les opérations suivantes :

  • Il télécharge les données d'entraînement et les charge dans un objet DataFrame pandas pouvant être utilisé par scikit-learn.
  • Il définit le pipeline scikit-learn à entraîner. Cet exemple n'utilise que trois caractéristiques numériques ('age', 'education-num' et 'hours-per-week') et trois caractéristiques catégoriques ('workclass', 'marital-status' et 'relationship') issues des données d'entrée. Il transforme les caractéristiques numériques à l'aide de la classe StandardScaler intégrée à scikit-learn et transforme les caractéristiques catégoriques à l'aide de l'encodeur one-hot personnalisé que vous avez défini dans my_pipeline.py. Il combine ensuite les données prétraitées en entrées à utiliser avec un classificateur.
  • Enfin, il exporte le modèle à l'aide de la version de joblib incluse dans scikit-learn et l'enregistre dans votre bucket Cloud Storage.

Écrivez le code suivant dans census_package/train.py :

import warnings
import argparse
from google.cloud import storage

import pandas as pd
import numpy as np
from sklearn.externals import joblib
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.pipeline import Pipeline, FeatureUnion, make_pipeline
import census_package.my_pipeline as mp
warnings.filterwarnings('ignore')


def download_data(bucket_name, gcs_path, local_path):
    bucket = storage.Client().bucket(bucket_name)
    blob = bucket.blob(gcs_path)
    blob.download_to_filename(local_path)


def upload_data(bucket_name, gcs_path, local_path):
    bucket = storage.Client().bucket(bucket_name)
    blob = bucket.blob(gcs_path)
    blob.upload_from_filename(local_path)


def get_features_target(local_path):
    strip = np.vectorize(str.strip)
    raw_df = pd.read_csv(local_path, header=None)
    target_index = len(raw_df.columns) - 1  # Last columns, 'income-level', is the target

    features_df = raw_df.drop(target_index, axis=1)
    features = features_df.as_matrix()
    target = strip(raw_df[target_index].values)
    return features, target


def create_pipeline():
    # We want to use 3 categorical and 3 numerical features in this sample.
    # Categorical features: age, education-num, and hours-per-week
    # Numerical features: workclass, marital-status, and relationship
    numerical_indices = [0, 4, 12]  # age, education-num, and hours-per-week
    categorical_indices = [1, 5, 7]  # workclass, marital-status, and relationship

    p1 = make_pipeline(mp.PositionalSelector(categorical_indices), mp.StripString(), mp.SimpleOneHotEncoder())
    p2 = make_pipeline(mp.PositionalSelector(numerical_indices), StandardScaler())

    feats = FeatureUnion([
        ('numericals', p1),
        ('categoricals', p2),
    ])

    pipeline = Pipeline([
        ('pre', feats),
        ('estimator', GradientBoostingClassifier(max_depth=4, n_estimators=100))
    ])
    return pipeline


def get_bucket_path(gcs_uri):
    if not gcs_uri.startswith('gs://'):
        raise Exception('{} does not start with gs://'.format(gcs_uri))
    no_gs_uri = gcs_uri[len('gs://'):]
    first_slash_index = no_gs_uri.find('/')
    bucket_name = no_gs_uri[:first_slash_index]
    gcs_path = no_gs_uri[first_slash_index + 1:]
    return bucket_name, gcs_path


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--gcs_data_path', action="store", required=True)
    parser.add_argument('--gcs_model_path', action="store", required=True)

    arguments, others = parser.parse_known_args()

    local_path = '/tmp/adul.data'
    data_bucket, data_path = get_bucket_path(arguments.gcs_data_path)
    print('Downloading the data...')
    download_data(data_bucket, data_path, local_path)
    features, target = get_features_target(local_path)
    pipeline = create_pipeline()

    print('Training the model...')
    pipeline.fit(features, target)

    joblib.dump(pipeline, './model.joblib')

    model_bucket, model_path = get_bucket_path(arguments.gcs_model_path)
    upload_data(model_bucket, model_path, './model.joblib')
    print('Model was successfully uploaded.')

Entraîner le pipeline sur AI Platform Training

Utilisez gcloud pour envoyer une tâche d'entraînement à AI Platform Training. La commande suivante permet d'empaqueter votre application d'entraînement, de l'importer dans Cloud Storage et d'indiquer à AI Platform Training d'exécuter votre module d'entraînement.

L'argument -- est un séparateur : le service AI Platform Training n'utilise pas d'arguments qui suivent le séparateur, mais votre module d'entraînement peut toujours y accéder.

gcloud ai-platform jobs submit training census_training_$(date +"%Y%m%d_%H%M%S") \
  --job-dir gs://$BUCKET_NAME/custom_pipeline_tutorial/job \
  --package-path ./census_package \
  --module-name census_package.train \
  --region $REGION \
  --runtime-version 1.13 \
  --python-version 3.5 \
  --scale-tier BASIC \
  --stream-logs \
  -- \
  --gcs_data_path gs://cloud-samples-data/ai-platform/census/data/adult.data.csv \
  --gcs_model_path gs://$BUCKET_NAME/custom_pipeline_tutorial/model/model.joblib

Déployer le pipeline et diffuser les prédictions

Pour diffuser des prédictions depuis AI Platform Prediction, vous devez déployer une ressource de modèle et une ressource de version. Le modèle vous aide à organiser plusieurs déploiements si vous modifiez et entraînez votre pipeline plusieurs fois. La version utilise votre modèle entraîné et votre code personnalisé pour diffuser des prédictions.

Pour déployer ces ressources, vous devez fournir les deux artefacts suivants :

  • Un bucket Cloud Storage contenant votre pipeline entraîné. La tâche d'entraînement de l'étape précédente a créé ce fichier lors de l'exportation de model.joblib vers votre bucket.
  • Un package de distribution source .tar.gz dans Cloud Storage contenant tous les transformateurs personnalisés utilisés par votre pipeline. Vous allez le créer lors de l'étape suivante.

Empaqueter les transformateurs personnalisés

Si vous déployez une version sans fournir le code de my_pipeline.py, le service AI Platform Prediction ne pourra pas importer les transformateurs personnalisés (par exemple, mp.SimpleOneHotEncoder), ni diffuser des prédictions.

Créez le fichier setup.py suivant pour définir un package de distribution source pour votre code :

import setuptools
setuptools.setup(name='census_package',
      packages=['census_package'],
      version="1.0",
      )

Exécutez ensuite la commande suivante pour créer dist/census_package-1.0.tar.gz :

python setup.py sdist --formats=gztar

Enfin, importez ce package tarball sur le bucket Cloud Storage :

gcloud storage cp ./dist/census_package-1.0.tar.gz gs://$BUCKET_NAME/custom_pipeline_tutorial/code/census_package-1.0.tar.gz

Créer des ressources de modèle et de version

Commencez par définir les noms de modèle et de version :

MODEL_NAME='CensusPredictor'
VERSION_NAME='v1'

Créez ensuite la ressource de modèle à l'aide de la commande suivante :

gcloud ai-platform models create $MODEL_NAME \
  --regions $REGION

Enfin, créez la ressource de version en fournissant des chemins d'accès Cloud Storage à votre répertoire de modèle (contenant model.joblib) et à votre code personnalisé (census_package-1.0.tar.gz) :

gcloud components install beta

gcloud beta ai-platform versions create $VERSION_NAME --model $MODEL_NAME \
  --origin gs://$BUCKET_NAME/custom_pipeline_tutorial/model/ \
  --runtime-version 1.13 \
  --python-version 3.5 \
  --framework SCIKIT_LEARN \
  --package-uris gs://$BUCKET_NAME/custom_pipeline_tutorial/code/census_package-1.0.tar.gz

Diffuser des prédictions en ligne

Testez votre déploiement en envoyant une requête de prédiction en ligne. Commencez par installer la bibliothèque cliente des API Google pour Python :

pip install --upgrade google-api-python-client

Envoyez ensuite deux instances issues des données de recensement à votre version déployée :

import googleapiclient.discovery

instances = [
  [39, 'State-gov', 77516, ' Bachelors .  ', 13, 'Never-married', 'Adm-clerical', 'Not-in-family',
   'White', 'Male', 2174, 0, 40, 'United-States', '<=50K'],
  [50, 'Self-emp-not-inc', 83311, 'Bachelors', 13, 'Married-civ-spouse', 'Exec-managerial', 'Husband',
   'White', 'Male', 0, 0, 13, 'United-States', '<=50K']
]

service = googleapiclient.discovery.build('ml', 'v1')
name = 'projects/{}/models/{}/versions/{}'.format(PROJECT_ID, MODEL_NAME, VERSION_NAME)

response = service.projects().predict(
    name=name,
    body={'instances': instances}
).execute()

if 'error' in response:
    raise RuntimeError(response['error'])
else:
  print(response['predictions'])

La version transmet les données d'entrée via le pipeline entraîné et renvoie les résultats du classificateur : <=50K ou >50K pour chaque instance, en fonction de sa prédiction concernant la tranche de revenu de la personne.

Nettoyer

Pour nettoyer toutes les ressources Google Cloud utilisées dans ce projet, vous pouvez supprimer le projet Google Cloud que vous avez utilisé dans le cadre de ce tutoriel.

Vous pouvez également nettoyer des ressources individuelles en exécutant les commandes suivantes :

# Delete version resource
gcloud ai-platform versions delete $VERSION_NAME --quiet --model $MODEL_NAME

# Delete model resource
gcloud ai-platform models delete $MODEL_NAME --quiet

# Delete Cloud Storage objects that were created
gcloud storage rm gs://$BUCKET_NAME/custom_pipeline_tutorial --recursive

Étape suivante