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 dataset Avro

Ver em TensorFlow.org

Executar no Google Colab

Ver fonte no GitHub

Baixar notebook

Visão geral

O objetivo da API de dataset Avro é carregar dados formatados em Avro nativamente no TensorFlow como um dataset do TensorFlow. O Avro é um sistema de serialização de dados similar ao Protocol Buffers. Ele é amplamente usado no Apache Hadoop e pode fornecer tanto um formato de serialização para dados persistentes quanto um formato para transmissão via rede para comunicação entre os nós do Hadoop. Os dados do Avro ficam em um formato de dados binário, compactado e em linhas. Ele depende do esquema armazenado como um arquivo JSON separado. Confira a especificação do formato Avro e a declaração do esquema no manual oficial.

Configure o pacote

Instale o pacote tensorflow-io obrigatório

!pip install tensorflow-io

Importe os pacotes

import tensorflow as tf
import tensorflow_io as tfio

Valide as importações de tf e tfio

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

Uso

Explore o dataset

Para este tutorial, vamos baixar o dataset Avro de exemplo.

Baixe um arquivo Avro de amostra:

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

Baixe o arquivo de esquema correspondente ao arquivo Avro de amostra:

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

No exemplo acima, um dataset Avro de teste foi criado com base no dataset MNIST. O dataset MNIST original no formato TFRecord é gerado a partir do dataset nomeado do TF. Porém, o dataset MNIST é grande demais para fins de demonstração. Por questões de simplicidade, a maior parte foi retirada, e somente os primeiros registros foram mantidos. Além disso, outra modificação foi feita para o campo image no dataset MNIST original, mapeado para o campo features (características) no Avro. Portanto, o arquivo Avro train.avro tem 4 registros, cada um com 3 campos: features, um array de int, label (rótulo), um int ou null, e dataType (tipo de dados), uma enumeração. Para ver o arquivo train.avro decodificado (Observação: o arquivo de dados Avro original não é legível por humanos, pois o Avro é um formato compactado):

Instale o pacote necessário para ler arquivos Avro:

!pip install avro

Para ler e exibir um arquivo Avro via print em um formato legível 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')

E o esquema de train.avro, representado por train.avsc, é um arquivo em formato JSON. Para ver 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')

Prepare o dataset

Carregue train.avro como dataset do TensorFlow usando a API de dataset 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("--------------------")

O exemplo acima converte train.avro em um dataset do TensorFlow. Cada elemento do dataset é um dicionário cuja chave é o nome da característica, e o valor é o tensor esparso ou denso convertido. Por exemplo, ele converte o campo features, label, dataType em um VarLenFeature(SparseTensor), FixedLenFeature(DenseTensor)e FixedLenFeature(DenseTensor), respectivamente. Como batch_size é 3, ele transforma 3 registros de train.avro em um elemento no dataset resultante. Para o primeiro registro em train.avro, cujo rótulo é null, o leitor do Avro o substitui pelo value(-100) especificado padrão. Neste exemplo, há 4 registros no total em train.avro. Como o tamanho do lote é 3, o dataset resultante contém 3 elementos, sendo que o tamanho de lote do último elemento é 1. Entretanto, se o usuário quiser descartar o último lote caso o tamanho seja menor do que o tamanho do lote, basta ativar drop_final_batch. Por exemplo:

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)

Também é possível aumentar num_parallel_reads (número de leituras paralelas) para acelerar o processamento de dados do Avro por meio do aumento de paralelismo de processamento/leitura.

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)

Confira o uso detalhado de make_avro_record_dataset na documentação da API.

Treine modelos do tf.keras com o dataset Avro

Agora, vamos conferir um exemplo completo de treinamento de modelos do tf.keras com o dataset Avro baseado no dataset MNIST.

Carregue train.avro como dataset do TensorFlow usando a API de dataset 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 um modelo simples do Keras:

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

model = build_and_compile_cnn_model()

Treine o modelo do Keras com o dataset 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)

O dataset Avro pode processar e transformar qualquer dado Avro em tensores do TensorFlow, incluindo registros em registros, mapas, arrays, ramificações e enumerações. As informações processadas são passadas para a implementação do dataset Avro como um mapa, em que as chaves codificam como processar os valores de dados e codificam como transformar os dados em tensores do TensorFlow, decidindo o tipo primitivo (por exemplo, bool, int, long, float, double, string) bem como o tipo de tensor (por exemplo, esparso ou denso). É fornecida uma lista de todos os parsers do TensorFlow (veja a tabela 1) e a transformação de tipos primitivos (veja a tabela 2).

Tabela 1 – Tipos de parser do TensorFlow disponíveis

Tipos de parser do TensorFlow	Tensores do TensorFlow	Explicação
tf.FixedLenFeature([], tf.int32)	tensor denso	Processa uma característica de comprimento fixo, ou seja, todas as linhas têm o mesmo número constante de elementos. Por exemplo: somente um elemento ou um array que sempre tem o mesmo número de elementos para cada linha
tf.SparseFeature(index_key=['key_1st_index', 'key_2nd_index'], value_key='key_value', dtype=tf.int64, size=[20, 50])	tensor esparso	Processa uma característica esparsa, em que cada linha tem uma lista de comprimento variável composta por índices e valores. 'index_key' identifica os índices; 'value_key' identifica o valor; 'dtype' é o tipo de dado; 'size' é o valor de índice máximo esperado para cada entrada de índice
tfio.experimental.columnar.VarLenFeatureWithRank([],tf.int64)	tensor esparso	Processa uma característica de comprimento variável, ou seja, cada linha de dados pode ter um número variável de elementos. Por exemplo: a linha 1 tem 5 elementos, e a linha 2 tem 7 elementos

Table 2 – Conversões de tipos Avro em tipos do TensorFlow permitidas:

Tipo primitivo Avro	Tipo primitivo TensorFlow
boolean: valor binário	tf.bool
bytes: sequência de bytes de 8 bits sem sinal	tf.string
double: número de ponto flutuante IEEE com precisão dupla de 64 bits	tf.float64
enum: tipo enumeração	tf.string usando o mesmo nome de símbolo
float: número de ponto flutuante IEEE com precisão simples de 32 bits	tf.float32
int: inteiro de 32 bits com sinal	tf.int32
long: inteiro de 64 bits com sinal	tf.int64
null: sem valor	usa o valor padrão
string: sequência de caracteres Unicode	tf.string

Veja um conjunto abrangente de exemplos da API de dataset Avro nos testes.