deployment/sagemaker-inference-immersion-day-studio-ver-attendee.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# AWS SageMaker Inference Immersion Day\n",
    "This notebook shows how to:\n",
    "* __Lab 1:__ Deploy a real time endpoint with a prebuilt container and invoke it.\n",
    "* __Lab 2:__ Deploy a real time endpoint with a custom container.\n",
    "* __Lab 3:__ Host an endpoint with multiple production variants with different traffic.\n",
    "* __Lab 4:__ Autoscaling your endpoint.\n",
    "* __Lab 5:__ Host a machine learning model in Amazon SageMaker and capture inference requests, results, and metadata.\n",
    "* Schedule Clarify bias monitor to monitor predictions for bias drift on a regular basis.\n",
    "* Schedule Clarify explainability monitor to monitor predictions for feature attribution drift on a regular basis."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## General Setup (~20 - 30 min)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Account Configuration with Event Engine"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Please follow the instructions to configure the AWS account that you will be using for this workshop. Go to https://dashboard.eventengine.run/ and paste the `Event hash` provided."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### SageMaker Studio Configuration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a SageMaker Studio User clicking on `Add user`. Once it's been created, click on the user name and copy the IAM role associated with this one."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![config](../assets/sagemaker-studio-2.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![config](../assets/sagemaker-studio-3.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Go to IAM and search for the role and click on it."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![config](../assets/sagemaker-studio-4.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Click on `Trust Relationships` and in the Edit Trust Relationship text area, paste the following JSON, then click on `Update Trust Policy`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![config](../assets/sagemaker-studio-5.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```json\n",
    "{\n",
    "    \"Version\": \"2012-10-17\",\n",
    "    \"Statement\": [\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Principal\": {\n",
    "              \"Service\": \"sagemaker.amazonaws.com\"\n",
    "            },\n",
    "            \"Action\": \"sts:AssumeRole\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Principal\": {\n",
    "              \"Service\": \"codebuild.amazonaws.com\"\n",
    "            },\n",
    "            \"Action\": \"sts:AssumeRole\"\n",
    "         }\n",
    "    ]\n",
    "}\n",
    "```\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `Trust Relationships` tab should show 2 Trusted entities now."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![config](../assets/sagemaker-studio-6.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To be able to run docker image building from Studio notebook, you need to add an inline policy to this role as follows. Frist click on `Permissions` tab, then click on `Add inline policy`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![config](../assets/sagemaker-studio-7.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the `Create policy` page, click on `JSON` tab then copy the following text and paste it in the text area replacing the default text, then click on `Review Policy` follow the remaining steps to create the inline policy."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![config](../assets/sagemaker-studio-8.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```bash\n",
    "{\n",
    "    \"Version\": \"2012-10-17\",\n",
    "    \"Statement\": [\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"codebuild:DeleteProject\",\n",
    "                \"codebuild:CreateProject\",\n",
    "                \"codebuild:BatchGetBuilds\",\n",
    "                \"codebuild:StartBuild\"\n",
    "            ],\n",
    "            \"Resource\": \"arn:aws:codebuild:*:*:project/sagemaker-studio*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": \"logs:CreateLogStream\",\n",
    "            \"Resource\": \"arn:aws:logs:*:*:log-group:/aws/codebuild/sagemaker-studio*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"logs:GetLogEvents\",\n",
    "                \"logs:PutLogEvents\"\n",
    "            ],\n",
    "            \"Resource\": \"arn:aws:logs:*:*:log-group:/aws/codebuild/sagemaker-studio*:log-stream:*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": \"logs:CreateLogGroup\",\n",
    "            \"Resource\": \"*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"ecr:CreateRepository\",\n",
    "                \"ecr:BatchGetImage\",\n",
    "                \"ecr:CompleteLayerUpload\",\n",
    "                \"ecr:DescribeImages\",\n",
    "                \"ecr:DescribeRepositories\",\n",
    "                \"ecr:UploadLayerPart\",\n",
    "                \"ecr:ListImages\",\n",
    "                \"ecr:InitiateLayerUpload\",\n",
    "                \"ecr:BatchCheckLayerAvailability\",\n",
    "                \"ecr:PutImage\"\n",
    "            ],\n",
    "            \"Resource\": \"arn:aws:ecr:*:*:repository/sagemaker-studio*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": \"ecr:GetAuthorizationToken\",\n",
    "            \"Resource\": \"*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"s3:GetObject\",\n",
    "                \"s3:DeleteObject\",\n",
    "                \"s3:PutObject\"\n",
    "            ],\n",
    "            \"Resource\": \"arn:aws:s3:::sagemaker-*/*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"s3:CreateBucket\"\n",
    "            ],\n",
    "            \"Resource\": \"arn:aws:s3:::sagemaker*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"iam:GetRole\",\n",
    "                \"iam:ListRoles\"\n",
    "            ],\n",
    "            \"Resource\": \"*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": \"iam:PassRole\",\n",
    "            \"Resource\": \"arn:aws:iam::*:role/*\",\n",
    "            \"Condition\": {\n",
    "                \"StringLikeIfExists\": {\n",
    "                    \"iam:PassedToService\": \"codebuild.amazonaws.com\"\n",
    "                }\n",
    "            }\n",
    "        }\n",
    "    ]\n",
    "}\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Imports\n",
    "\n",
    "Import APIs to be used by the notebook. For almost all of the tasks presented in this notebook, we'll be using [SageMaker Python SDK](https://sagemaker.readthedocs.io/en/stable/)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import copy\n",
    "import json\n",
    "import random\n",
    "import time\n",
    "import boto3\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "from datetime import datetime, timedelta\n",
    "from sagemaker.session import production_variant\n",
    "from sagemaker import get_execution_role, image_uris, Session\n",
    "from sagemaker.serializers import CSVSerializer\n",
    "from sagemaker.clarify import (\n",
    "    BiasConfig,\n",
    "    DataConfig,\n",
    "    ModelConfig,\n",
    "    ModelPredictedLabelConfig,\n",
    "    SHAPConfig,\n",
    ")\n",
    "from sagemaker.model import Model\n",
    "from sagemaker.model_monitor import (\n",
    "    BiasAnalysisConfig,\n",
    "    CronExpressionGenerator,\n",
    "    DataCaptureConfig,\n",
    "    EndpointInput,\n",
    "    ExplainabilityAnalysisConfig,\n",
    "    ModelBiasMonitor,\n",
    "    ModelExplainabilityMonitor,\n",
    ")\n",
    "from sagemaker.s3 import S3Downloader, S3Uploader"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Handful of configuration\n",
    "Here, we are configuring the execution role that we'll be using for deploying everything, the bucket where we'll save artifacts and data and some prefixes to save data in separate folders according to our needs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "role = get_execution_role()\n",
    "print(f\"RoleArn: {role}\")\n",
    "\n",
    "boto_session = boto3.session.Session()\n",
    "\n",
    "sagemaker_session = Session(boto_session)\n",
    "sagemaker_client = sagemaker_session.sagemaker_client\n",
    "sagemaker_runtime_client = sagemaker_session.sagemaker_runtime_client\n",
    "\n",
    "region = sagemaker_session.boto_region_name\n",
    "print(f\"AWS region: {region}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# A different bucket can be used, but make sure the role for this notebook has\n",
    "# the s3:PutObject permissions. This is the bucket into which the data is captured\n",
    "bucket = sagemaker_session.default_bucket()\n",
    "print(f\"Demo Bucket: {bucket}\")\n",
    "prefix = \"sagemaker/DEMO-ClarifyModelMonitor-20200901\"\n",
    "s3_key = f\"s3://{bucket}/{prefix}\"\n",
    "print(f\"S3 key: {s3_key}\")\n",
    "\n",
    "s3_capture_upload_path = f\"{s3_key}/datacapture\"\n",
    "ground_truth_upload_path = f\"{s3_key}/ground_truth_data/{datetime.now():%Y-%m-%d-%H-%M-%S}\"\n",
    "s3_report_path = f\"{s3_key}/reports\"\n",
    "\n",
    "print(f\"Capture path: {s3_capture_upload_path}\")\n",
    "print(f\"Ground truth path: {ground_truth_upload_path}\")\n",
    "print(f\"Report path: {s3_report_path}\")\n",
    "\n",
    "baseline_results_uri = f\"{s3_key}/baselining\"\n",
    "print(f\"Baseline results uri: {baseline_results_uri}\")\n",
    "\n",
    "endpoint_instance_count = 1\n",
    "endpoint_instance_type = \"ml.m5.large\"\n",
    "schedule_expression = CronExpressionGenerator.hourly()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Model files and data files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_file = \"model/xgb-churn-prediction-model.tar.gz\"\n",
    "test_file = \"test_data/test-file.txt\"\n",
    "test_dataset = \"test_data/test.csv\"\n",
    "validation_dataset = \"test_data/validation-dataset-with-header.csv\"\n",
    "dataset_type = \"text/csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open(validation_dataset) as f:\n",
    "    headers_line = f.readline().rstrip()\n",
    "all_headers = headers_line.split(\",\")\n",
    "label_header = all_headers[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To verify that the execution role for this notebook has the necessary permissions to proceed. Put a simple test object into the S3 bucket speciﬁed above. If this command fails, update the role to have `s3:PutObject` permission on the bucket and try again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Upload a test file\n",
    "S3Uploader.upload(test_file, f\"s3://{bucket}/test_upload\", sagemaker_session=sagemaker_session)\n",
    "print(\"Success! We are all set to proceed.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# LAB 1: Deploying a real-time endpoint on Amazon SageMaker (~20-25 min)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here, we trained a model beforehand for you, so we'll be using the resulting artifact here to save some time. If you want to know how this model was trained, please refer to the [training notebook](../training/xgboost_customer_churn.ipynb)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Upload the pre-trained model to Amazon S3\n",
    "As an example, this code uploads a pre-trained XGBoost model that is ready for to be deployed. This model was trained using the [code that you can find on training folder](../training/xgboost_customer_churn.ipynb) in SageMaker. In order to deploy an endpoint, we will need to first upload the model artifact (the serialized object) to S3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_url = S3Uploader.upload(local_path=model_file, desired_s3_uri=s3_key, sagemaker_session=sagemaker_session)\n",
    "print(f\"Model file has been uploaded to {model_url}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploy the model with Amazon SageMaker"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Start with deploying a pre-trained churn prediction model. Here, create the SageMaker `Model` object with the image and model data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "model_name = f\"DEMO-xgb-churn-pred-model-monitor-{datetime.utcnow():%Y-%m-%d-%H%M%S}\"\n",
    "print(\"Model name: \", model_name)\n",
    "endpoint_name = f\"DEMO-xgb-churn-model-monitor-{datetime.utcnow():%Y-%m-%d-%H%M%S}\"\n",
    "print(\"Endpoint name: \", endpoint_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once you have a trained model, you can include it in a Docker container that runs your inference code. A container provides an effectively isolated environment, ensuring a consistent runtime regardless of where the container is deployed. Containerizing your model and code enables fast and reliable deployment of your model.\n",
    "\n",
    "We are going to use an already pre-built docker image with `xgboost 0.90-1` version by SageMaker. In order to do this, we will retrieve the ECR docker image URL from Amazon ECR."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "image_uri = image_uris.retrieve(\"xgboost\", region, \"0.90-1\")\n",
    "print(f\"XGBoost image uri: {image_uri}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Amazon SageMaker already has multiple pre-built docker images for you to use or extend. For more info on this please refer to these links:\n",
    "\n",
    "- [Deep Learning Docker Images](https://docs.aws.amazon.com/sagemaker/latest/dg/pre-built-containers-frameworks-deep-learning.html)\n",
    "- [Sklearn and Spark ML](https://docs.aws.amazon.com/sagemaker/latest/dg/pre-built-docker-containers-scikit-learn-spark.html)\n",
    "\n",
    "Also, Sagemaker Python SDK has already a high-level interface called `Estimator` to handle end-to-end training and deployment of most common ML and Deep Learning frameworks that you can find out there. Also, if __you already have a model that you trained somewhere else__, you can use `Model` interface to deploy a model as an endpoint in SageMaker. Refer to this links if you want to go deeper on your framework of interest.\n",
    "\n",
    "- [Scikit-learn](https://sagemaker.readthedocs.io/en/stable/frameworks/sklearn/index.html)\n",
    "- [SparkML](https://sagemaker.readthedocs.io/en/stable/frameworks/sparkml/index.html)\n",
    "- [Tensorflow](https://sagemaker.readthedocs.io/en/stable/frameworks/tensorflow/index.html)\n",
    "- [PyTorch](https://sagemaker.readthedocs.io/en/stable/frameworks/pytorch/index.html)\n",
    "- [HuggingFace](https://sagemaker.readthedocs.io/en/stable/frameworks/huggingface/index.html)\n",
    "- [MXNet](https://sagemaker.readthedocs.io/en/stable/frameworks/mxnet/index.html)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To create the model, we need to pass the mentioned Docker ECR uri, an IAM role used to access the data on S3 and create the endpoint on SageMaker, the model url on S3. Also in order to deploy an endpoint, we will to configure the instance (count and type) and a serializer that will define how the data will be encoded."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = Model(\n",
    "    # CODE STARTS HERE\n",
    "\n",
    "    # CODE ENDS HERE\n",
    ")\n",
    "print(f\"Deploying model {model_name} to endpoint {endpoint_name}\")\n",
    "model.deploy(\n",
    "    # CODE STARTS HERE\n",
    "    \n",
    "    # CODE ENDS HERE\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Invoke the endpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here, we are using the `SageMaker runtime client` to send some data to our realtime endpoint."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Sending test traffic to the endpoint {endpoint_name}. \\nPlease wait\", end=\"\")\n",
    "test_dataset_size = 0  # record the number of rows in data we're sending for inference\n",
    "count = 0\n",
    "with open(test_dataset, \"r\") as f:\n",
    "    for row in f:\n",
    "        if test_dataset_size < 10:\n",
    "            payload = row.rstrip(\"\\n\")\n",
    "            response = sagemaker_runtime_client.invoke_endpoint(\n",
    "                # CODE STARTS HERE\n",
    "                # CODE ENDS HERE\n",
    "            )\n",
    "            prediction = response[\"Body\"].read()\n",
    "            print(prediction)\n",
    "            \n",
    "            time.sleep(0.5)\n",
    "        test_dataset_size += 1\n",
    "\n",
    "print()\n",
    "print(\"Done!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Predictions correspond here to the actual probability of a client to churn or not, which goes from 0 to 1."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# LAB 2: Deploy a model with a custom docker image (~15-20 min)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are some cases where you want to use a custom docker image environment to deploy your models. For those, you can actually build your own customized docker image to deploy your model with SageMaker. If you want to know how this model was trained, please refer to the [training notebook](../training/scikitlearn_churn_prediction.py.ipynb).\n",
    "\n",
    "This will cover:\n",
    "- Building and pushing a docker image to Amazon Elastic Container Registry (Amazon ECR).\n",
    "- Using that image for deploying the model with SageMaker.\n",
    "- Invoking the endpoint using the SageMaker runtime client."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The docker image that we will be using has the following files:\n",
    "\n",
    "- __Dockerfile__: Specification for building your docker image.\n",
    "- __nginx.conf__: Nginx configuration file.\n",
    "- __wsgi.py__: Wrapper for gunicorn.\n",
    "- __serve__: Entrypoint for sagemaker to start the gunicorn server and nginx proxy.\n",
    "- __sklearn_model.joblib__: Model artifact result of [training notebook](../training/scikitlearn_churn_prediction.py.ipynb).\n",
    "- __predictor.py__: Inference code, flask simple rest api with 2 endpoints, `/ping` and `/invocations`.\n",
    "- __requirements.txt__: Python requirements.\n",
    "- __build_and_push.sh__: Utility script for build and pushing your ECR image locally. This can used in replacement of `sm-docker`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This files can be found inside the [custom_container folder](custom_container/)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You might encounter another examples that use a most updated way for deploying your custom containers called __sagemaker inference toolkit__ which is the recommended. However, you can still use both without problems. For more info on this please refer to this [link](https://github.com/aws/sagemaker-inference-toolkit)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Build and push your docker image to ECR"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will be using a docker image for packaging our new model trained with the same dataset the first model was trained on but with a different algorithm, this case implemented with scikit-learn. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following will:\n",
    "- Create an ECR repository if does not exist.\n",
    "- Build a docker image and tag it accordingly\n",
    "- Push the docker image that has been built to the created ECR repo.\n",
    "\n",
    "In order to do this from SageMaker studio notebook we need to use `sm-docker`. If you want to replicate this running locally, please refer to [sagemaker-inference-immersion-day-local-ver.ipynb](local_notebook/sagemaker-inference-immersion-day-local-ver.ipynb) notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install sagemaker-studio-image-build"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!cd custom_container && sm-docker build . --repository sagemaker-studio-sklearn-custom:latest"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create the endpoint using the custom image"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Note__: Before executing next cells, assign the image uri/repository that you got from last cell in the next one to the `image_uri_custom` variable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "image_uri_custom = \"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_name_custom = f\"DEMO-sklearn-churn-predictor-{datetime.utcnow():%Y-%m-%d-%H%M%S}\"\n",
    "print(\"Model name: \", model_name_custom)\n",
    "endpoint_name_custom = f\"DEMO-sklearn-churn-predictor-{datetime.utcnow():%Y-%m-%d-%H%M%S}\"\n",
    "print(\"Endpoint name: \", endpoint_name_custom)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = Model(\n",
    "    role=role,\n",
    "    name=model_name_custom,\n",
    "    image_uri=image_uri_custom,\n",
    "    sagemaker_session=sagemaker_session,\n",
    ")\n",
    "print(f\"Deploying model {model_name_custom} to endpoint {endpoint_name_custom}\")\n",
    "model.deploy(\n",
    "    initial_instance_count=endpoint_instance_count,\n",
    "    instance_type=endpoint_instance_type,\n",
    "    endpoint_name=endpoint_name_custom\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Invoke the endpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we did before, for invoking the endpoint, we use the `sagemaker runtime client`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Sending test traffic to the endpoint {endpoint_name_custom}. \\nPlease wait\", end=\"\")\n",
    "test_dataset_size = 0  # record the number of rows in data we're sending for inference\n",
    "count = 0\n",
    "with open(test_dataset, \"r\") as f:\n",
    "    for row in f:\n",
    "        if test_dataset_size < 10:\n",
    "            payload = row.rstrip(\"\\n\")\n",
    "            response = sagemaker_runtime_client.invoke_endpoint(\n",
    "                EndpointName=endpoint_name_custom,\n",
    "                Body=payload[2:],\n",
    "                ContentType=dataset_type,\n",
    "            )\n",
    "            prediction = response[\"Body\"].read()\n",
    "            print(prediction)\n",
    "            \n",
    "            time.sleep(0.5)\n",
    "        test_dataset_size += 1\n",
    "\n",
    "print()\n",
    "print(\"Done!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# LAB 3: Production Variants and A/B Testing (~15-20 min)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Amazon SageMaker enables you to test multiple models or model versions behind the same endpoint using production variants. Each production variant identifies a machine learning (ML) model and the resources deployed for hosting the model. You can distribute endpoint invocation requests across multiple production variants by providing the traffic distribution for each variant, or you can invoke a specific variant directly for each request."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploy a real-time endpoint with 2 production variants"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For this case, we'll be using both models that were already configured in previous steps. Each one is created as a production variant of the endpoint:\n",
    "- XGBoost Model with 60% of the traffic.\n",
    "- ScikitLearn Model with 40% left."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "production_variants = [\n",
    "    production_variant(\n",
    "        # CODE STARTS HERE\n",
    "        \n",
    "        # CODE ENDS HERE\n",
    "    ),\n",
    "    production_variant(\n",
    "        # CODE STARTS HERE\n",
    "        \n",
    "        # CODE ENDS HERE\n",
    "    )\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "production_variant_endpoint = sagemaker_session.endpoint_from_production_variants(\n",
    "    name=f\"DEMO-production-variant-endpoint-{datetime.utcnow():%Y-%m-%d-%H%M%S}\",\n",
    "    production_variants=production_variants,\n",
    "    wait=True,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Invoke the endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Each endpoint variant will receive a proportion of the calls, defined by the weight. \n",
    "# A specific variant can be called by passing the 'TargetVariant' parameter\n",
    "\n",
    "def invoke_endpoint(payload, **kwargs):\n",
    "    response = sagemaker_runtime_client.invoke_endpoint(Body=payload, **kwargs)\n",
    "    prediction = response[\"Body\"].read()\n",
    "    variant = response['ResponseMetadata']['HTTPHeaders']['x-amzn-invoked-production-variant']\n",
    "    return prediction, variant\n",
    "\n",
    "print(f\"Sending test traffic to the endpoint {production_variant_endpoint}. \\nPlease wait\\n\", end=\"\")\n",
    "#params = {'EndpointName':production_variant_endpoint, 'ContentType':dataset_type, 'TargetVariant':\"sklearn-variant\",} # You can pass the endpoint variant\n",
    "params = {'EndpointName':production_variant_endpoint, 'ContentType':dataset_type,}\n",
    "with open(test_dataset, \"r\") as f:\n",
    "    for i, row in enumerate(f):\n",
    "        if i < 15:\n",
    "            response, variant = invoke_endpoint(row.rstrip(\"\\n\")[2:], **params)\n",
    "            print('Received prediction :' + str(response) + ' from variant ' + variant)\n",
    "            time.sleep(0.1)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## A/B Testing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In many cases, such as e-commerce applications, offline model evaluation isn’t sufficient, and you need to A/B test models in production before making the decision of updating models. With Amazon SageMaker, you can easily perform A/B testing on ML models by running multiple production variants on an endpoint. You can use production variants to test ML models that have been trained using different training datasets, algorithms, and ML frameworks; test how they perform on different instance types; or a combination of all of the above.\n",
    "\n",
    "In A/B testing, you test different variants of your models and compare how each variant performs relative to each other. You then choose the best-performing model to replace a previously-existing model new version delivers better performance than the previously-existing version."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Lets see how often each variant is called.\n",
    "\n",
    "cw = boto_session.client(\"cloudwatch\")\n",
    "\n",
    "def get_invocation_metrics_for_endpoint_variant(endpoint_name, variant_name, start_time, end_time):\n",
    "    metrics = cw.get_metric_statistics(\n",
    "        Namespace=\"AWS/SageMaker\",\n",
    "        MetricName=\"Invocations\",\n",
    "        StartTime=start_time,\n",
    "        EndTime=end_time,\n",
    "        Period=60,\n",
    "        Statistics=[\"Sum\"],\n",
    "        Dimensions=[\n",
    "            {\"Name\": \"EndpointName\", \"Value\": endpoint_name},\n",
    "            {\"Name\": \"VariantName\", \"Value\": variant_name},\n",
    "        ],\n",
    "    )\n",
    "    return (\n",
    "        pd.DataFrame(metrics[\"Datapoints\"])\n",
    "        .sort_values(\"Timestamp\")\n",
    "        .set_index(\"Timestamp\")\n",
    "        .drop(\"Unit\", axis=1)\n",
    "        .rename(columns={\"Sum\": variant_name})\n",
    "    )\n",
    "\n",
    "\n",
    "def plot_endpoint_metrics(start_time=None):\n",
    "    start_time = start_time or datetime.now() - timedelta(minutes=60)\n",
    "    end_time = datetime.now()\n",
    "    metrics_variant1 = get_invocation_metrics_for_endpoint_variant(\n",
    "        production_variant_endpoint, \"xgboost-variant\", start_time, end_time\n",
    "    )\n",
    "    metrics_variant2 = get_invocation_metrics_for_endpoint_variant(\n",
    "        production_variant_endpoint, \"sklearn-variant\", start_time, end_time\n",
    "    )\n",
    "    metrics_variants = metrics_variant1.join(metrics_variant2, how=\"outer\")\n",
    "    metrics_variants.plot()\n",
    "    return metrics_variants"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Waiting a few seconds for initial metric creation...\")\n",
    "time.sleep(30)\n",
    "m = plot_endpoint_metrics()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make a few calls to a specific variant and re-check the stats\n",
    "\n",
    "params = {'EndpointName':production_variant_endpoint, 'ContentType':dataset_type, 'TargetVariant':\"sklearn-variant\",} # You can pass the endpoint variant\n",
    "with open(test_dataset, \"r\") as f:\n",
    "    for i, row in enumerate(f):\n",
    "        if i < 15:\n",
    "            response, _ = invoke_endpoint(row.rstrip(\"\\n\")[2:], **params)\n",
    "            time.sleep(0.1)\n",
    "time.sleep(10)\n",
    "m = plot_endpoint_metrics()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Let's evaluate the performance of each endpoint by calling it with test data\n",
    "\n",
    "params_sklearn = {'EndpointName':production_variant_endpoint, 'ContentType':dataset_type, 'TargetVariant':\"sklearn-variant\",}\n",
    "params_xgboost = {'EndpointName':production_variant_endpoint, 'ContentType':dataset_type, 'TargetVariant':\"xgboost-variant\",}\n",
    "label = []\n",
    "predict_sklearn = []\n",
    "predict_xgboost = []\n",
    "with open(test_dataset, \"r\") as f:\n",
    "    for i, row in enumerate(f):\n",
    "        if i < 100:\n",
    "            label.append(int(row.rstrip(\"\\n\")[0]))\n",
    "            response, _ = invoke_endpoint(row.rstrip(\"\\n\")[2:], **params_sklearn)\n",
    "            predict_sklearn.append(round(eval(response)['pred']))\n",
    "            response, _ = invoke_endpoint(row.rstrip(\"\\n\")[2:], **params_xgboost)\n",
    "            predict_xgboost.append(round(eval(response)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will get then some metrics for each of the endpoint with predictions you got from both variants to evaluate both models."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "label = np.array(label)\n",
    "predict_sklearn = np.array(predict_sklearn)\n",
    "predict_xgboost = np.array(predict_xgboost)\n",
    "\n",
    "\n",
    "sklearn_accuracy = sum(predict_sklearn == label) / len(label)\n",
    "xgboost_accuracy = sum(predict_xgboost == label) / len(label)\n",
    "print('Accuracy -> sklearn: {}, xgboost: {}'.format(sklearn_accuracy, xgboost_accuracy))\n",
    "\n",
    "# Calculate precision\n",
    "sklearn_precision = round(sum(predict_sklearn[predict_sklearn == 1] == label[predict_sklearn == 1]) / len(predict_sklearn[predict_sklearn == 1]), 2)\n",
    "xgboost_precision = round(sum(predict_xgboost[predict_xgboost == 1] == label[predict_xgboost == 1]) / len(predict_xgboost[predict_xgboost == 1]), 2)\n",
    "print('Precision -> sklearn: {}, xgboost: {}'.format(sklearn_precision, xgboost_precision))\n",
    "\n",
    "# Calculate recall\n",
    "sklearn_recall = round(sum(predict_sklearn[predict_sklearn == 1] == label[predict_sklearn == 1]) / len(label[label == 1]), 2)\n",
    "xgboost_recall = round(sum(predict_xgboost[predict_xgboost == 1] == label[predict_xgboost == 1]) / len(label[label == 1]), 2)\n",
    "print('Recall -> sklearn: {}, xgboost: {}'.format(sklearn_recall, xgboost_precision))\n",
    "\n",
    "# Calculate F1 score\n",
    "sklearn_f1_score = round(2 * (sklearn_precision * sklearn_recall) / (sklearn_precision + sklearn_recall), 2)\n",
    "xgboost_f1_score = round(2 * (xgboost_precision * xgboost_recall) / (xgboost_precision + xgboost_recall), 2)\n",
    "print('F1 Score -> sklearn: {}, xgboost: {}'.format(sklearn_f1_score, xgboost_f1_score))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's update the traffic weights accordingly in our endpoint."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We see that the xgboost variant is performing better, so lets increase the weight given to this variant\n",
    "sagemaker_client.update_endpoint_weights_and_capacities(\n",
    "    EndpointName=production_variant_endpoint,\n",
    "    DesiredWeightsAndCapacities=[\n",
    "        {\"DesiredWeight\": 20, \"VariantName\": \"sklearn-variant\"},\n",
    "        {\"DesiredWeight\": 80, \"VariantName\": \"xgboost-variant\"},\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We can confirm that the change was succesfull by describing the enpoint\n",
    "{\n",
    "    variant[\"VariantName\"]: variant[\"CurrentWeight\"]\n",
    "    for variant in sagemaker_client.describe_endpoint(EndpointName=production_variant_endpoint)[\"ProductionVariants\"]\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# LAB 4: Autoscaling your endpoint (~ 15-20 min)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Amazon SageMaker supports automatic scaling (autoscaling) for your hosted models. Autoscaling dynamically adjusts the number of instances provisioned for a model in response to changes in your workload. When the workload increases, autoscaling brings more instances online. When the workload decreases, autoscaling removes unnecessary instances so that you don't pay for provisioned instances that you aren't using."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's configure autoscaling for the first endpoint from LAB 1."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "app_auto_scale_client = boto_session.client(\"application-autoscaling\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "resource_id = f\"endpoint/{endpoint_name}/variant/AllTraffic\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, you need to define your sagemaker endpoint as scalable target. A scalable target is a resource that Application Auto Scaling can scale out and scale in. Scalable targets are uniquely identified by the combination of resource ID, scalable dimension, and namespace. Resource ID for sagemaker is build as follows: `endpoint/<endpoint_name>/variant/<variant_name>`. By default, variant is called AllTrafic when you deploy just one production variant endpoint.\n",
    "\n",
    "When you register a new scalable target, you must specify values for minimum and maximum capacity. Current capacity will be adjusted within the specified range when scaling starts. Application Auto Scaling scaling policies __will not scale capacity to values that are outside of this range__. For our case, we set capacity between 1 and 3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "app_auto_scale_client.register_scalable_target(\n",
    "    # CODE STARTS HERE\n",
    "    \n",
    "    # CODE ENDS HERE\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that you have register your endpoint as scalable target, you need to define the scaling policy and then apply it. To specify the metrics and target values for a scaling policy, you configure a target-tracking scaling policy. You can use either a predefined metric or a custom metric.\n",
    "\n",
    "If you need to define a target-tracking scaling policy that meets your custom requirements, define a custom metric. You can define a custom metric based on any production variant metric that changes in proportion to scaling. Not all SageMaker metrics work for target tracking. The metric must be a valid utilization metric, and it must describe how busy an instance is. The value of the metric must increase or decrease in inverse proportion to the number of variant instances. That is, the value of the metric should decrease when the number of instances increases."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "app_auto_scale_client.put_scaling_policy(\n",
    "    # CODE STARTS HERE\n",
    "    \n",
    "    # CODE ENDS HERE\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### How to determine your `SageMakerVariantInvocationsPerInstance`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This variable is determined using the following formula:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`SageMakerVariantInvocationsPerInstance = (MAX_RPS * SAFETY_FACTOR) * 60`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Where `MAX_RPS` is the maximum RPS, and `SAFETY_FACTOR` is the safety factor that you chose to ensure that your clients don't exceed the maximum RPS. Multiply by 60 to convert from RPS to invocations-per-minute to match the per-minute CloudWatch metric that SageMaker uses to implement automatic scaling (you don't need to do this if you measured requests-per-minute instead of requests-per-second). `MAX_RPS` can be calculated using a loading testing tool.\n",
    "\n",
    "__Note__\n",
    "SageMaker recommends that you start testing with a `SAFETY_FACTOR` of 0.5. Test your automatic scaling configuration to ensure it operates in the way you expect with your model for both increasing and decreasing customer traffic on your endpoint."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Lab 5: Capturing real-time inference data from Amazon SageMaker endpoints (~20-25 min)\n",
    "Create an endpoint to showcase the data capture capability in action. (In next parts, model monitors will be created to process the data.)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Background\n",
    "\n",
    "Amazon SageMaker Model Monitor continuously monitors the quality of Amazon SageMaker machine learning models in production. It enables developers to set alerts for when there are deviations in the model quality. Early and pro-active detection of these deviations enables corrective actions, such as retraining models, auditing upstream systems, or fixing data quality issues without having to monitor models manually or build additional tooling. \n",
    "\n",
    "Amazon SageMaker Clarify bias monitoring helps data scientists and ML engineers monitor predictions for bias on a regular basis. One way bias can be introduced in deployed ML models is when the data used in training differs from the data used to generate predictions. This is especially pronounced if the data used for training changes over time (e.g. fluctuating mortgage rates), and the model prediction will not be accurate unless the model is retrained with updated data. For example, model for predicting home prices can be biased if the mortgage rates used to train the model differ from the most current real-world mortgage rate.\n",
    "\n",
    "Aamazon SageMaker Clarify explainability monitoring offers tools to provide global explanations of models and to explain the predictions of a deployed model producing inferences. Such model explanation tools can help ML modelers and developers and other internal stakeholders understand model characteristics as a whole prior to deployment and to debug predictions provided by the model once deployed. The current offering includes a scalable and efficient implementation of [SHAP](https://papers.nips.cc/paper/7062-a-unified-approach-to-interpreting-model-predictions), based on the concept of the [Shapley value](https://en.wikipedia.org/wiki/Shapley_value) from the field of cooperative game theory that assigns each feature an importance value for a particular prediction.\n",
    "\n",
    "As the model is monitored, customers can view exportable reports and graphs detailing bias and feature attributions in SageMaker Studio and configure alerts in Amazon CloudWatch to receive notifications if violations are detected."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_url = S3Uploader.upload(local_path=model_file, desired_s3_uri=s3_key, sagemaker_session=sagemaker_session)\n",
    "print(f\"Model file has been uploaded to {model_url}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## PART A: Deploying a real-time endpoint with SageMaker"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploy the model to Amazon SageMaker\n",
    "Start with deploying a pre-trained churn prediction model. Here, create the SageMaker `Model` object with the image and model data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_name = f\"DEMO-xgb-churn-pred-model-monitor-{datetime.utcnow():%Y-%m-%d-%H%M%S}\"\n",
    "print(\"Model name: \", model_name)\n",
    "endpoint_name = f\"DEMO-xgb-churn-model-monitor-{datetime.utcnow():%Y-%m-%d-%H%M%S}\"\n",
    "print(\"Endpoint name: \", endpoint_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To enable data capture for monitoring jobs, here specify the new capture option called `DataCaptureConfig`, it enables capturing the request payload and the response payload of the endpoint. The capture config applies to all variants. Go ahead with the deployment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "image_uri = image_uris.retrieve(\"xgboost\", region, \"0.90-1\")\n",
    "print(f\"XGBoost image uri: {image_uri}\")\n",
    "model = Model(\n",
    "    role=role,\n",
    "    name=model_name,\n",
    "    image_uri=image_uri,\n",
    "    model_data=model_url,\n",
    "    sagemaker_session=sagemaker_session,\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",
    "print(f\"Deploying model {model_name} to endpoint {endpoint_name}\")\n",
    "model.deploy(\n",
    "    initial_instance_count=endpoint_instance_count,\n",
    "    instance_type=endpoint_instance_type,\n",
    "    endpoint_name=endpoint_name,\n",
    "    data_capture_config=data_capture_config,\n",
    "    serializer=CSVSerializer()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Invoke the deployed model\n",
    "\n",
    "Now send data to this endpoint to get inferences in real time. Because data capture is enabled in the previous steps, the request and response payload, along with some additional metadata, is saved in the Amazon S3 location specified in the DataCaptureConfig."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.read_csv(validation_dataset)\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "print(f\"Sending test traffic to the endpoint {endpoint_name}. \\nPlease wait\", end=\"\")\n",
    "test_dataset_size = 0  # record the number of rows in data we're sending for inference\n",
    "count = 0\n",
    "with open(test_dataset, \"r\") as f:\n",
    "    for row in f:\n",
    "        if test_dataset_size < 120:\n",
    "            payload = row.rstrip(\"\\n\")\n",
    "            response = sagemaker_runtime_client.invoke_endpoint(\n",
    "                EndpointName=endpoint_name,\n",
    "                Body=payload[2:],\n",
    "                ContentType=dataset_type,\n",
    "            )\n",
    "            prediction = response[\"Body\"].read()\n",
    "            print(prediction)\n",
    "            print(\".\", end=\"\", flush=True)\n",
    "            \n",
    "            time.sleep(0.5)\n",
    "        test_dataset_size += 1\n",
    "\n",
    "print()\n",
    "print(\"Done!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## View captured data\n",
    "\n",
    "Now list the data capture files stored in Amazon S3. There should be different files from different time periods organized based on the hour in which the invocation occurred. The format of the Amazon S3 path is:\n",
    "\n",
    "`s3://{destination-bucket-prefix}/{endpoint-name}/{variant-name}/yyyy/mm/dd/hh/filename.jsonl`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Waiting 30 seconds for captures to show up\", end=\"\")\n",
    "for _ in range(30):\n",
    "    capture_files = sorted(S3Downloader.list(f\"{s3_capture_upload_path}/{endpoint_name}\", sagemaker_session=sagemaker_session))\n",
    "    if capture_files:\n",
    "        break\n",
    "    print(\".\", end=\"\", flush=True)\n",
    "    time.sleep(1)\n",
    "print()\n",
    "print(\"Found Capture Files:\")\n",
    "print(\"\\n \".join(capture_files[-5:]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, view the content of a single capture file. Take a quick peek at the first few lines in the captured file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "capture_file = S3Downloader.read_file(capture_files[-1], sagemaker_session=sagemaker_session).split(\"\\n\")[-10:-1]\n",
    "print(capture_file[-1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, the contents of a single line is present below in a formatted JSON file to observe a little better."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "print(json.dumps(json.loads(capture_file[-1]), indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Start generating some artificial traffic\n",
    "The cell below starts a thread to send some traffic to the endpoint. If there is no traffic, the monitoring jobs are marked as `Failed` since there is no data to process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "\n",
    "\n",
    "class WorkerThread(threading.Thread):\n",
    "    def __init__(self, do_run, *args, **kwargs):\n",
    "        super(WorkerThread, self).__init__(*args, **kwargs)\n",
    "        self.__do_run = do_run\n",
    "        self.__terminate_event = threading.Event()\n",
    "\n",
    "    def terminate(self):\n",
    "        self.__terminate_event.set()\n",
    "\n",
    "    def run(self):\n",
    "        while not self.__terminate_event.is_set():\n",
    "            self.__do_run(self.__terminate_event)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def invoke_endpoint(terminate_event):\n",
    "    with open(test_dataset, \"r\") as f:\n",
    "        i = 0\n",
    "        for row in f:\n",
    "            payload = row.rstrip(\"\\n\")\n",
    "            response = sagemaker_runtime_client.invoke_endpoint(\n",
    "                EndpointName=endpoint_name,\n",
    "                ContentType=\"text/csv\",\n",
    "                Body=payload[2:],\n",
    "                InferenceId=str(i),  # unique ID per row\n",
    "            )\n",
    "            i += 1\n",
    "            response[\"Body\"].read()\n",
    "            time.sleep(1)\n",
    "            if terminate_event.is_set():\n",
    "                break\n",
    "\n",
    "\n",
    "# Keep invoking the endpoint with test data\n",
    "invoke_endpoint_thread = WorkerThread(do_run=invoke_endpoint)\n",
    "invoke_endpoint_thread.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice the `inferenceId` attribute used above to invoke. If this is present, it will be used to join with ground truth data (otherwise `eventId` will be used):"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Start generating some fake ground truth\n",
    "\n",
    "Besides captures, model bias monitoring execution also requires ground truth data. In real use cases, ground truth data should be regularly collected and uploaded to designated S3 location. In this example notebook, below code snippet is used to generate fake ground truth data. The first-party merge container will combine captures and ground truth data, and the merged data will be passed to model bias monitoring job for analysis. Similar to captures, the model bias monitoring execution will fail if there's no data to merge."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def ground_truth_with_id(inference_id):\n",
    "    random.seed(inference_id)  # to get consistent results\n",
    "    rand = random.random()\n",
    "    # format required by the merge container\n",
    "    return {\n",
    "        \"groundTruthData\": {\n",
    "            \"data\": \"1\" if rand < 0.7 else \"0\",  # randomly generate positive labels 70% of the time\n",
    "            \"encoding\": \"CSV\",\n",
    "        },\n",
    "        \"eventMetadata\": {\n",
    "            \"eventId\": str(inference_id),\n",
    "        },\n",
    "        \"eventVersion\": \"0\",\n",
    "    }\n",
    "\n",
    "\n",
    "def upload_ground_truth(upload_time):\n",
    "    records = [ground_truth_with_id(i) for i in range(test_dataset_size)]\n",
    "    fake_records = [json.dumps(r) for r in records]\n",
    "    data_to_upload = \"\\n\".join(fake_records)\n",
    "    target_s3_uri = f\"{ground_truth_upload_path}/{upload_time:%Y/%m/%d/%H/%M%S}.jsonl\"\n",
    "    print(f\"Uploading {len(fake_records)} records to\", target_s3_uri)\n",
    "    S3Uploader.upload_string_as_file_body(data_to_upload, target_s3_uri, sagemaker_session=sagemaker_session)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate data for the last hour\n",
    "upload_ground_truth(datetime.utcnow() - timedelta(hours=1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate data once a hour\n",
    "def generate_fake_ground_truth(terminate_event):\n",
    "    upload_ground_truth(datetime.utcnow())\n",
    "    for _ in range(0, 60):\n",
    "        time.sleep(60)\n",
    "        if terminate_event.is_set():\n",
    "            break\n",
    "\n",
    "\n",
    "ground_truth_thread = WorkerThread(do_run=generate_fake_ground_truth)\n",
    "ground_truth_thread.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PART B: Model Bias Monitor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Model bias monitor can detect bias drift of Machine Learning models in a regular basis. Similar to the other monitoring types, the standard procedure of creating a model bias monitor is first baselining and then monitoring schedule."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_bias_monitor = ModelBiasMonitor(\n",
    "    role=role,\n",
    "    sagemaker_session=sagemaker_session,\n",
    "    max_runtime_in_seconds=1800,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "## Create a baselining job\n",
    "\n",
    "A baselining job runs predictions on training dataset and suggests constraints. `suggest_baseline()` method starts a `SageMakerClarifyProcessor` processing job using SageMaker Clarify container to generate the constraints.\n",
    "\n",
    "The step is not mandatory, but providing constraints file to the monitor can enable violations file generation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Configurations\n",
    "\n",
    "Information about the input data need to be provided to the processor."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`DataConfig` stores information about the dataset to be analyzed, for example the dataset file, its format (CSV or JSONLines), headers (if any) and label."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_bias_baselining_job_result_uri = f\"{baseline_results_uri}/model_bias\"\n",
    "model_bias_data_config = DataConfig(\n",
    "    s3_data_input_path=validation_dataset,\n",
    "    s3_output_path=model_bias_baselining_job_result_uri,\n",
    "    label=label_header,\n",
    "    headers=all_headers,\n",
    "    dataset_type=dataset_type,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`BiasConfig` is the configuration of the sensitive groups in the dataset. Typically, bias is measured by computing a metric and comparing it across groups. The group of interest is specified using the \"facet.\" For post-training bias, the possitive label should also be taken into account."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_bias_config = BiasConfig(\n",
    "    label_values_or_threshold=[1],\n",
    "    facet_name=\"Account Length\",\n",
    "    facet_values_or_threshold=[100],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`ModelPredictedLabelConfig` specifies how to extract a predicted label from the model output. This model returns probability that user will churn. Here choose an arbitrary 0.8 cutoff to consider that a customer will churn. For more complicated outputs, there are a few more options, like \"label\" is the index, name or JSONpath to locate predicted label in endpoint response payload."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_predicted_label_config = ModelPredictedLabelConfig(\n",
    "    probability_threshold=0.8,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`ModelConfig` is configuration related to model to be used for inferencing. In order to compute post-training bias metrics, the computation needs to get inferences for the model name provided. To accomplish this, the processing job will use the model to create an ephemeral endpoint (also known as \"shadow endpoint\"). The processing job will delete the shadow endpoint after the computations are completed. The configuration is also used by explainability monitor."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_config = ModelConfig(\n",
    "    model_name=model_name,\n",
    "    instance_count=endpoint_instance_count,\n",
    "    instance_type=endpoint_instance_type,\n",
    "    content_type=dataset_type,\n",
    "    accept_type=dataset_type,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Kick off baselining job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_bias_monitor.suggest_baseline(\n",
    "    data_config=model_bias_data_config,\n",
    "    model_config=model_config,\n",
    "    \n",
    "    bias_config=model_bias_config,\n",
    "    model_predicted_label_config=model_predicted_label_config,\n",
    ")\n",
    "print(f\"ModelBiasMonitor baselining job: {model_bias_monitor.latest_baselining_job_name}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Below cell waits until the baselining job is completed and then inspects the suggested constraints. This step can be skipped, because the monitor to be scheduled will automatically pick up baselining job name and wait for it before monitoring execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_bias_monitor.latest_baselining_job.wait(logs=False)\n",
    "model_bias_constraints = model_bias_monitor.suggested_constraints()\n",
    "print()\n",
    "print(f\"ModelBiasMonitor suggested constraints: {model_bias_constraints.file_s3_uri}\")\n",
    "print(S3Downloader.read_file(model_bias_constraints.file_s3_uri, sagemaker_session=sagemaker_session))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Schedule model bias monitor\n",
    "\n",
    "With above constraints collected, now call `create_monitoring_schedule()` method to schedule a hourly monitor, to analyze the data with monitoring schedule. If a baselining job has been submitted, then the monitor will automatically pick up analysis configuration from the baselining job. But if the baselining step is skipped, or the capture dataset has different nature than the training dataset, then analysis configuration has to be provided.\n",
    "\n",
    "`BiasAnalysisConfig` is a subset of the configuration of the baselining job, many options are not needed because,\n",
    "* Model bias monitor will merge captures and ground truth data and use merged data as dataset. (~~DataConfig~~)\n",
    "* Captures already include predictions, so there is no need to create shadow endpoint. (~~ModelConfig~~)\n",
    "* Attributes like probability threshold are provided as part of EndpointInput. (~~ModelPredictedLabelConfig~~)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_bias_analysis_config = None\n",
    "if not model_bias_monitor.latest_baselining_job:\n",
    "    model_bias_analysis_config = BiasAnalysisConfig(\n",
    "        model_bias_config,\n",
    "        headers=all_headers,\n",
    "        label=label_header,\n",
    "    )\n",
    "model_bias_monitor.create_monitoring_schedule(\n",
    "    analysis_config=model_bias_analysis_config,\n",
    "    output_s3_uri=s3_report_path,\n",
    "    endpoint_input=EndpointInput(\n",
    "        endpoint_name=endpoint_name,\n",
    "        destination=\"/opt/ml/processing/input/endpoint\",\n",
    "        start_time_offset=\"-PT1H\",\n",
    "        end_time_offset=\"-PT0H\",\n",
    "        probability_threshold_attribute=0.8,\n",
    "    ),\n",
    "    ground_truth_input=ground_truth_upload_path,\n",
    "    schedule_cron_expression=schedule_expression,\n",
    ")\n",
    "print(f\"Model bias monitoring schedule: {model_bias_monitor.monitoring_schedule_name}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Wait for the first execution\n",
    "\n",
    "The schedule starts jobs at the previously specified intervals. Code below wait util time crosses the hour boundary (in UTC) to see executions kick off.\n",
    "\n",
    "Note: Even for an hourly schedule, Amazon SageMaker has a buffer period of 20 minutes to schedule executions. The execution might start in anywhere from zero to ~20 minutes from the hour boundary. This is expected and done for load balancing in the backend."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def wait_for_execution_to_start(model_monitor):\n",
    "    print(\n",
    "        \"A hourly schedule was created above and it will kick off executions ON the hour (plus 0 - 20 min buffer).\"\n",
    "    )\n",
    "\n",
    "    print(\"Waiting for the first execution to happen\", end=\"\")\n",
    "    schedule_desc = model_monitor.describe_schedule()\n",
    "    while \"LastMonitoringExecutionSummary\" not in schedule_desc:\n",
    "        schedule_desc = model_monitor.describe_schedule()\n",
    "        print(\".\", end=\"\", flush=True)\n",
    "        time.sleep(60)\n",
    "    print()\n",
    "    print(\"Done! Execution has been created\")\n",
    "\n",
    "    print(\"Now waiting for execution to start\", end=\"\")\n",
    "    while schedule_desc[\"LastMonitoringExecutionSummary\"][\"MonitoringExecutionStatus\"] in \"Pending\":\n",
    "        schedule_desc = model_monitor.describe_schedule()\n",
    "        print(\".\", end=\"\", flush=True)\n",
    "        time.sleep(10)\n",
    "\n",
    "    print()\n",
    "    print(\"Done! Execution has started\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wait_for_execution_to_start(model_bias_monitor)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In real world, a monitoring schedule is supposed to be active all the time. But in this example, it can be stopped to avoid incurring extra charges. A stopped schedule will not trigger further executions, but the ongoing execution will continue. And if needed, the schedule can be restarted by `start_monitoring_schedule()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_bias_monitor.stop_monitoring_schedule()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Wait for the execution to finish\n",
    "\n",
    "In the previous cell, the first execution has started. This section waits for the execution to finish so that its analysis results are available. Here are the possible terminal states and what each of them mean:\n",
    "\n",
    "* Completed - This means the monitoring execution completed and no issues were found in the violations report.\n",
    "* CompletedWithViolations - This means the execution completed, but constraint violations were detected.\n",
    "* Failed - The monitoring execution failed, maybe due to client error (perhaps incorrect role permissions) or infrastructure issues. Further examination of FailureReason and ExitMessage is necessary to identify what exactly happened.\n",
    "* Stopped - job exceeded max runtime or was manually stopped."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Waits for the schedule to have last execution in a terminal status.\n",
    "def wait_for_execution_to_finish(model_monitor):\n",
    "    schedule_desc = model_monitor.describe_schedule()\n",
    "    execution_summary = schedule_desc.get(\"LastMonitoringExecutionSummary\")\n",
    "    if execution_summary is not None:\n",
    "        print(\"Waiting for execution to finish\", end=\"\")\n",
    "        while execution_summary[\"MonitoringExecutionStatus\"] not in [\n",
    "            \"Completed\",\n",
    "            \"CompletedWithViolations\",\n",
    "            \"Failed\",\n",
    "            \"Stopped\",\n",
    "        ]:\n",
    "            print(\".\", end=\"\", flush=True)\n",
    "            time.sleep(60)\n",
    "            schedule_desc = model_monitor.describe_schedule()\n",
    "            execution_summary = schedule_desc[\"LastMonitoringExecutionSummary\"]\n",
    "        print()\n",
    "        print(\"Done! Execution has finished\")\n",
    "    else:\n",
    "        print(\"Last execution not found\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wait_for_execution_to_finish(model_bias_monitor)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Inspect execution results\n",
    "\n",
    "List the generated reports,"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "schedule_desc = model_bias_monitor.describe_schedule()\n",
    "execution_summary = schedule_desc.get(\"LastMonitoringExecutionSummary\")\n",
    "if execution_summary and execution_summary[\"MonitoringExecutionStatus\"] in [\n",
    "    \"Completed\",\n",
    "    \"CompletedWithViolations\",\n",
    "]:\n",
    "    last_model_bias_monitor_execution = model_bias_monitor.list_executions()[-1]\n",
    "    last_model_bias_monitor_execution_report_uri = (\n",
    "        last_model_bias_monitor_execution.output.destination\n",
    "    )\n",
    "    print(f\"Report URI: {last_model_bias_monitor_execution_report_uri}\")\n",
    "    last_model_bias_monitor_execution_report_files = sorted(\n",
    "        S3Downloader.list(last_model_bias_monitor_execution_report_uri, sagemaker_session=sagemaker_session)\n",
    "    )\n",
    "    print(\"Found Report Files:\")\n",
    "    print(\"\\n \".join(last_model_bias_monitor_execution_report_files))\n",
    "else:\n",
    "    last_model_bias_monitor_execution = None\n",
    "    print(\n",
    "        \"====STOP==== \\n No completed executions to inspect further. Please wait till an execution completes or investigate previously reported failures.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If there are violations compared to the baseline, they will be listed here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if last_model_bias_monitor_execution:\n",
    "    model_bias_violations = last_model_bias_monitor_execution.constraint_violations()\n",
    "    if model_bias_violations:\n",
    "        print(model_bias_violations.body_dict)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The analysis results and CloudWatch metrics are visualized in SageMaker Studio. Select the Endpoints tab, then double click the endpoint to show the UI."
   ]
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "interpreter": {
   "hash": "c6e4e9f98eb68ad3b7c296f83d20e6de614cb42e90992a65aa266555a3137d0d"
  },
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "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.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}