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 para implementar una canalización de scikit-learn que usa transformadores personalizados.

Las canalizaciones de scikit-learn pueden ayudarte a redactar varios estimadores. Por ejemplo, puedes usar transformadores para preprocesar 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 quieres implementar una canalización que use transformadores personalizados en AI Platform Prediction, debes proporcionar ese código a AI Platform como un paquete de distribución de fuente.

En este instructivo, se presenta un ejemplo de un problema que incluye datos de censo para orientarte en los pasos próximos:

  • Entrena 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
  • Entrega 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 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 Curso intensivo de aprendizaje automático sobre la equidad para obtener más información sobre fuentes de sesgo en el conjunto de datos del censo, y sobre la equidad del aprendizaje automático en general.

Costos

En este instructivo, se usan componentes facturables de Google Cloud Platform (GCP):

  • AI Platform
  • Cloud Storage

Obtén información sobre los precios de AI Platform y los precios de Cloud Storage. Usa la calculadora de precios para generar una estimación de los costos según el uso previsto.

Antes de comenzar

Debes completar varios pasos antes de poder entrenar y, luego, implementar un modelo en AI Platform:

  • Configura tu entorno de desarrollo local.
  • Configura un proyecto de GCP para que la facturación y las API necesarias estén habilitadas.
  • Crea un depósito de Cloud Storage para almacenar tu paquete de entrenamiento y tu modelo entrenado.

Configura el entorno de desarrollo local

Para completar este instructivo, necesitarás lo siguiente:

  • Python 3
  • virtualenv
  • El SDK de Cloud

En la guía de Google Cloud Configura 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 en la siguiente sección para instalar el SDK de Cloud.

Configura tu proyecto de GCP

  1. Accede a tu Cuenta de Google.

    Si todavía no tienes una cuenta, regístrate para obtener una nueva.

  2. Selecciona o crea un proyecto de GCP.

    Ir a la página Administrar recursos

  3. Asegúrate de tener habilitada la facturación para tu proyecto.

    Aprende a habilitar la facturación

  4. Habilita las AI Platform (“Cloud Machine Learning Engine”) y Compute Engine API necesarias.

    Habilita las API

  5. Realiza la instalación y la inicialización del SDK de Cloud.

Autentica tu 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. Crea una clave de cuenta de servicio para la autenticación:
    1. En GCP Console, ve a la página Crea una clave de la cuenta de servicio.

      Ir a la página Crea una clave de la cuenta de servicio
    2. En la lista desplegable de Cuenta de servicio, selecciona Nueva cuenta de servicio.
    3. Ingresa un nombre en el campo Nombre de cuenta de servicio.
    4. En la lista desplegable Función, selecciona Machine Learning Engine > Administrador de ML Engine y Almacenamiento > Administrador de objeto de almacenamiento.

      Nota: El campo Función autoriza el acceso de tu cuenta de servicio a los recursos. Puedes ver y cambiar este campo más tarde mediante GCP Console. Si desarrollas una app de producción, es posible que tengas que especificar permisos más detallados que Machine Learning Engine > Administrador de ML Engine y Almacenamiento > Administrador de objeto de almacenamiento. Para obtener más información, consulta el control de acceso de AI Platform.
    5. Haz clic en Crear. Se descargará en tu computadora un archivo JSON que contiene tu clave.
  2. Configura la variable de entorno GOOGLE_APPLICATION_CREDENTIALS a la ruta de acceso al 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 depósito 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 tu código de entrenamiento a un depósito de Cloud Storage. AI Platform ejecuta el código de este paquete.

  • En este instructivo, AI Platform también guarda un modelo entrenado que surge de tu trabajo en el mismo depósito.

  • A fin de implementar tu canalización de scikit-learn con código personalizado a modo de entregar predicciones, debes subir los transformadores personalizados que usa tu canalización para Cloud Storage.

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

Establece el nombre de tu depósito 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 cual el entrenamiento y la predicción de AI Platform estén disponibles y crea otra variable de entorno. No se puede utilizar un depósito Multi-Regional Storage para el entrenamiento con AI Platform. Por ejemplo:

REGION="us-central1"

Crea tu depósito 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 depósito si todavía no existe:

gsutil mb -l $REGION gs://$BUCKET_NAME

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

La plataforma 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 incluso pueden aprender 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: un codificador de one-hot simple que puede aplicarse a una matriz de string.

Para ello, escribe el código siguiente 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 procesados con anterioridad como entrada para un clasificador.
  • Por último, exporta el modelo con joblib y lo guarda en tu depósito de Cloud Storage.

Escribe el código siguiente 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

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

El argumento -- es un separador: el servicio de AI Platform no usa los argumentos que siguen 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, debes implementar un recurso de modelo y uno de versión. El modelo te ayuda a organizar varias implementaciones si modificas y entrenas tu 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 tu 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 tus 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 depósito de Cloud Storage:

gsutil 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 las 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.

Realiza una limpieza

Para limpiar todos los recursos de GCP usados en este proyecto, puedes borrar el proyecto de GCP que usaste para este 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
gsutil -m rm -r gs://$BUCKET_NAME/custom_pipeline_tutorial

Próximos pasos

¿Te sirvió esta página? Envíanos tu opinión:

Enviar comentarios sobre…

¿Necesitas ayuda? Visita nuestra página de asistencia.