Chapter 13 - Loading and Preprocessing Data with TensorFlow

dataset = tf.data.Dataset.range(10).repeat(3) # 0 to 9, three times
dataset = dataset.shuffle(buffer_size=5, seed=42).batch(7)

filepath_dataset = tf.data.Dataset.list_files(train_filepaths, seed=42)
n_readers = 5
dataset = filepath_dataset.interleave(
    lambda filepath: tf.data.TextLineDataset(filepath).skip(1),
    cycle_length=n_readers)

X_mean, X_std = [...] # mean and scale of each feature in the training set
n_inputs = 8
def preprocess(line):
    defs = [0.] * n_inputs + [tf.constant([], dtype=tf.float32)]
    fields = tf.io.decode_csv(line, record_defaults=defs)
    x = tf.stack(fields[:-1])
    y = tf.stack(fields[-1:])
    return (x - X_mean) / X_std, y # (X,y) tuple compatible with Keras model

def csv_reader_dataset(filepaths, repeat=1, n_readers=5,
                        n_read_threads=None, shuffle_buffer_size=10000,
                        n_parse_threads=5, batch_size=32):
    dataset = tf.data.Dataset.list_files(filepaths).repeat(repeat)
    dataset = dataset.interleave(
        lambda filepath: tf.data.TextLineDataset(filepath).skip(1),
        cycle_length=n_readers, num_parallel_calls=n_read_threads)
    dataset = dataset.map(preprocess, num_parallel_calls=n_parse_threads)
    dataset = dataset.shuffle(shuffle_buffer_size)
    return dataset.batch(batch_size).prefetch(1)

• other functions from TF Data
  • concatenate() Creates a Dataset by concatenating given dataset with this dataset.
  • filter() Filters this dataset according to predicate.
  • flat_map() Maps map_func across this dataset and flattens the result.
  • from_generator() Creates a Dataset whose elements are generated by generator.
  • from_tensors() Creates a Dataset with a single element, comprising the given tensors.
  • padded_batch() Combines consecutive elements of this dataset into padded batches.
  • shard() Creates a Dataset that includes only 1/num_shards of this dataset.
  • window() Combines input elements into a dataset of windows.
  • zip() Creates a Dataset by zipping together the given datasets.

TFRecord Format

• if loading/parsing from e.g. csv files is the bottle-neck, consider TFRecord
• efficient serialized data format
• can be used in TF Functions (with TF protocol buffer data structures)
• protocol buffer: "protobuf", defines format for serial data
• use TF Protobufs (Example, SequenceExample) so data can be serialized/parsed with TF functions ()
• typical workflow for writing to TFRecord:
  • read data from CSV to dataset
  • create Example protobuf for each instance
  • serialize the dataset
  • save to several TFRecord files with shuffling
• typical workflow for loading and parsing from TFRecord:
  • define description dictionary of serialized data in the TFRecord
  • read data in batches from TFRecord using TFRecordDataset class
  • parse the examples in each batch

feature_description = {
    "name": tf.io.FixedLenFeature([], tf.string, default_value=""),
    "id": tf.io.FixedLenFeature([], tf.int64, default_value=0),
    "emails": tf.io.VarLenFeature(tf.string),
}

dataset = tf.data.TFRecordDataset(["my_contacts.tfrecord"]).batch(10)
for serialized_examples in dataset:
    parsed_examples = tf.io.parse_example(serialized_examples,
                                          feature_description)

• any tensor can be stored parsed directly from a ByteList in an Example protobuff using tf.io.serialize_tensor() and tf.io.parse:tensor()

Preprocessing Input Features

• convert categorical features to numeric features, normalize, discretize, etc.
• can be done with Data API (e.g. .map()) but better to implement as custom Keras layer or use existing ones (self-contained, flexible, easy to incorporate into a model or preprocessing pipeline)
• encode categorical features to one-hot vectors
  • usually okay for <10 features, but doesn't scale well
  • create lookup table from category to index, including extra "out of vocabulary buckets" for categories not encountered when making the lookup table
  • convert to one-hot vectors with tf.one_hot(cat_indicies, depth=...)
  • bundle functionality into a custom layer, or use experimental keras.layers.TextVectorization:
    • .adapt(data_sample): extract data categories and create lookup table
    • .call(train.data): use lookup table to map categories to indices
    • use tf.one_hot() in a keras Lambda layer to vectorize indices

• keras preprocessing layers
  • Normalization
  • TextVectorization (experimental)
    • encode word to index (make lookup table)
    • not only integer indicies, but also counts for each word index ("bag of words)
  • Discretization: split continuous feature into bins
  • PreprocessingStage: combine multiple preprocessing layers

normalization = keras.layers.Normalization()
discretization = keras.layers.Discretization([...])
pipeline = keras.layers.PreprocessingStage([normalization, discretization])
pipeline.adapt(data_sample)

TF Transform

• if on the fly preprocessing is too slow, it may be better to do it all in advance of training (and avoid repetition each epoch)
• defining preprocessing function using TF Transfrom gives a TF Function that can be deployed across platforms with the TF model

import tensorflow_transform as tft

def preprocess(inputs): # inputs = a batch of input features
    median_age = inputs["housing_median_age"]
    ocean_proximity = inputs["ocean_proximity"]
    standardized_age = tft.scale_to_z_score(median_age)
    ocean_proximity_id = tft.compute_and_apply_vocabulary(ocean_proximity)
    return {
        "standardized_median_age": standardized_age,
        "ocean_proximity_id": ocean_proximity_id
    }

• Done on the entire dataset, so need to account for limited RAM (look at Apache Beam, Spark to handle large datasets)

TF Datasets

• lots of common datasets
• datasets load as dictionaries, so use .map() to convert to (X,y) tuple for Keras, or specify as_supervised=True when loading

import tensorflow_datasets as tfds

datasets = tfds.load(name="mnist", batch_size=32, as_supervised=True)
mnist_train = datasets["train"].repeat().prefetch(1)
model = keras.models.Sequential(...)
model.compile(...)
model.fit(mnist_train, epochs=5)