simple.ipynb

Kernel: Python 3

Copyright 2021 The TensorFlow Authors.

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Pré-processamento de dados com o TensorFlow Transform

O componente de engenharia de características do TensorFlow Extended (TFX)

Observação: recomendamos executar este tutorial em um notebook Colab, sem necessidade de configuração! Basta clicar em “Executar no Google Colab”.

Ver em TensorFlow.org

Executar no Google Colab

Ver fonte no GitHub

Baixar notebook

Este exemplo de notebook colab fornece um exemplo muito simples de como o TensorFlow Transform (tf.Transform) pode ser usado para pré-processar dados usando exatamente o mesmo código para treinar um modelo e servir inferências em produção.

O TensorFlow Transform é uma biblioteca para pré-processamento de dados de entrada para TensorFlow, incluindo a criação de características que exigem um passo completo do dataset de treinamento. Por exemplo, usando o TensorFlow Transform você poderia:

• Normalizar um valor de entrada usando a média e o desvio padrão

• Converter strings em inteiros gerando um vocabulário sobre todos os valores de entrada

• Converter números de ponto flutuante em números inteiros atribuindo-os a intervalos, com base na distribuição de dados observada

O TensorFlow tem suporte integrado para manipulações em um único exemplo ou num lote de exemplos. O tf.Transform estende esses recursos para oferecer suporte a passos completos (full pass) em todo o dataset de treinamento.

A saída do tf.Transform é exportada como um grafo do TensorFlow que você pode usar para treinamento e serviço. Usar o mesmo grafo para treinamento e serviço pode evitar desvios, uma vez que as mesmas transformações são aplicadas em ambos os estágios.

Atualize o Pip

Para evitar a atualização do Pip num sistema ao executar localmente, garanta que estamos executando no Colab. Os sistemas locais podem, claro, ser atualizados separadamente.

try:
  import colab
  !pip install --upgrade pip
except:
  pass

Instale o TensorFlow Transform

!pip install -q -U tensorflow_transform

# This cell is only necessary because packages were installed while python was
# running. It avoids the need to restart the runtime when running in Colab.
import pkg_resources
import importlib

importlib.reload(pkg_resources)

Importações

import pathlib
import pprint
import tempfile

import tensorflow as tf
import tensorflow_transform as tft

import tensorflow_transform.beam as tft_beam
from tensorflow_transform.tf_metadata import dataset_metadata
from tensorflow_transform.tf_metadata import schema_utils

Dados: crie alguns dados fictícios

Criaremos alguns dados fictícios simples para nosso exemplo:

• raw_data são os dados brutos iniciais que iremos pré-processar

• raw_data_metadata contém o esquema que nos informa os tipos de cada uma das colunas em raw_data. Neste caso, é bem simples.

raw_data = [
      {'x': 1, 'y': 1, 's': 'hello'},
      {'x': 2, 'y': 2, 's': 'world'},
      {'x': 3, 'y': 3, 's': 'hello'}
  ]

raw_data_metadata = dataset_metadata.DatasetMetadata(
    schema_utils.schema_from_feature_spec({
        'y': tf.io.FixedLenFeature([], tf.float32),
        'x': tf.io.FixedLenFeature([], tf.float32),
        's': tf.io.FixedLenFeature([], tf.string),
    }))

Transform: crie uma função de pré-processamento

A função de pré-processamento é o conceito mais importante do tf.Transform. Uma função de pré-processamento é onde a transformação do dataset realmente acontece. Ela recebe e retorna um dicionário de tensores, onde um tensor significa um Tensor ou SparseTensor. Existem dois grupos principais de chamadas de API que normalmente formam o coração de uma função de pré-processamento:

1. Ops do TensorFlow: qualquer função que aceita e retorna tensores, o que geralmente correspondem às ops do TensorFlow. Elas adicionam operações do TensorFlow ao grafo que transforma dados brutos em dados transformados, um vetor de características por vez. Eles serão executados para cada exemplo, tanto durante o treinamento quanto durante o serviço.

2. Analisadores/mapeadores do Tensorflow Transform: qualquer um dos analisadores/mapeadores fornecidos pelo tf.Transform. Eles também recebem e retornam tensores e normalmente contêm uma combinação de ops do Tensorflow e computações do Beam, mas, ao contrário das ops do TensorFlow, eles são executados apenas no pipeline do Beam durante a análise, exigindo uma passagem completa (full pass) por todo o dataset de treinamento. As computações do Beam rodam apenas uma vez (antes do treinamento, durante a análise) e normalmente fazem uma passagem completa por todo o dataset de treinamento. Criam tensores tf.constant, que são adicionados ao grafo. Por exemplo, tft.min calcula o mínimo de um tensor no dataset de treinamento.

Atenção: Quando você aplica sua função de pré-processamento para servir inferências, as constantes que foram criadas pelos analisadores durante o treinamento não mudam. Se seus dados tiverem componentes de tendência ou sazonalidade, planeje levando isso em consideração.

Observação: O preprocessing_fn não pode ser chamada diretamente. Isso significa que chamar preprocessing_fn(raw_data) não funciona. Em vez disso, ela deve ser passada para a API Beam do Transform conforme mostrado nas células a seguir.

def preprocessing_fn(inputs):
    """Preprocess input columns into transformed columns."""
    x = inputs['x']
    y = inputs['y']
    s = inputs['s']
    x_centered = x - tft.mean(x)
    y_normalized = tft.scale_to_0_1(y)
    s_integerized = tft.compute_and_apply_vocabulary(s)
    x_centered_times_y_normalized = (x_centered * y_normalized)
    return {
        'x_centered': x_centered,
        'y_normalized': y_normalized,
        's_integerized': s_integerized,
        'x_centered_times_y_normalized': x_centered_times_y_normalized,
    }

Sintaxe

Você está quase pronto para juntar tudo e usar o Apache Beam para executá-lo.

O Apache Beam usa uma sintaxe especial para definir e invocar transformações. Por exemplo, nesta linha:

result = pass_this | 'name this step' >> to_this_call

O método to_this_call está sendo invocado e passado ao objeto chamado pass_this, e esta operação será chamada de name this step num rastreamento de pilha. O resultado da chamada para to_this_call é retornado em result. Freqüentemente, você verá estágios de um pipeline encadeados assim:

result = apache_beam.Pipeline() | 'first step' >> do_this_first() | 'second step' >> do_this_last()

e como isso começou com um novo pipeline, você pode continuar assim:

next_result = result | 'doing more stuff' >> another_function()

Juntando tudo

Agora estamos prontos para transformar nossos dados. Usaremos o Apache Beam com um executor direto e forneceremos três entradas:

1. raw_data – Os dados de entrada brutos que criamos acima

2. raw_data_metadata – O esquema para os dados brutos

3. preprocessing_fn – A função que criamos para fazer nossa transformação

def main(output_dir):
  # Ignore the warnings
  with tft_beam.Context(temp_dir=tempfile.mkdtemp()):
    transformed_dataset, transform_fn = (  # pylint: disable=unused-variable
        (raw_data, raw_data_metadata) | tft_beam.AnalyzeAndTransformDataset(
            preprocessing_fn))

  transformed_data, transformed_metadata = transformed_dataset  # pylint: disable=unused-variable

  # Save the transform_fn to the output_dir
  _ = (
      transform_fn
      | 'WriteTransformFn' >> tft_beam.WriteTransformFn(output_dir))

  return transformed_data, transformed_metadata

output_dir = pathlib.Path(tempfile.mkdtemp())

transformed_data, transformed_metadata = main(str(output_dir))

print('\nRaw data:\n{}\n'.format(pprint.pformat(raw_data)))
print('Transformed data:\n{}'.format(pprint.pformat(transformed_data)))

Será que esta é a resposta certa?

Previamente, usamos o tf.Transform para fazer o seguinte:

x_centered = x - tft.mean(x)
y_normalized = tft.scale_to_0_1(y)
s_integerized = tft.compute_and_apply_vocabulary(s)
x_centered_times_y_normalized = (x_centered * y_normalized)

• x_centered - Com entrada de [1, 2, 3] a média de x é 2, e subtraímos de x para centralizar nossos valores de x em 0. Portanto, nosso resultado de [-1.0, 0.0, 1.0] está correto.

• y_normalized - Queríamos dimensionar nossos valores de y entre 0 e 1. Nossa entrada foi [1, 2, 3], então nosso resultado de [0.0, 0.5, 1.0] está correto.

• s_integerized - Queríamos mapear nossas strings para índices em um vocabulário e havia apenas 2 palavras em nosso vocabulário ("hello" e "world"). Portanto, com a entrada de ["hello", "world", "hello"] nosso resultado de [0, 1, 0] está correto. Como “olá” ocorre com mais frequência nesses dados, será a primeira entrada no vocabulário.

• x_centered_times_y_normalized - Queríamos criar uma nova característicaa cruzando x_centered e y_normalized usando multiplicação. Observe que isso multiplica os resultados, não os valores originais, e nosso novo resultado de [-0.0, 0.0, 1.0] está correto.

Use o transform_fn resultante

!ls -l {output_dir}

O diretório transform_fn/ contém uma implementação de tf.saved_model com todos os resultados da análise das constantes do Tensorflow Transform incorporadas ao grafo.

É possível carregar isso diretamente com tf.saved_model.load, mas não é fácil de usar:

loaded = tf.saved_model.load(str(output_dir/'transform_fn'))
loaded.signatures['serving_default']

Uma abordagem melhor seria carregá-lo usando tft.TFTransformOutput. O método TFTransformOutput.transform_features_layer retorna um objeto tft.TransformFeaturesLayer que pode ser usado para aplicar a transformação:

tf_transform_output = tft.TFTransformOutput(output_dir)

tft_layer = tf_transform_output.transform_features_layer()
tft_layer

Este tft.TransformFeaturesLayer espera um dicionário de características em lote. Portanto, crie um Dict[str, tf.Tensor] de List[Dict[str, Any]] em raw_data:

raw_data_batch = {
    's': tf.constant([ex['s'] for ex in raw_data]),
    'x': tf.constant([ex['x'] for ex in raw_data], dtype=tf.float32),
    'y': tf.constant([ex['y'] for ex in raw_data], dtype=tf.float32),
}

Você pode usar o tft.TransformFeaturesLayer de forma independente:

transformed_batch = tft_layer(raw_data_batch)

{key: value.numpy() for key, value in transformed_batch.items()}

Exportação

Um caso de uso mais típico seria usar tf.Transform para aplicar a transformação aos datasets de treinamento e avaliação (veja um exemplo no próximo tutorial). Nesse caso, após o treinamento, antes de exportar o modelo, anexe o tft.TransformFeaturesLayer como a primeira camada para que você possa exportá-lo como parte do seu tf.saved_model. Para um exemplo concreto, continue lendo.

Um exemplo de modelo de treinamento

Abaixo temos um modelo que:

1. recebe o lote transformado,

2. empilha tudo numa matriz (batch, features) simples,

3. passa todos por algumas camadas densas e

4. produz 10 saídas lineares.

Num caso de uso real, você aplicaria um one-hot à característica s_integerized.

Você poderia treinar este modelo num dataset transformado por tf.Transform:

class StackDict(tf.keras.layers.Layer):
  def call(self, inputs):
    values = [
        tf.cast(v, tf.float32)
        for k,v in sorted(inputs.items(), key=lambda kv: kv[0])]
    return tf.stack(values, axis=1)

class TrainedModel(tf.keras.Model):
  def __init__(self):
    super().__init__(self)
    self.concat = StackDict()
    self.body = tf.keras.Sequential([
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(10),
    ])

  def call(self, inputs, training=None):
    x = self.concat(inputs)
    return self.body(x, training)

trained_model = TrainedModel()

Imagine que treinamos o modelo.

trained_model.compile(loss=..., optimizer='adam')
trained_model.fit(...)

Este modelo é executado nas entradas transformadas

trained_model_output = trained_model(transformed_batch)
trained_model_output.shape

Um exemplo de wrapper de exportação

Imagine que você treinou o modelo acima e deseja exportá-lo.

Você vai querer incluir a função de transformação no modelo exportado:

class ExportModel(tf.Module):
  def __init__(self, trained_model, input_transform):
    self.trained_model = trained_model
    self.input_transform = input_transform

  @tf.function
  def __call__(self, inputs, training=None):
    x = self.input_transform(inputs)
    return self.trained_model(x)

export_model = ExportModel(trained_model=trained_model,
                           input_transform=tft_layer)

Esse modelo combinado funciona com dados brutos e produz exatamente os mesmos resultados que chamar diretamente o modelo treinado:

export_model_output = export_model(raw_data_batch)
export_model_output.shape

tf.reduce_max(abs(export_model_output - trained_model_output)).numpy()

Este export_model inclui o tft.TransformFeaturesLayer e é totalmente independente. Você pode salvá-lo e restaurá-lo em outro ambiente e ainda obter exatamente o mesmo resultado:

import tempfile
model_dir = tempfile.mkdtemp(suffix='tft')

tf.saved_model.save(export_model, model_dir)

reloaded = tf.saved_model.load(model_dir)

reloaded_model_output = reloaded(raw_data_batch)
reloaded_model_output.shape

tf.reduce_max(abs(export_model_output - reloaded_model_output)).numpy()