Se usó la API de Cloud Translation para traducir esta página.
Switch to English

Comience con TensorFlow Transform

Esta guía presenta los conceptos básicos de tf.Transform y cómo usarlos. Va a:

  • Defina una función de preprocesamiento , una descripción lógica de la tubería que transforma los datos sin procesar en los datos utilizados para entrenar un modelo de aprendizaje automático.
  • Muestre la implementación de Apache Beam utilizada para transformar datos mediante la conversión de la función de preprocesamiento en una canalización de Beam .
  • Mostrar ejemplos de uso adicionales.

Definir una función de preprocesamiento

La función de preprocesamiento es el concepto más importante de tf.Transform . La función de preprocesamiento es una descripción lógica de una transformación del conjunto de datos. La función de preprocesamiento acepta y devuelve un diccionario de tensores, donde un tensor significa Tensor o SparseTensor . Hay dos tipos de funciones que se utilizan para definir la función de preprocesamiento:

  1. Cualquier función que acepte y devuelva tensores. Estos agregan operaciones TensorFlow al gráfico que transforman los datos sin procesar en datos transformados.
  2. Cualquiera de los analizadores proporcionados por tf.Transform . Los analizadores también aceptan y devuelven tensores, pero a diferencia de las funciones TensorFlow, no agregan operaciones al gráfico. En cambio, los analizadores hacen que tf.Transform calcule una operación de paso completo fuera de TensorFlow. Utilizan los valores del tensor de entrada en todo el conjunto de datos para generar un tensor constante que se devuelve como salida. Por ejemplo, tft.min calcula el mínimo de un tensor sobre el conjunto de datos. tf.Transform proporciona un conjunto fijo de analizadores, pero esto se extenderá en futuras versiones.

Ejemplo de función de preprocesamiento

Al combinar analizadores y funciones regulares de TensorFlow, los usuarios pueden crear canales flexibles para transformar datos. La siguiente función de preprocesamiento transforma cada una de las tres características de diferentes maneras y combina dos de las características:

 import tensorflow as tf
import tensorflow_transform as tft
import tensorflow_transform.beam as tft_beam

def preprocessing_fn(inputs):
  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,
      'x_centered_times_y_normalized': x_centered_times_y_normalized,
      's_integerized': s_integerized
  }
 

Aquí, x , y y s son Tensor s que representan características de entrada. El primer tensor nuevo que se crea, x_centered , se tft.mean aplicando tft.mean a x y restando esto de x . tft.mean(x) devuelve un tensor que representa la media del tensor x . x_centered es el tensor x con la media restada.

El segundo tensor nuevo, y_normalized , se crea de manera similar pero utilizando el método de conveniencia tft.scale_to_0_1 . Este método hace algo similar a calcular x_centered , es decir, calcular un máximo y un mínimo y usarlos para escalar y .

El tensor s_integerized muestra un ejemplo de manipulación de cadenas. En este caso, tomamos una cadena y la asignamos a un número entero. Utiliza la función de conveniencia tft.compute_and_apply_vocabulary . Esta función usa un analizador para calcular los valores únicos tomados por las cadenas de entrada, y luego usa las operaciones TensorFlow para convertir las cadenas de entrada en índices en la tabla de valores únicos.

La columna final muestra que es posible usar operaciones TensorFlow para crear nuevas características combinando tensores.

La función de preprocesamiento define una tubería de operaciones en un conjunto de datos. Para aplicar la canalización, confiamos en una implementación concreta de la API tf.Transform . La implementación de Apache Beam proporciona PTransform que aplica la función de preprocesamiento de un usuario a los datos. El flujo de trabajo típico de un usuario de tf.Transform construirá una función de preprocesamiento, luego la incorporará a una tubería de Beam más grande, creando los datos para la capacitación.

Procesamiento por lotes

El procesamiento por lotes es una parte importante de TensorFlow. Dado que uno de los objetivos de tf.Transform es proporcionar un gráfico TensorFlow para el preprocesamiento que se puede incorporar en el gráfico de servicio (y, opcionalmente, el gráfico de capacitación), el procesamiento por lotes también es un concepto importante en tf.Transform .

Si bien no es obvio en el ejemplo anterior, a la función de preprocesamiento definida por el usuario se le pasan tensores que representan lotes y no instancias individuales, como sucede durante el entrenamiento y el servicio con TensorFlow. Por otro lado, los analizadores realizan un cálculo sobre todo el conjunto de datos que devuelve un solo valor y no un lote de valores. x es un Tensor con forma de (batch_size,) , mientras que tft.mean(x) es un Tensor con forma de () . La resta x - tft.mean(x) difunde donde el valor de tft.mean(x) se resta de cada elemento del lote representado por x .

Implementación de Apache Beam

Si bien la función de preprocesamiento pretende ser una descripción lógica de una tubería de preprocesamiento implementada en múltiples marcos de procesamiento de datos, tf.Transform proporciona una implementación canónica utilizada en Apache Beam. Esta implementación demuestra la funcionalidad requerida de una implementación. No hay una API formal para esta funcionalidad, por lo que cada implementación puede usar una API que sea idiomática para su marco de procesamiento de datos particular.

La implementación de Apache Beam proporciona dos PTransform s utilizados para procesar datos para una función de preprocesamiento. A continuación se muestra el uso del PTransform AnalyzeAndTransformDataset :

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

raw_data_metadata = ...
transformed_dataset, transform_fn = (
    (raw_data, raw_data_metadata) | tft_beam.AnalyzeAndTransformDataset(
        preprocessing_fn))
transformed_data, transformed_metadata = transformed_dataset
 

El contenido de Datos transformed_data se muestra a continuación y contiene las columnas transformadas en el mismo formato que los datos sin procesar. En particular, los valores de s_integerized son [0, 1, 0] estos valores dependen de cómo se mapearon las palabras hello y world a los enteros, lo cual es determinista. Para la columna x_centered , restamos la media para que los valores de la columna x , que eran [1.0, 2.0, 3.0] , se convirtieran en [-1.0, 0.0, 1.0] . Del mismo modo, el resto de las columnas coinciden con sus valores esperados.

 [{u's_integerized': 0,
  u'x_centered': -1.0,
  u'x_centered_times_y_normalized': -0.0,
  u'y_normalized': 0.0},
 {u's_integerized': 1,
  u'x_centered': 0.0,
  u'x_centered_times_y_normalized': 0.0,
  u'y_normalized': 0.5},
 {u's_integerized': 0,
  u'x_centered': 1.0,
  u'x_centered_times_y_normalized': 1.0,
  u'y_normalized': 1.0}]
 

Ambos raw_data y transformed_data son conjuntos de datos. Las siguientes dos secciones muestran cómo la implementación de Beam representa conjuntos de datos y cómo leer y escribir datos en el disco. El otro valor de retorno, transform_fn , representa la transformación aplicada a los datos, que se detalla a continuación.

AnalyzeAndTransformDataset es la composición de las dos transformaciones fundamentales proporcionadas por la implementación AnalyzeDataset y TransformDataset . Por lo tanto, los siguientes dos fragmentos de código son equivalentes:

 transformed_data, transform_fn = (
    my_data | tft_beam.AnalyzeAndTransformDataset(preprocessing_fn))
 
 transform_fn = my_data | tft_beam.AnalyzeDataset(preprocessing_fn)
transformed_data = (my_data, transform_fn) | tft_beam.TransformDataset()
 

transform_fn es una función pura que representa una operación que se aplica a cada fila del conjunto de datos. En particular, los valores del analizador ya se calculan y se tratan como constantes. En el ejemplo, transform_fn contiene como constantes la media de la columna x , el mínimo y el máximo de la columna y , y el vocabulario utilizado para asignar las cadenas a enteros.

Una característica importante de tf.Transform es que transform_fn representa un mapa sobre filas: es una función pura que se aplica a cada fila por separado. Todo el cálculo para agregar filas se realiza en AnalyzeDataset . Además, transform_fn se representa como un Graph TensorFlow que se puede incrustar en el gráfico de servicio.

AnalyzeAndTransformDataset se proporciona para optimizaciones en este caso especial. Este es el mismo patrón utilizado en scikit-learn , que proporciona los métodos fit , transform y fit_transform .

Formatos de datos y esquema

En los ejemplos de código anteriores, se omite el código que define raw_data_metadata . Los metadatos contienen el esquema que define el diseño de los datos para que se lean y escriban en varios formatos. Incluso el formato en memoria que se muestra en la última sección no se describe a sí mismo y requiere que el esquema se interprete como tensores.

Aquí está la definición del esquema para los datos de ejemplo:

 from tensorflow_transform.tf_metadata import dataset_metadata
from tensorflow_transform.tf_metadata import schema_utils

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

La clase dataset_schema.Schema contiene la información necesaria para analizar los datos de su formato en disco o en memoria, en tensores. Por lo general, se construye llamando a schema_utils.schema_from_feature_spec con una clave de función de asignación dict a tf.io.FixedLenFeature , tf.io.VarLenFeature y tf.io.SparseFeature . Consulte la documentación de tf.parse_example para obtener más detalles.

Arriba usamos tf.io.FixedLenFeature para indicar que cada característica contiene un número fijo de valores, en este caso un solo valor escalar. Debido a que tf.Transform lotes de instancias, el Tensor real que representa la entidad tendrá forma (None,) donde la dimensión desconocida es la dimensión de lote.

Entrada y salida con Apache Beam

Hasta ahora, el formato de datos para los ejemplos utilizaba listas de diccionarios. Esta es una simplificación que se basa en la capacidad de Apache Beam para trabajar con listas, así como su representación principal de datos, la PCollection . Una PCollection es una representación de datos que forma parte de una tubería de Beam. Se forma una tubería Beam aplicando varios PTransform s, incluidos AnalyzeDataset y TransformDataset , y ejecutando la tubería. PCollection se crea una PCollection en la memoria del binario principal, sino que se distribuye entre los trabajadores (aunque esta sección utiliza el modo de ejecución en memoria).

El siguiente ejemplo requiere leer y escribir datos en el disco y representar los datos como una PCollection (no una lista), consulte: census_example.py . A continuación mostramos cómo descargar los datos y ejecutar este ejemplo. El conjunto de datos "Ingresos del censo" es proporcionado por el Depósito de aprendizaje automático de UCI . Este conjunto de datos contiene datos categóricos y numéricos.

Los datos están en formato CSV, aquí están las dos primeras líneas:

 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
 

Las columnas del conjunto de datos son categóricas o numéricas. Como hay muchas columnas, se genera un Schema (similar al ejemplo anterior) recorriendo todas las columnas de cada tipo. Este conjunto de datos describe un problema de clasificación: predecir la última columna donde el individuo gana más o menos de 50K por año. Sin embargo, desde la perspectiva de tf.Transform , esta etiqueta es solo otra columna categórica.

Use este esquema para leer los datos del archivo CSV. La constante ordered_columns contiene la lista de todas las columnas en el orden en que aparecen en el archivo CSV, requerido porque el esquema no contiene esta información. Algunas transformaciones de Beam adicionales se eliminan ya que ya se realizaron al leer el archivo CSV. Cada fila CSV se convierte en una instancia en el formato en memoria.

En este ejemplo, permitimos que falte la función education-num . Esto significa que se representa como un tf.io.VarLenFeature en feature_spec, y como tf.SparseTensor en preprocessing_fn. Para manejar el valor de la característica posiblemente faltante, completamos las instancias faltantes con un valor predeterminado, en este caso 0.

 converter = tft.coders.CsvCoder(ordered_columns, raw_data_schema)

raw_data = (
    p
    | 'ReadTrainData' >> textio.ReadFromText(train_data_file)
    | ...
    | 'DecodeTrainData' >> beam.Map(converter.decode))
 

El preprocesamiento es similar al ejemplo anterior, excepto que la función de preprocesamiento se genera mediante programación en lugar de especificar manualmente cada columna. En la siguiente función de preprocesamiento, NUMERICAL_COLUMNS y CATEGORICAL_COLUMNS son listas que contienen los nombres de las columnas numéricas y categóricas:

 def preprocessing_fn(inputs):
  """Preprocess input columns into transformed columns."""
  # Since we are modifying some features and leaving others unchanged, we
  # start by setting `outputs` to a copy of `inputs.
  outputs = inputs.copy()

  # Scale numeric columns to have range [0, 1].
  for key in NUMERIC_FEATURE_KEYS:
    outputs[key] = tft.scale_to_0_1(outputs[key])

  for key in OPTIONAL_NUMERIC_FEATURE_KEYS:
    # This is a SparseTensor because it is optional. Here we fill in a default
    # value when it is missing.
      sparse = tf.sparse.SparseTensor(outputs[key].indices, outputs[key].values,
                                      [outputs[key].dense_shape[0], 1])
      dense = tf.sparse.to_dense(sp_input=sparse, default_value=0.)
    # Reshaping from a batch of vectors of size 1 to a batch to scalars.
    dense = tf.squeeze(dense, axis=1)
    outputs[key] = tft.scale_to_0_1(dense)

  # For all categorical columns except the label column, we generate a
  # vocabulary but do not modify the feature.  This vocabulary is instead
  # used in the trainer, by means of a feature column, to convert the feature
  # from a string to an integer id.
  for key in CATEGORICAL_FEATURE_KEYS:
    tft.vocabulary(inputs[key], vocab_filename=key)

  # For the label column we provide the mapping from string to index.
  initializer = tf.lookup.KeyValueTensorInitializer(
      keys=['>50K', '<=50K'],
      values=tf.cast(tf.range(2), tf.int64),
      key_dtype=tf.string,
      value_dtype=tf.int64)
  table = tf.lookup.StaticHashTable(initializer, default_value=-1)

  outputs[LABEL_KEY] = table.lookup(outputs[LABEL_KEY])

  return outputs
 

Una diferencia con respecto al ejemplo anterior es que la columna de etiqueta especifica manualmente la asignación de la cadena a un índice. Entonces '>50' se asigna a 0 y '<=50K' se asigna a 1 porque es útil saber qué índice en el modelo entrenado corresponde a qué etiqueta.

La variable raw_data representa una PCollection que contiene datos en el mismo formato que la lista raw_data (del ejemplo anterior), usando la misma transformación AnalyzeAndTransformDataset . El esquema se usa en dos lugares: leer los datos del archivo CSV y como entrada a AnalyzeAndTransformDataset . Tanto el formato CSV como el formato en memoria deben combinarse con un esquema para interpretarlos como tensores.

La etapa final es escribir los datos transformados en el disco y tiene una forma similar a la lectura de los datos sin procesar. El esquema utilizado para hacer esto es parte de la salida de AnalyzeAndTransformDataset que infiere un esquema para los datos de salida. El código para escribir en el disco se muestra a continuación. El esquema es parte de los metadatos pero usa los dos de manera intercambiable en la API tf.Transform (es decir, pasa los metadatos al ExampleProtoCoder ). Tenga en cuenta que esto escribe en un formato diferente. En lugar de textio.WriteToText , use el soporte integrado de Beam para el formato TFRecord y use un codificador para codificar los datos como protos de Example . Este es un mejor formato para usar en la capacitación, como se muestra en la siguiente sección. transformed_eval_data_base proporciona el nombre de archivo base para los fragmentos individuales que se escriben.

 transformed_data | "WriteTrainData" >> tfrecordio.WriteToTFRecord(
    transformed_eval_data_base,
    coder=tft.coders.ExampleProtoCoder(transformed_metadata))
 

Además de los datos de entrenamiento, transform_fn también se escribe con los metadatos:

 _ = (
    transform_fn
    | 'WriteTransformFn' >> tft_beam.WriteTransformFn(working_dir))
transformed_metadata | 'WriteMetadata' >> tft_beam.WriteMetadata(
    transformed_metadata_file, pipeline=p)
 

Ejecute toda la canalización de Beam con p.run().wait_until_finish() . Hasta este punto, la tubería Beam representa un cálculo diferido y distribuido. Proporciona instrucciones para lo que se hará, pero las instrucciones no se han ejecutado. Esta última llamada ejecuta la canalización especificada.

Descargar el conjunto de datos del censo

Descargue el conjunto de datos del censo utilizando los siguientes comandos de shell:

  wget https://archive.ics.uci.edu/ml/machine-learning-databases/adult/adult.data
  wget https://archive.ics.uci.edu/ml/machine-learning-databases/adult/adult.test

Al ejecutar el script census_example.py , pase el directorio que contiene estos datos como primer argumento. El script crea un subdirectorio temporal para agregar los datos preprocesados.

Integrar con el entrenamiento TensorFlow

La sección final de census_example.py muestra cómo se utilizan los datos preprocesados ​​para entrenar un modelo. Consulte la documentación de Estimadores para más detalles. El primer paso es construir un Estimator que requiera una descripción de las columnas preprocesadas. Cada columna numérica se describe como una real_valued_column que es un contenedor alrededor de un vector denso con un tamaño fijo ( 1 en este ejemplo). Cada columna categórica se asigna de una cadena a enteros y luego se pasa al indicator_column . tft.TFTransformOutput se utiliza para encontrar la ruta del archivo de vocabulario para cada característica categórica.

 real_valued_columns = [feature_column.real_valued_column(key)
                       for key in NUMERIC_FEATURE_KEYS]

one_hot_columns = [
    tf.feature_column.indicator_column(
        tf.feature_column.categorical_column_with_vocabulary_file(
            key=key,
            vocabulary_file=tf_transform_output.vocabulary_file_by_name(
                vocab_filename=key)))
    for key in CATEGORICAL_FEATURE_KEYS]

estimator = tf.estimator.LinearClassifier(real_valued_columns + one_hot_columns)
 

El siguiente paso es crear un generador para generar la función de entrada para capacitación y evaluación. Esto difiere del entrenamiento utilizado por tf.Learn ya que no se requiere una especificación de función para analizar los datos transformados. En su lugar, use los metadatos de los datos transformados para generar una especificación de característica.

 def _make_training_input_fn(tf_transform_output, transformed_examples,
                            batch_size):
  ...
  def input_fn():
    """Input function for training and eval."""
    dataset = tf.data.experimental.make_batched_features_dataset(
        ..., tf_transform_output.transformed_feature_spec(), ...)

    transformed_features = tf.compat.v1.data.make_one_shot_iterator(
        dataset).get_next()
    ...

  return input_fn
 

El código restante es el mismo que usar la clase Estimator . El ejemplo también contiene código para exportar el modelo en el formato SavedModel . El modelo exportado puede ser utilizado por Tensorflow Serving o Cloud ML Engine .