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.

Preprocesamiento de datos con TensorFlow Transform

El componente de ingeniería de características de TensorFlow Extended (TFX)

Nota: Recomendamos ejecutar este tutorial en un bloc de notas de Colab, ¡no es necesario configurarlo! Simplemente haga clic en "Ejecutar en Google Colab".

Ver en TensorFlow.org

Ejecutar en Google Colab

Ver fuente en GitHub

Descargar el bloc de notas

Este bloc de notas de Colab de ejemplo proporciona un ejemplo muy simple de cómo se puede usar TensorFlow Transform (tf.Transform) para preprocesar datos de manera que se use exactamente el mismo código tanto para entrenar un modelo como para hacer inferencias en producción.

TensorFlow Transform es una biblioteca que permite preprocesar datos de entrada para TensorFlow, lo que incluye la creación de características que requieren un paso completo sobre el conjunto de datos de entrenamiento. Por ejemplo, mediante el uso de TensorFlow Transform podría hacer lo siguiente:

• Normalizar un valor de entrada a través del uso de la media y la desviación estándar

• Convertir cadenas a números enteros mediante la generación de un vocabulario sobre todos los valores de entrada

• Convertir valores flotantes en enteros al asignarlos a cubos, en función de la distribución de datos observada

TensorFlow tiene soporte integrado para manipulaciones en un solo ejemplo o un lote de ejemplos. tf.Transform amplía estas capacidades para permitir pasos completos sobre todo el conjunto de datos de entrenamiento.

La salida de tf.Transform se exporta como un grafo de TensorFlow que se puede usar tanto para entrenamiento como para servicio. Usar el mismo grafo tanto para el entrenamiento como para el servicio sirve para evitar sesgos, ya que se aplican las mismas transformaciones en ambas etapas.

Actualización de pip

Para evitar actualizar Pip en un sistema cuando se ejecuta localmente, verifique que se esté ejecutando en Colab. Por supuesto, los sistemas locales se pueden actualizar por separado.

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

Instale 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)

Importaciones

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

Datos: cree algunos datos ficticios

Crearemos algunos datos ficticios simples para nuestro ejemplo simple:

• raw_data son los datos sin procesar iniciales que vamos a preprocesar

• raw_data_metadata contiene el esquema que nos indica los tipos de cada una de las columnas en raw_data. En este caso, es muy sencillo.

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),
    }))

Transformación: cree una función de preprocesamiento

La función de preprocesamiento es el concepto más importante de tf.Transform. Una función de preprocesamiento es la que realmente ejecuta la transformación del conjunto de datos. Acepta y devuelve un diccionario de tensores, donde tensor significa Tensor o SparseTensor. Hay dos grupos principales de llamadas API que normalmente constituyen el núcleo de una función de preprocesamiento:

1. TensorFlow Ops: cualquier función que acepte y devuelva tensores, lo que generalmente significa operaciones de TensorFlow. Estas agregan operaciones de TensorFlow al grafo que transforma datos sin procesar en datos transformados, un vector de características a la vez. Estas se ejecutarán para cada ejemplo, tanto durante el entrenamiento como durante el servicio.

2. Tensorflow Transform Analyzers/Mappers: cualquiera de los analizadores/asignadores que ofrece tf.Transform. Estos también aceptan y devuelven tensores y, por lo general, contienen una combinación de operaciones de Tensorflow y cálculos de Beam, pero a diferencia de las operaciones de TensorFlow, solo se ejecutan en la canalización de Beam durante el análisis, lo que requiere un paso completo por todo el conjunto de datos de entrenamiento. El cálculo de Beam se ejecuta solo una vez (antes del entrenamiento, durante el análisis) y, por lo general, hace un recorrido completo por todo el conjunto de datos de entrenamiento. Se encargan de crear tensores tf.constant, que se agregan a su grafo. Por ejemplo, tft.min calcula el mínimo de un tensor sobre el conjunto de datos de entrenamiento.

Atención: Cuando aplique su función de preprocesamiento para servir inferencias, las constantes creadas por los analizadores durante el entrenamiento no cambiarán. Si sus datos tienen componentes de tendencia o estacionalidad, planifique en consecuencia.

Nota: No es posible llamar directamente la preprocessing_fn. Esto significa que llamar preprocessing_fn(raw_data) no funcionará. En su lugar, se debe pasar a la API Transform Beam como se muestra en las siguientes celdas.

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,
    }

Sintaxis

Está casi listo para armar todo y usar Apache Beam para ejecutarlo.

Apache Beam usa una sintaxis especial para definir e invocar transformaciones. Por ejemplo, en esta línea:

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

El método to_this_call se invoca y se pasa el objeto llamado pass_this, y esta operación se conocerá como name this step en un seguimiento de pila. El resultado de la llamada a to_this_call se devuelve en result. A menudo verá etapas de una canalización encadenadas de esta manera:

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

y como comenzó con una nueva canalización, puede continuar así:

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

Unimos todo

Ahora estamos listos para transformar nuestros datos. Usaremos Apache Beam con un ejecutor directo y proporcionaremos tres entradas:

1. raw_data: los datos de entrada sin procesar que creamos anteriormente

2. raw_data_metadata: el esquema de los datos sin procesar

3. preprocessing_fn: la función que creamos para ejecutar nuestra transformación.

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

¿Es esta la respuesta correcta?

Anteriormente, usamos tf.Transform para hacer esto:

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: con la entrada de [1, 2, 3] la media de x es 2, y la restamos de x para centrar nuestros valores de x en 0. Entonces, nuestro resultado de [-1.0, 0.0, 1.0] es correcto.

• y_normalized: queríamos escalar nuestros valores de y entre 0 y 1. Nuestra entrada fue [1, 2, 3], por lo que nuestro resultado de [0.0, 0.5, 1.0] es correcto.

• s_integerized: queríamos asignar nuestras cadenas a índices en un vocabulario, y solo había 2 palabras en nuestro vocabulario ("hola" y "mundo"). Entonces, con la entrada de ["hello", "world", "hello"] nuestro resultado de [0, 1, 0] es correcto. Dado que "hola" aparece con mayor frecuencia en estos datos, será la primera entrada en el vocabulario.

• x_centered_times_y_normalized: queríamos crear una nueva característica al cruzar x_centered con y_normalized mediante una multiplicación. Tenga en cuenta que esto multiplica los resultados, no los valores originales, y nuestro nuevo resultado de [-0.0, 0.0, 1.0] es correcto.

Use la transform_fn resultante

!ls -l {output_dir}

El directorio transform_fn/ contiene una implementación tf.saved_model con todas las constantes de los resultados del análisis de transformación de tensorflow integrados en el grafo.

Es posible cargar esto directamente con tf.saved_model.load, pero no es fácil de usar:

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

Una mejor opción sería cargarlo con tft.TFTransformOutput. El método TFTransformOutput.transform_features_layer devuelve un objeto tft.TransformFeaturesLayer que se puede usar para aplicar la transformación:

tf_transform_output = tft.TFTransformOutput(output_dir)

tft_layer = tf_transform_output.transform_features_layer()
tft_layer

El objeto tft.TransformFeaturesLayer espera un diccionario de funciones por lotes. Por lo tanto, debe crear un Dict[str, tf.Tensor] de List[Dict[str, Any]] en 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),
}

Puede usar tft.TransformFeaturesLayer por sí solo:

transformed_batch = tft_layer(raw_data_batch)

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

Exporte

Un caso de uso más típico sería usar tf.Transform para aplicar la transformación a los conjuntos de datos de entrenamiento y evaluación (consulte el siguiente tutorial para ver un ejemplo). Luego, después del entrenamiento, antes de exportar el modelo, adjunte tft.TransformFeaturesLayer como primera capa para que pueda exportarlo como parte de su tf.saved_model. Continúe leyendo para ver un ejemplo concreto.

Un modelo de entrenamiento de ejemplo

A continuación, se muestra un modelo que cumple con las siguientes características:

1. toma el lote transformado,

2. los apila todos juntos en una matriz simple (batch, features),

3. los ejecuta a través de algunas capas densas, y

4. produce 10 salidas lineales.

En un caso de uso real, se aplicaría una codificación única a la función s_integerized.

Podría entrenar este modelo en un conjunto de datos 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()

Imaginemos que entrenamos el modelo.

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

Este modelo se ejecuta en las entradas transformadas.

trained_model_output = trained_model(transformed_batch)
trained_model_output.shape

Un ejemplo de envoltorio de exportación

Imaginemos que hemos entrenado el modelo anterior y queremos exportarlo.

Será conveniente incluir la función de transformación en el 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)

Este modelo combinado funciona con los datos sin procesar y produce exactamente los mismos resultados que llamar directamente al modelo entrenado:

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 incluye tft.TransformFeaturesLayer y es completamente autónomo. Puede guardarlo y restaurarlo en otro entorno y seguir obteniendo exactamente el mismo 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()