avro.ipynb

Kernel: Python 3

Copyright 2020 The TensorFlow IO 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.

API de conjunto de datos de Avro

Ver en TensorFlow.org

Ejecutar en Google Colab

Ver código fuente en GitHub

Descargar el bloc de notas

Descripción general

El objetivo de la API del conjunto de datos de Avro es cargar datos formateados en Avro de forma nativa en TensorFlow como un conjunto de datos de TensorFlow. Avro es un sistema de serialización de datos similar a Protocol Buffers. Se usa mucho en Apache Hadoop, donde puede proporcionar un formato de serialización para datos persistentes y un formato de conexión para la comunicación entre nodos de Hadoop. Los datos de Avro son un formato de datos binarios compactados y orientados a filas. Se basa en un esquema que se almacena como un archivo JSON independiente. Para conocer las especificaciones del formato Avro y la declaración del esquema, consulte el manual oficial.

Paquete de preparación

Instalar el paquete tensorflow-io necesario

!pip install tensorflow-io

Importe paquetes

import tensorflow as tf
import tensorflow_io as tfio

Validar las importaciones de tf y de tfio

print("tensorflow-io version: {}".format(tfio.__version__))
print("tensorflow version: {}".format(tf.__version__))

Uso

Explorar el conjunto de datos

Para los fines de este tutorial, descarguemos el conjunto de datos de muestra de Avro.

Descargue un archivo Avro de muestra:

!curl -OL https://github.com/tensorflow/io/raw/master/docs/tutorials/avro/train.avro
!ls -l train.avro

Descargue el archivo de esquema correspondiente del archivo Avro de muestra:

!curl -OL https://github.com/tensorflow/io/raw/master/docs/tutorials/avro/train.avsc
!ls -l train.avsc

En el ejemplo anterior, se creó un conjunto de datos de Avro de prueba basado en el conjunto de datos mnist. El conjunto de datos mnist original en formato TFRecord se genera a partir del conjunto de datos con nombre TF. Sin embargo, el conjunto de datos mnist es demasiado grande como conjunto de datos de demostración. Para simplificar, se recortó bastante y solo se conservaron los primeros registros. Además, se realizó un recorte adicional para el campo image en el conjunto de datos original de mnist y se asignó al campo features en Avro. Entonces, el archivo avro train.avro tiene 4 registros, cada uno de los cuales tiene 3 campos: features, que es un arreglo de int, label, un int o nulo y dataType, una enum. Para ver el train.avro decodificado (tenga en cuenta que el archivo de datos de Avro original no es legible por humanos ya que avro es un formato compactado):

Instale el paquete requerido para leer el archivo Avro:

!pip install avro

Para leer e imprimir un archivo Avro en un formato legible por humanos:

from avro.io import DatumReader
from avro.datafile import DataFileReader

import json

def print_avro(avro_file, max_record_num=None):
    if max_record_num is not None and max_record_num <= 0:
        return

    with open(avro_file, 'rb') as avro_handler:
        reader = DataFileReader(avro_handler, DatumReader())
        record_count = 0
        for record in reader:
            record_count = record_count+1
            print(record)
            if max_record_num is not None and record_count == max_record_num:
               break

print_avro(avro_file='train.avro')

Y el esquema de train.avro representado por train.avsc es un archivo con formato JSON. Para ver el train.avsc:

def print_schema(avro_schema_file):
    with open(avro_schema_file, 'r') as handle:
        parsed = json.load(handle)
    print(json.dumps(parsed, indent=4, sort_keys=True))

print_schema('train.avsc')

Preparar el conjunto de datos

Cargue train.avro como conjunto de datos de TensorFlow con la API del conjunto de datos de Avro:

features = {
    'features[*]': tfio.experimental.columnar.VarLenFeatureWithRank(dtype=tf.int32),
    'label': tf.io.FixedLenFeature(shape=[], dtype=tf.int32, default_value=-100),
    'dataType': tf.io.FixedLenFeature(shape=[], dtype=tf.string)
}

schema = tf.io.gfile.GFile('train.avsc').read()

dataset = tfio.experimental.columnar.make_avro_record_dataset(file_pattern=['train.avro'],
                                                              reader_schema=schema,
                                                              features=features,
                                                              shuffle=False,
                                                              batch_size=3,
                                                              num_epochs=1)

for record in dataset:
    print(record['features[*]'])
    print(record['label'])
    print(record['dataType'])
    print("--------------------")

En el ejemplo anterior se convierte train.avro en un conjunto de datos de TensorFlow. Cada elemento del conjunto de datos es un diccionario cuya clave es el nombre de la función, el valor es el tensor denso o disperso convertido. Por ejemplo, convierte el campo features, label y dataType en VarLenFeature(SparseTensor), FixedLenFeature(DenseTensor) y FixedLenFeature(DenseTensor) respectivamente. Dado que batch_size es 3, coacciona 3 registros de train.avro en un elemento en el conjunto de datos resultante. Para el primer registro en train.avro cuya etiqueta es nula, el lector de Avro lo reemplaza con el valor predeterminado especificado (-100). En este ejemplo, hay 4 registros en total en train.avro. Dado que el tamaño del lote es 3, el conjunto de datos resultante contiene 3 elementos, el último de los cuales tiene un tamaño de lote de 1. Sin embargo, el usuario también puede descartar el último lote si el tamaño es menor que el tamaño del lote habilitando drop_final_batch. Por ejemplo:

dataset = tfio.experimental.columnar.make_avro_record_dataset(file_pattern=['train.avro'],
                                                              reader_schema=schema,
                                                              features=features,
                                                              shuffle=False,
                                                              batch_size=3,
                                                              drop_final_batch=True,
                                                              num_epochs=1)

for record in dataset:
    print(record)

También se puede aumentar num_parallel_reads para acelerar el procesamiento de datos de Avro aumentando el paralelismo de parseo/lectura de Avro.

dataset = tfio.experimental.columnar.make_avro_record_dataset(file_pattern=['train.avro'],
                                                              reader_schema=schema,
                                                              features=features,
                                                              shuffle=False,
                                                              num_parallel_reads=16,
                                                              batch_size=3,
                                                              drop_final_batch=True,
                                                              num_epochs=1)

for record in dataset:
    print(record)

Para conocer el uso detallado de make_avro_record_dataset, consulte el documento de API.

Entrenar modelos tf.keras con el conjunto de datos de Avro

Ahora veamos un ejemplo de principio a fin de entrenamiento del modelo tf.keras con un conjunto de datos de Avro basado en un conjunto de datos mnist.

Cargue train.avro como conjunto de datos de TensorFlow con la API del conjunto de datos de Avro:

features = {
    'features[*]': tfio.experimental.columnar.VarLenFeatureWithRank(dtype=tf.int32),
    'label': tf.io.FixedLenFeature(shape=[], dtype=tf.int32, default_value=-100),
}


schema = tf.io.gfile.GFile('train.avsc').read()

dataset = tfio.experimental.columnar.make_avro_record_dataset(file_pattern=['train.avro'],
                                                              reader_schema=schema,
                                                              features=features,
                                                              shuffle=False,
                                                              batch_size=1,
                                                              num_epochs=1)

Defina un modelo keras simple:

def build_and_compile_cnn_model():
    model = tf.keras.Sequential()
    model.compile(optimizer='sgd', loss='mse')
    return model

model = build_and_compile_cnn_model()

Entrene el modelo keras con el conjunto de datos de Avro:

def extract_label(feature):
  label = feature.pop('label')
  return tf.sparse.to_dense(feature['features[*]']), label

model.fit(x=dataset.map(extract_label), epochs=1, steps_per_epoch=1, verbose=1)

El conjunto de datos de Avro puede parsear y convertir cualquier dato de Avro en tensores de TensorFlow, incluidos registros en registros, mapas, arreglos, ramas y enumeraciones. La información de parseo se pasa a la implementación del conjunto de datos de Avro como un mapa donde las claves codifican cómo parsear los valores de datos y codifican cómo convertir los datos en tensores de TensorFlow. Así se decide el tipo primitivo (por ejemplo, bool, int, long, float, double, string) así como también el tipo de tensor (por ejemplo, disperso o denso). A continuación, se proporciona una lista de los tipos de parseadores de TensorFlow (consulte la Tabla 1) y la coerción de los tipos primitivos (Tabla 2).

Tabla 1: tipos de parseadores de TensorFlow admitidos:

Tipos de parseadores de TensorFlow	Tensores de TensorFlow	Explicación
tf.FixedLenFeature([], tf.int32)	tensor denso	Parsee una característica de longitud fija; es decir, todas las filas tienen una misma cantidad constante de elementos, por ejemplo, solo un elemento o un arreglo que siempre tiene la misma cantidad de elementos para cada fila.
tf.SparseFeature(index_key=['key_1st_index', 'key_2nd_index'], value_key='key_value', dtype=tf.int64, size=[20, 50])	tensor disperso	Parsee una característica dispersa donde cada fila tiene una lista de índices y valores de longitud variable. La 'index_key' identifica los índices. La 'value_key' identifica el valor. El 'dtype' es el tipo de datos. El 'size' es el valor de índice máximo esperado para cada entrada de índice.
tfio.experimental.columnar.VarLenFeatureWithRank([],tf.int64)	tensor disperso	Parsee una característica de longitud variable; eso significa que cada fila de datos puede tener una cantidad variable de elementos, por ejemplo, la primera fila tiene 5 elementos, la segunda fila tiene 7 elementos

Tabla 2, conversión admitida de tipos Avro a tipos de TensorFlow:

Tipo primitivo de Avro	Tipo primitivo de TensorFlow
booleano: un valor binario	tf.bool
bytes: una secuencia de bytes de 8 bits sin signo	tf.string
double: número de punto flotante IEEE de 64 bits de doble precisión	tf.float64
enum: tipo de enumeración	tf.string con el nombre del símbolo
float: número de punto flotante IEEE de 32 bits de precisión simple	tf.float32
int: entero de 32 bits con signo	tf.int32
int: entero de 64 bits con signo	tf.int64
null: sin valor	usa el valor predeterminado
string: secuencia de caracteres de Unicode	tf.string

En las pruebas puede encontrar un conjunto completo de ejemplos de la API del conjunto de datos de Avro.