sm_model_monitor.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "3a33ab0a",
   "metadata": {},
   "source": [
    "# Bring your own custom drift detector with Amazon SageMaker Model Monitor\n",
    "\n",
    "This notebook shows how to:\n",
    "\n",
    "* Host a machine learning model in Amazon SageMaker and capture inference requests, results, and metadata\n",
    "* Build a Docker container to include your custom drift algorithms\n",
    "* Monitor a live endpoint for detecting drifts\n",
    "* Visualize the drift results\n",
    "\n",
    "## Background\n",
    "\n",
    "Amazon SageMaker provides every developer and data scientist with the ability to build, train, and deploy machine learning models quickly. Amazon SageMaker is a fully-managed service that encompasses the entire machine learning workflow. You can label and prepare your data, choose an algorithm, train a model, and then tune and optimize it for deployment. You can deploy your models to production with Amazon SageMaker to make predictions and lower costs than was previously possible.\n",
    "\n",
    "In addition, Amazon SageMaker enables you to capture the input, output and metadata for invocations of the models that you deploy. It also enables you to bring your own metrics to analyze the data and monitor its quality. In this notebook, you learn how Amazon SageMaker enables these capabilities.\n",
    "\n",
    "## Setup\n",
    "\n",
    "To get started, make sure you have these prerequisites completed.\n",
    "\n",
    "* Specify an AWS Region to host your model.\n",
    "* An IAM role ARN exists that is used to give Amazon SageMaker access to your data in Amazon Simple Storage Service (Amazon S3). See the documentation for how to fine tune the permissions needed.\n",
    "* Create an S3 bucket used to store the data used to train your model, any additional model data, and the data captured from model invocations. For demonstration purposes, you are using the same bucket for these. In reality, you might want to separate them with different security policies."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8019c79f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Handful of configuration\n",
    "\n",
    "import os\n",
    "import boto3\n",
    "import json\n",
    "from sagemaker import get_execution_role, session\n",
    "\n",
    "region= boto3.Session().region_name\n",
    "\n",
    "sm_client = boto3.client('sagemaker')\n",
    "\n",
    "role = get_execution_role()\n",
    "print(\"RoleArn: {}\".format(role))\n",
    "\n",
    "# You can use a different bucket, but make sure the role you chose for this notebook\n",
    "# has the s3:PutObject permissions. This is the bucket into which the data is captured\n",
    "bucket =  session.Session(boto3.Session()).default_bucket()\n",
    "print(\"Demo Bucket: {}\".format(bucket))\n",
    "prefix = 'sagemaker/DEMO-ModelMonitor'\n",
    "\n",
    "s3_capture_upload_path = f's3://{bucket}/{prefix}/datacapture'\n",
    "s3_report_path = f's3://{bucket}/{prefix}/reports'\n",
    "\n",
    "print(f\"Capture path: {s3_capture_upload_path}\")\n",
    "print(f\"Report path: {s3_report_path}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cd1fe234",
   "metadata": {},
   "source": [
    "### Upload train and test datasets, and model file to S3\n",
    "\n",
    "The dataset is taken from [UCI Census Income Data Set](https://archive.ics.uci.edu/ml/datasets/adult). The task is to predict whether income exceeds $50K/yr based on census data. We have split the dataset into train and test datasets. The model was trained using XGBoost and the model file is provided here. \n",
    "\n",
    "We need test datasets for calculating projected accuracy."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bfc8f1f9",
   "metadata": {},
   "outputs": [],
   "source": [
    "model_file = open(\"model/model.tar.gz\", 'rb')\n",
    "train_file = open(\"data/train.csv\", 'rb')\n",
    "test_file = open(\"data/test.csv\", 'rb')\n",
    "\n",
    "s3_model_key = os.path.join(prefix, 'model.tar.gz')\n",
    "s3_train_key = os.path.join(prefix, 'train.csv')\n",
    "s3_test_key = os.path.join(prefix, 'test.csv')\n",
    "\n",
    "boto3.Session().resource('s3').Bucket(bucket).Object(s3_model_key).upload_fileobj(model_file)\n",
    "boto3.Session().resource('s3').Bucket(bucket).Object(s3_train_key).upload_fileobj(train_file)\n",
    "boto3.Session().resource('s3').Bucket(bucket).Object(s3_test_key).upload_fileobj(test_file)\n",
    "\n",
    "print(\"Success! You are all set to proceed.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e77339ef",
   "metadata": {},
   "source": [
    "## Bring your own custom model drift detection algorithm\n",
    "\n",
    "In order to bring your own custom model drift detection algorithm, you need to do following things:\n",
    "* Create custom detection algorithms. We have included algorithms under src folder\n",
    "* Create a Docker container.\n",
    "* Set enviornmental variables where the container can find the datacapture data from SageMaker Model Monitor. These variables have to match with the values we provide to monitor scheduler later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0d8f13b9",
   "metadata": {},
   "outputs": [],
   "source": [
    "!pygmentize Dockerfile"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eef5edbf",
   "metadata": {},
   "source": [
    "### Build the container and upload it to ECR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7070dbf2",
   "metadata": {},
   "outputs": [],
   "source": [
    "from docker_utils import build_and_push_docker_image\n",
    "\n",
    "repository_short_name = 'custom-model-monitor'\n",
    "\n",
    "image_name = build_and_push_docker_image(repository_short_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1f5f0cfd",
   "metadata": {},
   "source": [
    "## Setup endoint and enable data capture"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3dd1564d",
   "metadata": {},
   "source": [
    "The data that is sent for inference to the endpoint needs to pre-processed before the XGBoost model can do prediction. Below code shows custom input handler for inference endpoint."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0ab64911",
   "metadata": {},
   "outputs": [],
   "source": [
    "!pygmentize script/inference.py"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f69c10ee",
   "metadata": {},
   "source": [
    "### Setting up model endpoint can take few minutes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "64ee65aa",
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.xgboost.model import XGBoostModel\n",
    "from sagemaker.serializers import CSVSerializer\n",
    "from sagemaker.model_monitor import DataCaptureConfig\n",
    "\n",
    "model_url = f's3://{bucket}/{s3_model_key}'\n",
    "\n",
    "xgb_inference_model = XGBoostModel(\n",
    "    model_data=model_url,\n",
    "    role=role,\n",
    "    entry_point='inference.py',\n",
    "    source_dir='script',\n",
    "    framework_version='1.2-1',\n",
    ")\n",
    "\n",
    "data_capture_config = DataCaptureConfig(\n",
    "                        enable_capture=True,\n",
    "                        sampling_percentage=100,\n",
    "                        destination_s3_uri=s3_capture_upload_path)\n",
    "\n",
    "predictor = xgb_inference_model.deploy(\n",
    "    initial_instance_count=1,\n",
    "    instance_type=\"ml.c5.xlarge\",\n",
    "    serializer=CSVSerializer(),\n",
    "    data_capture_config=data_capture_config)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "58af9dbd",
   "metadata": {},
   "source": [
    "## Create monitoring schedule to detect drifts on hourly basis\n",
    "\n",
    "Default Model monitor can be setup to monitor the inference on an hourly basis against the baseline metrics and violations. In this example, we are setting custom model monitor. For this purpose, we are using Boto3 calls directly to setup model monitor with the container we built above. Note that we need to setup input and output paths on the container. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "33c16058",
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_train_path = f's3://{bucket}/{s3_train_key}'\n",
    "s3_test_path = f's3://{bucket}/{s3_test_key}'\n",
    "s3_result_path = f's3://{bucket}/{prefix}/result/{predictor.endpoint_name}'\n",
    "\n",
    "sm_client.create_monitoring_schedule(\n",
    "    MonitoringScheduleName=predictor.endpoint_name,\n",
    "    MonitoringScheduleConfig={\n",
    "        'ScheduleConfig': {\n",
    "            'ScheduleExpression': 'cron(0 * ? * * *)'\n",
    "        },\n",
    "        'MonitoringJobDefinition': {\n",
    "            'MonitoringInputs': [\n",
    "                {\n",
    "                    'EndpointInput': {\n",
    "                        'EndpointName': predictor.endpoint_name,\n",
    "                        'LocalPath': '/opt/ml/processing/endpointdata'\n",
    "                    }\n",
    "                },\n",
    "            ],\n",
    "            'MonitoringOutputConfig': {\n",
    "                'MonitoringOutputs': [\n",
    "                    {\n",
    "                        'S3Output': {\n",
    "                            'S3Uri': s3_result_path,\n",
    "                            'LocalPath': '/opt/ml/processing/resultdata',\n",
    "                            'S3UploadMode': 'EndOfJob'\n",
    "                        }\n",
    "                    },\n",
    "                ]\n",
    "            },\n",
    "            'MonitoringResources': {\n",
    "                'ClusterConfig': {\n",
    "                    'InstanceCount': 1,\n",
    "                    'InstanceType': 'ml.c5.xlarge',\n",
    "                    'VolumeSizeInGB': 10\n",
    "                }\n",
    "            },\n",
    "            'MonitoringAppSpecification': {\n",
    "                'ImageUri': image_name,\n",
    "                'ContainerArguments': [\n",
    "                    '--train_s3_uri',\n",
    "                    s3_train_path,\n",
    "                    '--test_s3_uri',\n",
    "                    s3_test_path,\n",
    "                    '--target_label',\n",
    "                    'income'\n",
    "                ]\n",
    "            },\n",
    "            'StoppingCondition': {\n",
    "                'MaxRuntimeInSeconds': 600\n",
    "            },\n",
    "            'Environment': {\n",
    "                'string': 'string'\n",
    "            },\n",
    "            'RoleArn': role\n",
    "        }\n",
    "    }\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1c648756",
   "metadata": {},
   "source": [
    "## Start sending pre-configured traffic to endpoint\n",
    "\n",
    "The cell below starts a thread to send some pre-configured traffic at a constant rate to the endpoint. The data points have been pre-conditioned to have drift, so that we can visualize it later. The traffic is sent for about 10 hours. If you like to stop the traffic, you need to stop the kernel to terminate this thread. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c5783679",
   "metadata": {},
   "outputs": [],
   "source": [
    "from threading import Thread\n",
    "from time import time, sleep\n",
    "\n",
    "def invoke_endpoint(ep_name, file_name, runtime_client):\n",
    "    pre_time = time()\n",
    "    with open(file_name) as f:\n",
    "        count = len(f.read().split('\\n')) - 2 # Remove EOF and header\n",
    "    \n",
    "    # Calculate time needed to sleep between inference calls if we need to have a constant rate of calls for 10 hours\n",
    "    ten_hours_in_sec = 10*60*60\n",
    "    sleep_time = ten_hours_in_sec/count\n",
    "    \n",
    "    with open(file_name, 'r') as f:\n",
    "        next(f) # Skip header\n",
    "        \n",
    "        for ind, row in enumerate(f):   \n",
    "            start_time = time()\n",
    "            payload = row.rstrip('\\n')\n",
    "            response = runtime_client(data=payload)\n",
    "            \n",
    "            # Print every 15 minutes (900 seconds)\n",
    "            if (ind+1) % int(count/ten_hours_in_sec*900) == 0:\n",
    "                print(f'Finished sending {ind+1} records.')\n",
    "            \n",
    "            # Sleep to ensure constant rate. Time spent for inference is subtracted\n",
    "            sleep(max(sleep_time - (time() - start_time), 0))\n",
    "                \n",
    "    print(\"Done!\")\n",
    "    \n",
    "print(f\"Sending test traffic to the endpoint {predictor.endpoint_name}. \\nPlease wait...\")\n",
    "\n",
    "thread = Thread(target = invoke_endpoint, args=(predictor.endpoint, 'data/infer.csv', predictor.predict))\n",
    "thread.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "08acd79e",
   "metadata": {},
   "source": [
    "## Visualization\n",
    "\n",
    "Here we provide several visualizations to capture model drift. The plots are launched in threads so that they can refesh plots automatically every hour."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c1da5bec",
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "\n",
    "%autoreload 1\n",
    "\n",
    "import sys\n",
    "from threading import Timer\n",
    "\n",
    "sys.path.append('src')\n",
    "\n",
    "%aimport drift_visualizer\n",
    "%aimport utils"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2457afc5",
   "metadata": {},
   "source": [
    "### Projected accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "769bb844",
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_accuracy():\n",
    "    df = utils.construct_df_from_result(s3_result_path)\n",
    "    if df is not None:    \n",
    "        drift_visualizer.plot_accuracy(df)\n",
    "    Timer(3600, plot_accuracy)\n",
    "    \n",
    "plot_accuracy()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5a59129f",
   "metadata": {},
   "source": [
    "### Normalized feature drift scores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a3189ec7",
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_drift_score():\n",
    "    df = utils.construct_df_from_result(s3_result_path) \n",
    "    if df is not None:    \n",
    "        drift_visualizer.plot_drift_score(df)\n",
    "    Timer(3600, plot_drift_score)\n",
    "    \n",
    "plot_drift_score()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4529f7c7",
   "metadata": {},
   "source": [
    "### Null hypothesis of features (based on p-values)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fa0c93f1",
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_p_values():\n",
    "    df = utils.construct_df_from_result(s3_result_path)   \n",
    "    if df is not None:            \n",
    "        drift_visualizer.plot_p_values(df)\n",
    "    Timer(3600, plot_p_values)\n",
    "    \n",
    "plot_p_values()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ae9b8a43",
   "metadata": {},
   "source": [
    "## Clean up resources\n",
    "* Monitor schedule - needs to deleted before deleting endpoint\n",
    "* Delete endpoint\n",
    "* Delete model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c0f73372",
   "metadata": {},
   "outputs": [],
   "source": [
    "sm_client.delete_monitoring_schedule(MonitoringScheduleName=predictor.endpoint)\n",
    "\n",
    "# predictor.delete_endpoint()\n",
    "# predictor.delete_model()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}