Drift Detection Tutorial Using Multiple Drift Detectors

Problem Statement

When evaluating and monitoring data after model deployment, it is important to test incoming data for potential drift which may affect model performance.

When to use

The dataeval.metrics.drift detector classes should be used when you would like to measure new data for operational drift.

What you will need

1. A set of image embeddings for each dataset (usually obtained with an AutoEncoder)

2. A python environment with the following packages installed:

   • dataeval[torch] or dataeval[all]

   • tensorflow-datasets

Setting up

Let’s import the required libraries needed to set up a minimal working example

from functools import partial

import numpy as np
import tensorflow_datasets as tfds
import torch

from dataeval.detectors import (
    DriftCVM,
    DriftKS,
    DriftMMD,
    preprocess_drift,
)
from dataeval.models.torch import AriaAutoencoder

device = "cuda" if torch.cuda.is_available() else "cpu"

Loading in data

Let’s start by loading in tensorflow’s MNIST dataset, then we will examine it.

# Load in the mnist dataset from tensorflow datasets
dataset, ds_info = tfds.load(
    "mnist",
    split="train[:4000]",
    with_info=True,
)
tfds.visualization.show_examples(dataset, ds_info)
dataset = dataset.shuffle(dataset.cardinality())
dataset = [i["image"] for i in dataset]
dataset = np.array(dataset, dtype=np.float32).transpose(0, 3, 1, 2)

print("Number of samples: ", len(dataset))
print("Image shape:", dataset[0].shape)

Number of samples:  4000
Image shape: (1, 28, 28)

Test reference against control

Let’s check for drift between the first 2000 images and the second 2000 images from this sample.

data_reference = dataset[0:2000]
data_control = dataset[2000:]

In order to reduce the dimensionality of the data, we can set a simple Autoencoder to the preprocess_fn. While this is optional for the MNIST data set, it is highly recommended for datasets that have higher dimensionality.

For the purposes of the tutorial, we will use 3 forms of drift detectors: Maximum Mean Discrepancy (MMD), Cramér-von Mises (CVM), and Kolmogorov-Smirnov (KS).

# define encoder
encoder_net = AriaAutoencoder(1).encoder.to(device)

# define preprocessing function
preprocess_fn = partial(preprocess_drift, model=encoder_net, batch_size=64, device=device)

# initialise drift detectors
detectors = [detector(data_reference, preprocess_fn=preprocess_fn) for detector in [DriftMMD, DriftCVM, DriftKS]]

We estimate that the test for drift is false for all detectors as both the reference and test data set is from the same MNIST training dataset.

[(type(detector).__name__, detector.predict(data_control)["is_drift"]) for detector in detectors]

[('DriftMMD', 0), ('DriftCVM', 0), ('DriftKS', 0)]

Loading in corrupted data

Now let’s load in a corrupted MNIST dataset.

corrupted, ds_info = tfds.load(
    "mnist_corrupted/translate",
    split="train[:2000]",
    with_info=True,
)
tfds.visualization.show_examples(corrupted, ds_info)
corrupted = corrupted.shuffle(corrupted.cardinality())
corrupted = [i["image"] for i in corrupted]
corrupted = np.array(corrupted, dtype=np.float32).transpose(0, 3, 1, 2)

print("Number of corrupted samples: ", len(corrupted))
print("Corrupted image shape:", corrupted[0].shape)

Number of corrupted samples:  2000
Corrupted image shape: (1, 28, 28)

Check for drift against corrupted data

Test for drift between the corrupted dataset and the original reference set using all 3 detectors.

[(type(detector).__name__, detector.predict(corrupted)["is_drift"]) for detector in detectors]

[('DriftMMD', 1), ('DriftCVM', 1), ('DriftKS', 1)]

We conclude that the translated MNIST images are significantly different from the original images according to all 3 measures of drift.