Predicción con una canalización de scikit-learn personalizada

Logotipo de Colab Ejecuta este instructivo como un notebook en Colab Logotipo de GitHub Visualiza el notebook en GitHub

En este instructivo, se muestra cómo usar AI Platform Prediction para implementar una canalización de scikit-learn que use transformadores personalizados.

Las canalizaciones de scikit-learn te permiten crear varios estimadores. Por ejemplo, puedes usar transformadores para realizar un procesamiento previo de los datos y pasar los datos transformados a un clasificador. Scikit-learn proporciona muchos transformadores en el paquete sklearn.

También puedes usar la clase FunctionTransformer o TransformerMixin de scikit-learn para crear tu propio transformador personalizado. Si deseas implementar una canalización en AI Platform Prediction que use transformadores personalizados, debes proporcionar ese código a AI Platform Prediction como un paquete de distribución de origen.

En este instructivo, se presenta un problema de ejemplo que incluye datos de censo para orientarte en estos pasos:

  • Entrenar una canalización de scikit-learn con transformadores personalizados en el entrenamiento de AI Platform
  • Implementar la canalización entrenada y tu código personalizado en AI Platform Prediction
  • Entregar solicitudes de predicción de esa implementación

Conjunto de datos

En este instructivo, se usa el conjunto de datos de ingresos del censo de Estados Unidos que brinda UC Irvine Machine Learning Repository. Este conjunto de datos contiene información sobre personas proveniente de una base de datos del censo de 1994 en el que se incluyen edad, educación, estado civil, ocupación y si ganan, o no, más de $50,000 al año.

Los datos que se usaron en este instructivo están disponibles en un depósito público de Cloud Storage: gs://cloud-samples-data/ai-platform/sklearn/census_data/

Objetivo

El objetivo es entrenar una canalización de scikit-learn que prediga si una persona gana más de $50,000 al año (etiqueta objetivo) en función de otra información sobre la persona (atributos) proveniente del censo.

Este instructivo se centra más en el uso de este modelo con AI Platform Prediction que en el diseño del modelo en sí. Sin embargo, cuando se compilan sistemas de aprendizaje automático, siempre es importante considerar los posibles problemas y las consecuencias no deseadas. Consulta el ejercicio sobre equidad del Curso intensivo de aprendizaje automático para obtener más información sobre las fuentes de sesgo del conjunto de datos del censo y sobre la equidad del aprendizaje automático en general.

Costos

En este instructivo, se usan los siguientes componentes facturables de Google Cloud:

  • AI Platform Training
  • AI Platform Prediction
  • Cloud Storage

Obtén información sobre los precios de AI Platform Training, AI Platform Prediction y Cloud Storage, y usa la calculadora de precios para generar una estimación de costos en función del uso previsto.

Antes de comenzar

Para poder entrenar y, luego, implementar un modelo en AI Platform Prediction, debes realizar lo siguiente:

  • Configura el entorno de desarrollo local.
  • Configura un proyecto de Google Cloud que tenga habilitadas la facturación y las API necesarias.
  • Crea un bucket de Cloud Storage para almacenar el paquete de entrenamiento y el modelo entrenado.

Configura el entorno de desarrollo local

Para completar este instructivo, necesitarás lo siguiente:

  • Python 3
  • virtualenv
  • El SDK de Google Cloud

En la guía de Google Cloud Configurar un entorno de desarrollo de Python, se brindan instrucciones detalladas para cumplir con estos requisitos. En los siguientes pasos, se brinda un conjunto abreviado de instrucciones:

  1. Instala Python 3.

  2. Instala virtualenv y crea un entorno virtual que use Python 3.

  3. Activa ese entorno.

  4. Completa los pasos de la siguiente sección para instalar el SDK de Google Cloud.

Configura el proyecto de 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

Autentica la cuenta de GCP

Para configurar la autenticación, debes crear una clave de cuenta de servicio y establecer una variable de entorno para la ruta de acceso del archivo a la clave de la cuenta de servicio.

  1. Crear una cuenta de servicio:

    1. En la consola de Google Cloud, ve a la página Crear cuenta de servicio.

      Ve a Crear cuenta de servicio

    2. Ingresa un nombre en el campo Nombre de cuenta de servicio.
    3. Opcional: en el campo Descripción de la cuenta de servicio, ingresa una descripción.
    4. Haz clic en Crear.
    5. Haz clic en el campo Seleccionar una función. En Todos los roles, selecciona AI Platform > Administrador de AI Platform.
    6. Haz clic en Agregar otra función.

    7. Haz clic en el campo Seleccionar una función. En Todas las funciones, selecciona Storage > Administrador de objeto de Storage.

    8. Haz clic en Listo para crear la cuenta de servicio.

      No cierres la ventana del navegador. La usarás en la próxima tarea.

  2. Crea una clave de cuenta de servicio para la autenticación:

    1. En la consola de Google Cloud, haz clic en la dirección de correo electrónico de la cuenta de servicio que creaste.
    2. Haga clic en Claves.
    3. Haz clic en Agregar clave -> Crear nueva clave.
    4. Haz clic en Crear. Se descargará un archivo de claves JSON en tu computadora.
    5. Haz clic en Cerrar.

  3. Configura la variable de entorno GOOGLE_APPLICATION_CREDENTIALS en la ruta de acceso del archivo JSON que contiene la clave de tu cuenta de servicio. Esta variable solo se aplica a tu sesión actual de shell. Por lo tanto, si abres una sesión nueva, deberás volver a configurar la variable.

Crea un bucket de Cloud Storage

En este instructivo se usa Cloud Storage de varias maneras:

  • Cuando envías un trabajo de entrenamiento mediante el SDK de Cloud, subes un paquete de Python que contiene el código de entrenamiento a un bucket de Cloud Storage. AI Platform Training ejecuta el código de este paquete.

  • En este instructivo, AI Platform Training también guarda un modelo entrenado que se genera a partir del trabajo en el mismo bucket.

  • A fin de implementar la canalización de scikit-learn que usa código personalizado para AI Platform Prediction, debes subir a Cloud Storage los transformadores personalizados que usa la canalización.

Cuando creas el recurso de versión de AI Platform Prediction que entrega las predicciones, debes proporcionar la canalización de scikit-learn entrenada y el código personalizado como URI de Cloud Storage.

Establece el nombre del bucket de Cloud Storage como una variable de entorno. Debe ser único en todos los depósitos de Cloud Storage:

BUCKET_NAME="your-bucket-name"

Selecciona una región en la que estén disponibles AI Platform Training y AI Platform Prediction, y crea otra variable de entorno. Por ejemplo:

REGION="us-central1"

Crea tu bucket de Cloud Storage en esta región y, a continuación, usa la misma región para el entrenamiento y la predicción. Ejecuta el siguiente comando para crear el bucket si todavía no existe:

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

Cómo crear una aplicación entrenada y código de canalización personalizado

Crea una aplicación para entrenar una canalización de scikit-learn con los datos del censo. En este instructivo, el paquete de entrenamiento también contiene el código personalizado que usa la canalización entrenada durante la predicción. Este es un patrón útil, porque las canalizaciones se suelen diseñar para usar los mismos transformadores durante el entrenamiento y la predicción.

Usa los pasos siguientes para crear un directorio con tres archivos en el interior que coincida con la estructura siguiente:

census_package/
    __init__.py
    my_pipeline.py
    train.py

Primero, crea el directorio vacío census_package/:

mkdir census_package

Dentro de census_package/, crea un archivo en blanco con el nombre __init__.py:

touch ./census_package/__init__.py

Esto permite importar census_package/ como un paquete en Python.

Crea transformadores personalizados

Scikit-learn proporciona muchos transformadores que puedes usar como parte de una canalización, pero también te permite definir tus propios transformadores personalizados. Estos transformadores pueden aprender incluso un estado guardado durante el entrenamiento que se usa más tarde en la predicción.

Extiende sklearn.base.TransformerMixin para definir tres transformadores:

  • PositionalSelector: Dada una lista de índices C y una matriz M, esto muestra una matriz con un subconjunto de columnas de M, indicado por C.

  • StripString: dada una matriz de strings, esto quita los espacios en blanco de cada string.

  • SimpleOneHotEncoder: Es un codificador de one-hot simple que se puede aplicar a una matriz de cadenas.

Para ello, escribe el siguiente código en un archivo llamado 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

Define la canalización y crea un módulo de entrenamiento

A continuación, crea un módulo de entrenamiento para entrenar tu canalización de scikit-learn con los datos del censo. Parte de este código implica definir la canalización.

Este módulo de entrenamiento realiza varias tareas:

  • Descarga datos de entrenamiento y los carga en un Pandas DataFrame que puede usar scikit-learn.
  • Define la canalización de scikit-learn para el entrenamiento. En este ejemplo, solo se usan tres atributos numéricos ('age', 'education-num' y 'hours-per-week') y tres atributos categóricos ('workclass', 'marital-status' y 'relationship') de los datos de entrada. Transforma los atributos numéricos con el StandardScaler incorporado de scikit-learn y transforma los categóricos con el codificador one-hot personalizado que definiste en my_pipeline.py. Luego, combina los datos ya procesados como entrada para un clasificador.
  • Por último, exporta el modelo mediante la versión de joblib incluida en scikit-learn y lo guarda en el depósito de Cloud Storage.

Escribe el siguiente código en 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.')

Entrena la canalización en AI Platform Training

Usa gcloud para enviar un trabajo de entrenamiento a AI Platform Training. Con el siguiente comando, se empaqueta la aplicación de entrenamiento, se la sube a Cloud Storage y se indica a AI Platform Training que ejecute el módulo de entrenamiento.

El argumento -- es un separador: el servicio de AI Platform Training no usa argumentos que sigan al separador, pero el módulo de entrenamiento puede acceder a ellos.

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

Implementa la canalización y entrega predicciones

Para entregar predicciones de AI Platform Prediction, debes implementar un recurso de modelo y uno de versión. El modelo te ayuda a organizar varias implementaciones si modificas y entrenas la canalización varias veces. La versión usa tu modelo entrenado y el código personalizado para entregar predicciones.

Para implementar estos recursos, debes proporcionar dos artefactos:

  • Un directorio de Cloud Storage que contiene la canalización entrenada. El trabajo de entrenamiento del paso anterior creó este archivo cuando exportó model.joblib a tu depósito.
  • Un paquete de distribución de fuente .tar.gz en Cloud Storage que contiene cualquier transformador personalizado que usa tu canalización. Crea esto en el paso siguiente.

Empaqueta los transformadores personalizados

Si implementas una versión sin proporcionar el código desde my_pipeline.py, AI Platform Prediction no podrá importar los transformadores personalizados (por ejemplo, mp.SimpleOneHotEncoder) y no podrá entregar predicciones.

Crea el setup.py siguiente si quieres definir un paquete de distribución de origen para tu código:

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

Luego, ejecuta el comando siguiente para crear dist/census_package-1.0.tar.gz:

python setup.py sdist --formats=gztar

Por último, sube este archivo comprimido a tu bucket de 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

Crea recursos de modelo y de versión

Primero, define los nombres del modelo y la versión:

MODEL_NAME='CensusPredictor'
VERSION_NAME='v1'

Luego, usa el comando siguiente para crear el recurso de modelo:

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

Por último, crea el recurso de la versión; para ello, proporciona rutas de Cloud Storage al directorio de tu modelo (el que contiene model.joblib) y tu código personalizado (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

Entrega predicciones en línea

Para probar tu implementación, envía una solicitud de predicción en línea. Primero, instala la biblioteca cliente de la API de Google para Python:

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

Luego, envía dos instancias de datos del censo a tu versión implementada:

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 versión pasa los datos de entrada a través de la canalización entrenada y muestra los resultados del clasificador: <=50K o >50K de cada instancia, según su predicción sobre el nivel de ingresos de la persona.

Realice una limpieza

Para limpiar todos los recursos de Google Cloud que se usaron en este proyecto, puedes borrar el proyecto de Google Cloud que usaste para el instructivo.

Alternativamente, puedes limpiar recursos individuales; para ello, ejecuta los siguientes comandos:

# 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

¿Qué sigue?