Previsão com um pipeline scikit-learn personalizado

Logotipo do Colab Executar este tutorial como um notebook no Colab Logotipo do GitHub Ver o notebook no GitHub

Neste tutorial, você aprenderá como usar o AI Platform Prediction para implantar um pipeline do scikit-learn que usa transformadores personalizados.

scikit-learn pipelines permitem compor vários estimadores. Por exemplo, use transformadores para pré-processar dados e transmitir os dados transformados para um classificador. O scikit-learn oferece muitos transformadores (link em inglês) no pacote sklearn.

Também é possível usar a biblioteca scikit-learn FunctionTransformer ou TransformerMixin para criar seu próprio transformador personalizado. Se você quiser implantar um pipeline que usa transformadores personalizados no AI Platform Prediction, precisará fornecer esse código ao AI Platform Prediction como um pacote de distribuição de origem.

Neste tutorial, apresentamos um exemplo de problema envolvendo dados do censo para orientar você nas etapas a seguir:

  • Treinamento de um pipeline scikit-learn com transformadores personalizados no AI Platform Training
  • Implantação do pipeline treinado e de seu código personalizado no AI Platform Prediction
  • Exibição das solicitações de predição dessa implantação

Conjunto de dados

Neste tutorial, usamos a métrica de renda do Censo dos Estados Unidos Conjunto de dados fornecida pela UC Irvine Machine Learning Repositório. Esse conjunto de dados contém informações sobre pessoas de um banco de dados do censo de 1994, inclusive idade, escolaridade, estado civil, profissão e se ganham mais de US$ 50.000 por ano.

Os dados usados neste tutorial estão disponíveis em um bucket público do Cloud Storage: gs://cloud-samples-data/ai-platform/sklearn/census_data/

Objetivo

A meta é treinar um pipeline scikit-learn que preveja se uma pessoa ganha mais de US$ 50.000 por ano (rótulo de destino) com base em outras informações do censo sobre a pessoa (atributos).

O foco deste tutorial está mais no uso desse modelo com o AI Platform Prediction do que na criação do modelo em si. No entanto, é sempre importante pensar em problemas em potencial e consequências não intencionais durante a criação dos sistemas de machine learning. Consulte o exercício do curso intensivo de machine learning sobre imparcialidade sobre as fontes de viés no conjunto de dados do Censo, além de e aprender sobre imparcialidade em geral.

Custos

Neste tutorial, há componentes faturáveis do Google Cloud:

  • AI Platform Training
  • AI Platform Prediction
  • Cloud Storage

Saiba mais sobre o AI Platform Training preços, AI Platform Prediction preços e Cloud Storage preços, e use a função Preços Calculadora para gerar uma estimativa de custo com base na projeção de uso.

Antes de começar

Antes de treinar e implantar um modelo no AI Platform Prediction, faça as seguintes etapas:

  • Configure o ambiente de desenvolvimento local.
  • Configure um projeto do Google Cloud com o faturamento e as APIs necessárias ativadas.
  • Crie um bucket do Cloud Storage para armazenar o pacote de treinamento e o modelo treinado.

Configurar o ambiente de desenvolvimento local

Você precisa dos itens a seguir para concluir este tutorial:

  • Python 3
  • virtualenv
  • O SDK Google Cloud

No guia do Google Cloud sobre Como configurar um ambiente de desenvolvimento do Python, apresentamos instruções detalhadas para atender a esses requisitos. Veja a seguir um conjunto condensado de instruções:

  1. Instale o Python 3.

  2. Instalar virtualenv e criar um ambiente virtual que usa o Python 3.

  3. Ative esse ambiente.

  4. Conclua as etapas na seção a seguir para instalar o SDK Google Cloud.

Configurar seu projeto do 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

Autenticar a conta do GCP

Para configurar a autenticação, crie uma chave da conta de serviço e defina uma variável de ambiente para o caminho do arquivo dessa chave.

  1. Crie uma conta de serviço:

    1. No Console do Google Cloud, acesse a página Criar conta de serviço.

      Acesse "Criar conta de serviço"

    2. No campo Nome da conta de serviço, insira um nome.
    3. Opcional: no campo Descrição da conta de serviço, digite uma descrição.
    4. Clique em Criar.
    5. Clique no campo Selecionar um papel. Em Todos os papéis, selecione AI Platform > Administrador do AI Platform.
    6. Clique em Adicionar outro papel.

    7. Clique no campo Selecionar um papel. Em Todos os papéis, selecione Armazenamento > Administrador de objetos do Storage.

    8. Clique em Concluído para criar a conta de serviço.

      Não feche a janela do navegador. Você vai usá-la na próxima etapa.

  2. Crie uma chave da conta de serviço para autenticação:

    1. No console do Google Cloud, clique no endereço de e-mail da conta de serviço que você criou.
    2. Clique em Chaves.
    3. Clique em Adicionar chave e, depois, em Criar nova chave.
    4. Clique em Criar. O download de um arquivo de chave JSON é feito no seu computador.
    5. Clique em Fechar.

  3. Defina a variável de ambiente GOOGLE_APPLICATION_CREDENTIALS como o caminho do arquivo JSON que contém a chave da conta de serviço. Essa variável aplica-se apenas à sessão de shell atual. Portanto, se você abrir uma nova sessão, defina a variável novamente.

Criar um bucket do Cloud Storage

Neste tutorial, usamos o Cloud Storage de várias maneiras:

  • Ao enviar um job de treinamento usando o SDK do Cloud, você faz upload de um pacote do Python que contém o código de treinamento para um bucket do Cloud Storage. O AI Platform Training executa o código desse pacote.

  • Neste tutorial, o AI Platform Training também salva o modelo treinado resultante do job no mesmo bucket.

  • Para implantar um pipeline do scikit-learn que usa código personalizado no AI Platform Prediction, faça o upload no Cloud Storage dos transformadores personalizados que o pipeline usa.

Ao criar o recurso de versão do AI Platform Prediction que exibe as previsões, você fornece o pipeline treinado do scikit-learn e seu código personalizado como URIs do Cloud Storage.

Defina o nome do bucket do Cloud Storage como uma variável de ambiente. Ele precisa ser exclusivo em todos os buckets do Cloud Storage:

BUCKET_NAME="your-bucket-name"

Selecione uma região em que o AI Platform Training e o AI Platform Prediction estejam disponíveis e crie outra variável de ambiente. Exemplo:

REGION="us-central1"

Crie o bucket do Cloud Storage nessa região e, posteriormente, use a mesma região no treinamento e na previsão. Execute o comando a seguir para criar o bucket, caso ele ainda não exista:

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

Como criar um aplicativo de treinamento e um código de pipeline personalizado

Crie um aplicativo para treinar um pipeline scikit-learn com os dados do censo. Neste tutorial, o pacote de treinamento também contém o código personalizado que o pipeline treinado usa durante a predição. Esse é um padrão útil, porque os pipelines geralmente são projetados para usar os mesmos transformadores durante o treinamento e a predição.

Use as etapas a seguir para criar um diretório com três arquivos que corresponda à estrutura abaixo:

census_package/
    __init__.py
    my_pipeline.py
    train.py

Primeiramente, crie o diretório census_package/ vazio:

mkdir census_package

Em census_package/, crie um arquivo em branco chamado __init__.py:

touch ./census_package/__init__.py

Desse modo, é possível importar census_package/ como um pacote no Python.

Criar transformadores personalizados

O scikit-learn oferece muitos transformadores que podem ser usadas como parte de um pipeline, mas também permite que você defina e transformadores personalizados. Esses transformadores podem até aprender um estado salvo durante o treinamento que é usado posteriormente na previsão.

Estender sklearn.base.TransformerMixin para definir três transformadores:

  • PositionalSelector: com uma lista de índices C e uma matriz M, ele retorna uma matriz com um subconjunto das colunas de M, indicado por C.

  • StripString: com uma matriz de strings, ele remove os espaços em branco de cada string.

  • SimpleOneHotEncoder: um método one-hot simples codificador que pode ser aplicado a uma matriz de strings.

Para fazer isso, grave o seguinte código em um arquivo chamado 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

Definir o pipeline e criar o módulo de treinamento

Em seguida, crie um módulo de treinamento para treinar seu pipeline scikit-learn nos dados do censo. Parte deste código envolve a definição do pipeline.

Esse módulo de treinamento realiza várias ações:

  • Download dos dados de treinamento e carregamento deles no DataFrame de pandas que pode ser usado pelo scikit-learn.
  • Definir o pipeline scikit-learn para treino. Neste exemplo, usamos apenas três recursos numéricos ('age', 'education-num' e 'hours-per-week') e três recursos categóricos ('workclass', 'marital-status' e 'relationship') com base nos dados de entrada. Os recursos numéricos são transformados usando o StandardScaler integrado do scikit-learn e transformados em categóricos com o codificador one-hot personalizado definido em my_pipeline.py. Em seguida, os dados pré-processados são combinados como entrada para um classificador.
  • Por fim, exporta o modelo usando a versão do joblib incluída no scikit-learn e o salva no seu bucket do Cloud Storage.

Grave o seguinte código em 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.')

Como treinar o pipeline no AI Platform Training

Use a gcloud para enviar um job de treinamento ao AI Platform Training. O comando a seguir empacota o aplicativo de treinamento, faz o upload dele para o Cloud Storage e instrui o AI Platform Training a executar o módulo de treinamento.

O argumento -- é um separador: o serviço do AI Platform Training não usa os argumentos depois do separador, mas seu módulo de treinamento ainda poderá acessá-los.

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

Como implantar o pipeline e exibir previsões

Para exibir as previsões do AI Platform Prediction, implante um recurso de modelo e outro de versão. O modelo ajudará a organizar muitas implantações se você modificar e treinar seu pipeline várias vezes. A versão usa o modelo treinado e o código personalizado para exibir as previsões.

Para implantar esses recursos, forneça dois artefatos:

  • Um diretório do Cloud Storage que contém o pipeline treinado. O job de treinamento da etapa anterior criou esse arquivo quando ele exportou model.joblib para seu bucket.
  • Um pacote de distribuição de origem .tar.gz no Cloud Storage com os transformadores personalizados que seu pipeline usa. Crie isso na próxima etapa.

Empacotar os transformadores personalizados

Se você implantar uma versão sem fornecer o código de my_pipeline.py, o AI Platform Prediction não poderá importar os transformadores personalizados, por exemplo, mp.SimpleOneHotEncoder, e não poderá exibir previsões.

Crie o seguinte setup.py a fim de definir um pacote de distribuição de origem para seu código:

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

Execute o seguinte comando para criar dist/census_package-1.0.tar.gz:

python setup.py sdist --formats=gztar

Por fim, faça upload desse tarball para o bucket do 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

Criar recursos de modelo e versão

Primeiro, defina os nomes de modelo e versão:

MODEL_NAME='CensusPredictor'
VERSION_NAME='v1'

Depois, use o comando a seguir para criar o recurso de modelo:

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

Por fim, crie o recurso da versão especificando os caminhos do Cloud Storage para o diretório do modelo (aquele que contém model.joblib) e seu 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

Como exibir previsões on-line

Teste a implantação enviando uma solicitação de predição on-line. Instale primeiro a biblioteca de cliente de APIs do Google para Python:

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

Em seguida, envie duas instâncias de dados do censo para a versão implantada:

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'])

A versão transmite os dados de entrada por meio do pipeline treinado e retorna os resultados do classificador, <=50K ou >50K, para cada instância, dependendo da respectiva previsão referente à renda da pessoa.

Limpar

Para limpar todos os recursos do Google Cloud usados neste projeto, faça o seguinte: exclua o Google Cloud projeto que você usou no tutorial.

Também é possível limpar recursos individuais com os comandos a seguir:

# 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

A seguir