01. MLOps: Project Settings

The goal of the series is to understand the basics of MLOps (model building, monitoring, configurations, testing, packaging, deployment, cicd). As a first step, Let's setup the project.

Raviraja Ganta is using NLP project in this series but I will be using Vanilla GAN Projects for this project.

https://papers.nips.cc/paper/5656-hidden-technical-debt-in-machine-learning-systems.pdf(https://papers.nips.cc/paper/5656-hidden-technical-debt-in-machine-learning-systems.pdf)

In this post, I will be going through the following topics:

Examine and understand data
process the data
Build an input pipeline
Build the model
Train the model
the inference

Note: Basic knowledge of Machine Learning is needed

Project Structure

Following is structure of this chapter. We will go through this.

01.project_setup
 ┣ experimental_notebooks
 ┣ README.md
 ┣ data.py
 ┣ inference.py
 ┣ model.py
 ┣ requirements.txt
 ┗ train.py

Deep Learning Library

There are many libraries available to develop deep learning projects. The prominent ones are:

Tensorflow
Pytorch
Pytorch Lightning

etc.

We will be using Tensorflow because its more convenient to use in real-life services.

This is final directory for this post. code can be found here

01.project_setup
 ┣ callbacks
 ┃ ┗ callback.py
 ┣ configs
 ┃ ┗ flower_photos.yml
 ┣ datasets
 ┃ ┣ flower_photos
 ┃ ┣ abstract_dataset.py
 ┃ ┣ flower_photos.py
 ┣ logs
 ┣ models
 ┃ ┣ flower_classifier.py
 ┃ ┗ model.py
 ┣ trainers
 ┃ ┗ trainer.py
 ┣ utils
 ┃ ┣ config.py
 ┃ ┗ data_utils.py
 ┣ README.md
 ┣ requirements.txt
 ┗ run.py

Download and Load data

Data pipelines can be created with:

Vanilla Pytorch DataLoaders
Pytorch Lightning DataModules

DataModules are more structured definition, which allows for additional optimizations such as automated distribution of workload between CPU & GPU. Using DataModules is recommended whenever possible!

A DataModule is defined by an interface:

prepare_data (optional) which is called only once and on 1 GPU -- typically something like the data download step we have below
train_dataloader, val_dataloader to load each dataset

A DataModule encapsulates the five steps involved in data processing:

Download / tokenize / process.
Clean and (maybe) save to disk.
Load inside Dataset.
Apply transforms (rotate, tokenize, etc…).
Wrap inside a DataLoader.

I will be using Google Flowers dataset. The task is about given a sentence it has to be classified into one of the five classes.

roses, dandelion, tulips, sunflowers, daisy

following code is abstract_dataset.py which is abstract class of our DataModule.

class Dataset(abc.ABC):
    def __init__(self, input_params, with_labels=False,):
        self.batch_size = input_params.batch_size
        self.buffer_size = input_params.buffer_size
        self.img_height = input_params.img_height
        self.img_width = input_params.img_width
        self.data_dir = ''
        self.prepare_data()
 
        self.train_dataset = self.train_dataloader()
        self.val_dataset = self.val_dataloader()
 
    @abstractmethod
    def prepare_data(self):
        raise NotImplementedError
 
    @abstractmethod
    def train_dataloader(self):
        raise NotImplementedError
    @abstractmethod
    def val_dataloader(self):
        raise NotImplementedError
 
    def __iter__(self):
        return iter(self.train_dataset)

The Flower Photos code for the project looks like:

class FlowerPhotosDataset(abstract_dataset.Dataset):
    def __init__(
            self,
            input_params,
            with_labels=False,
    ):
        super().__init__(input_params, with_labels)
 
    def __call__(self, *args, **kwargs):
        return self.train_dataset
 
    def prepare_data(self):
        import os
        path=os.getcwd() + '/datasets/'
        self.data_dir = path+r'flower_photos'
 
        if(pathlib.Path(self.data_dir+'.tgz').is_file == False):
            import wget
            url='https://storage.googleapis.com/download.tensorflow.org/example_images/flower_photos.tgz'
            wget.download(url)
 
        if(pathlib.Path(self.data_dir).is_dir() == False):
            data_utils.extract(path, extract_path='./datasets')
 
        self.data_dir = pathlib.Path(path)
        image_count = len(list(self.data_dir.glob('*/*.jpg')))
 
    def train_dataloader(self):
        train_ds = tf.keras.preprocessing.image_dataset_from_directory(
            self.data_dir,
            validation_split=0.2,
            subset="training",
            seed=123,
            image_size=(self.img_height, self.img_width),
            batch_size=self.batch_size
        )
        return train_ds
 
    def val_dataloader(self):
        val_ds = tf.keras.preprocessing.image_dataset_from_directory(
            self.data_dir,
            validation_split=0.2,
            subset="validation",
            seed=123,
            image_size=(self.img_height, self.img_width),
            batch_size=self.batch_size
        )
        return val_ds

Building a Model

class Model(ABC):
 
    def __init__(
            self,
            model_parameters: edict = None,
    ):
        self._model_parameters = model_parameters
        self._model = self.define_model()
 
    def __call__(self, inputs, **kwargs):
        return self.model(inputs=inputs, **kwargs)
 
    @abstractmethod
    def define_model(self) -> keras.Model:
        raise NotImplementedError
 
    @property
    def model(self):
        return self._model
 
    @property
    def model_parameters(self) -> edict:
        return self._model_parameters
 
    @property
    def trainable_variables(self):
        return self.model.trainable_variables
 
    @property
    def model_name(self) -> str:
        return self.__class__.__name__
 
    def __repr__(self):
        return self.model_name

class FlowerClassifierModel(model.Model):
  def __init__(self, model_parameters: edict):
    super().__init__(model_parameters)
 
 
  def define_model(self):
    model = Sequential([
      layers.Rescaling(1./255, input_shape=(self.model_parameters.img_height, self.model_parameters.img_width, 3)),
      layers.Conv2D(16, 3, padding='same', activation='relu'),
      layers.MaxPooling2D(),
      layers.Conv2D(32, 3, padding='same', activation='relu'),
      layers.MaxPooling2D(),
      layers.Conv2D(64, 3, padding='same', activation='relu'),
      layers.MaxPooling2D(),
      layers.Flatten(),
      layers.Dense(128, activation='relu'),
      layers.Dense(self.model_parameters.num_channels)
    ])
 
    return model

Training

We setup Trainer and can customize several options, such as logging, gradient accumulation, half precision training, distributed computing, etc.

We'll stick to the basics for this example.

following code in run.py

trainer = Trainer(
  model_parameters = problem_params,
  model=model,
  optimizer='adam',
  loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
  callbacks=[checkpoint_callback, early_stopping_callback, tensorboard_callback]
)
 
trainer.compile()
trainer.train(dataset=dataset)

this is structure of trainer.py

class Trainer:
    def __init__(
            self,
            model_parameters,
 
            model: model.Model,
            optimizer: optimizers,
            loss: loss,
            callbacks: List[callback.Callback] = None,
    ):
        self.model_parameters = model_parameters
        self.batch_size = model_parameters.batch_size
        self.model = model
 
        self.global_step = 0
        self.epoch = 0
 
        self.optimizer = optimizer
        self.loss = loss
        self.callbacks = callbacks
 
    def compile(self,):
        self.model.model.compile(
            optimizer=self.optimizer,
            loss=self.loss,
            metrics=['accuracy']
        )
 
        self.model.model.summary()
 
    def train(self, dataset: abstract_dataset.Dataset):
 
        model = self.model.model
        history = model.fit(
            dataset.train_dataset,
            validation_data=dataset.val_dataset,
            epochs=self.model_parameters.num_epochs,
            callbacks=self.callbacks
        )

Callback

Callback is a self-contained program that can be reused across projects.

As an example, I will be implementing ModelCheckpoint callback. This will save the trained model. We can selectively choose which model to save by monitoring a metric.(val_loss in this case). The best model will be saved in the dirpath.

Refer to the documentation to learn more about callbacks.

checkpoint_filepath = '/tmp/checkpoint'
checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(
    filepath=checkpoint_filepath,
    save_weights_only=True,
    monitor='val_accuracy',
    mode='max',
    save_best_only=True)
 
trainer = Trainer(
    model_parameters = problem_params,
    model=model,
    optimizer='adam',
    loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    callbacks=[checkpoint_callback]
)

We can also chain multiple callbacks. EarlyStopping callback helps the model not to overfit by mointoring on a certain parameter (val_loss in this case).

checkpoint_filepath = '/tmp/checkpoint'
 
checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(
    filepath=checkpoint_filepath,
    save_weights_only=True,
    monitor='val_accuracy',
    mode='max',
    save_best_only=True)
 
early_stopping_callback = tf.keras.callbacks.EarlyStopping(
  monitor='val_loss', min_delta=0, patience=0, verbose=0,
  mode='auto', baseline=None, restore_best_weights=False
)
 
trainer = Trainer(
    model_parameters = problem_params,
    model=model,
    optimizer='adam',
    loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    callbacks=[checkpoint_callback, early_stopping_callback]
)

Logging

Logging of the model training is as simple as

checkpoint_filepath = '/tmp/checkpoint'
 
checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(
    filepath=checkpoint_filepath,
    save_weights_only=True,
    monitor='val_accuracy',
    mode='max',
    save_best_only=True)
 
early_stopping_callback = tf.keras.callbacks.EarlyStopping(
  monitor='val_loss', min_delta=0, patience=0, verbose=0,
  mode='auto', baseline=None, restore_best_weights=False
)
 
tensorboard_callback = tf.keras.callbacks.TensorBoard(
    log_dir='logs', histogram_freq=0, write_graph=True,
    write_images=False, write_steps_per_second=False, update_freq='epoch',
    profile_batch=0, embeddings_freq=0, embeddings_metadata=None
  )
 
trainer = Trainer(
  model_parameters = problem_params,
  model=model,
  optimizer='adam',
  loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
  callbacks=[checkpoint_callback, early_stopping_callback, tensorboard_callback]
)

It will create a directory called logs if not present. You can visualize the tensorboard logs using the following command

tensorboard --logdir=path_to_your_logs

You can see the tensorboard at http://localhost:6006/

Inference

Once the model is trained, we can use the trained model to get predictions on the run time data. Typically Inference contains:

Load the trained model
Get the run time (inference) input
Convert the input in the required format
Get the predictions

import tensorflow as tf
 
from datasets.flower_photos import FlowerPhotosDataset as Dataset
from models.flower_classifier import FlowerClassifierModel as Model
from utils import config
 
class Predictor:
    def __init__(self, model_path):
        self.model_path = model_path
        self.model = tf.keras.models.load_model(self.model_path)
 
    def predict(self, img):
        img_array = tf.keras.preprocessing.image.img_to_array(img)
        img_array = tf.expand_dims(img_array, 0)
        predictions = self.model.predict(img_array)
        score = tf.nn.softmax(predictions[0])
        return predictions

This conculdes the post. In the next post, I will be going through:

How to monitor model performance with Weights and Bias?

Complete code for this post can also be found here: