{"cells": [{"cell_type": "markdown", "metadata": {}, "source": ["# Safe MLOps Deployment Pipeline\n", "\n", "\n", "## Overview\n", "\n", "In this notebook you will step through an MLOps pipeline to build, train, deploy and monitor an XGBoost regression model for predicting the expected taxi fare using the New York City Taxi [dataset](https://registry.opendata.aws/nyc-tlc-trip-records-pds/)\u21d7. This safe pipeline features a [canary deployment](https://docs.aws.amazon.com/wellarchitected/latest/machine-learning-lens/canary-deployment.html) strategy with rollback on error. You will learn how to trigger and monitor the pipeline, inspect the training workflow, use model monitor to set up alerts, and create a canary deployment.\n", "\n", "\n", " Note: This notebook assumes prior familiarity with the basics training ML models on Amazon SageMaker. Data preparation and visualization, although present, will be kept to a minimum. If you are not familiar with the basic concepts and features of SageMaker, we recommend reading the
SageMaker documentation\u21d7 and completing the workshops and samples in
AWS SageMaker Examples GitHub\u21d7 and
AWS Samples GitHub\u21d7. \n", "
\n", " Tip: You may need to wait a minute to see the DataSource stage turn green. The page will refresh automatically.\n", "
\n", "\n", ""]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["from IPython.core.display import HTML\n", "\n", "HTML(\n", " 'Code Pipeline'.format(\n", " region, pipeline_name\n", " )\n", ")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Inspect Build Logs\n", "\n", "Once the build stage is running, you will see the AWS CodeBuild job turn blue with a status of **In progress**.\n", "\n", ""]}, {"cell_type": "markdown", "metadata": {}, "source": ["You can click on the **Details** link displayed in the CodePipeline UI or click the link below to jump directly to the CodeBuild logs.\n", "\n", "\n", " Tip: You may need to wait a few seconds for the pipeline to transition into the active (blue) state and for the build to start.\n", "
"]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["codepipeline = boto3.client(\"codepipeline\")\n", "\n", "\n", "def get_pipeline_stage(pipeline_name, stage_name):\n", " response = codepipeline.get_pipeline_state(name=pipeline_name)\n", " for stage in response[\"stageStates\"]:\n", " if stage[\"stageName\"] == stage_name:\n", " return stage\n", "\n", "\n", "# Get last execution id\n", "build_stage = get_pipeline_stage(pipeline_name, \"Build\")\n", "if not \"latestExecution\" in build_stage:\n", " raise (Exception(\"Please wait. Build not started\"))\n", "\n", "build_url = build_stage[\"actionStates\"][0][\"latestExecution\"][\"externalExecutionUrl\"]\n", "\n", "# Out a link to the code build logs\n", "HTML('Code Build Logs'.format(build_url))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["The AWS CodeBuild process is responsible for creating a number of AWS CloudFormation templates which we will explore in more detail in the next section. Two of these templates are used to set up the **Train** step by creating the AWS Step Functions worklow and the custom AWS Lambda functions used within this workflow."]}, {"cell_type": "markdown", "metadata": {}, "source": ["## Train Model\n", "\n", "### Inspect Training Job\n", "\n", "Wait until the pipeline has started running the Train step (see screenshot) before continuing with the next cells in this notebook. \n", "\n", "\n", "\n", "When the pipeline has started running the train step, you can click on the **Details** link displayed in the CodePipeline UI (see screenshot above) to view the Step Functions workflow which is running the training job. \n", "\n", "Alternatively, you can click on the Workflow link from the cell output below once it's available."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["from stepfunctions.workflow import Workflow\n", "\n", "while True:\n", " try:\n", " workflow = Workflow.attach(workflow_pipeline_arn)\n", " break\n", " except ClientError as e:\n", " print(e.response[\"Error\"][\"Message\"])\n", " time.sleep(10)\n", "\n", "workflow"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Review Build Script\n", "\n", "While you wait for the training job to complete, let's take a look at the `run.py` code which was used by the AWS CodeBuild process.\n", "\n", "This script takes all of the input parameters, including the dataset locations and hyperparameters which you saved to JSON files earlier in this notebook, and uses them to generate the templates which the pipeline needs to run the training job. It *does not* create the actual Step Functions instance - it only generates the templates which define the Step Functions workflow, as well as the CloudFormation input templates which CodePipeline uses to instantiate the Step Functions instance.\n", "\n", "Step-by-step, the script does the following:\n", "\n", "1. It collects all the input parameters it needs to generate the templates. This includes information about the environment container needed to run the training job, the input and output data locations, IAM roles needed by various components, encryption keys, and more. It then sets up some basic parameters like the AWS region and the function names.\n", "1. If the input parameters specify an environment container stored in ECR, it fetches that container. Otherwise, it fetches the URI of the AWS managed environment container needed for the training job.\n", "1. It reads the input data JSON file which you generated earlier in this notebook (and which was included in the zip source for the pipeline), thereby fetching the locations of the train, validation, and baseline data files. Then it formats more parameters which will be needed later in the script, including version IDs and output data locations.\n", "1. It reads the hyperparameter JSON file which you generated earlier in this notebook.\n", "1. It defines the Step Functions workflow, starting with the input schema, followed by each step of the workflow (i.e. Create Experiment, Baseline Job, Training Job), and finally combines those steps into a workflow graph. \n", "1. The workflow graph is saved to file, along with a file containing all of the input parameters saved according to the schema defined in the workflow.\n", "1. It saves parameters to file which will be used by CloudFormation to instantiate the Step Functions workflow."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["!pygmentize ../model/run_pipeline.py"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Customize Workflow (Optional)\n", "\n", "If you are interested in customising the workflow used in the Build Script, store the `input_data` to be used within the local [workflow.ipynb](workflow.ipynb) notebook. The workflow notebook can be used to experiment with the Step Functions workflow and training job definitions for your model."]}, {"cell_type": "code", "execution_count": null, "metadata": {"scrolled": true}, "outputs": [], "source": ["%store input_data PROVISIONED_PRODUCT_NAME"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Training Analytics\n", "\n", "Once the training and baseline jobs are complete (meaning they are displayed in a green color in the Step Functions workflow, this takes around 5 minutes), you can inspect the experiment metrics. The code below will display all experiments in a table. Note that the baseline processing job won't have RMSE metrics - it calculates metrics based on the training data, but does not train a machine learning model. \n", "\n", "You will [explore the baseline](#Explore-Baseline) results later in this notebook. "]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["from sagemaker import analytics\n", "\n", "experiment_name = \"mlops-{}\".format(model_name)\n", "model_analytics = analytics.ExperimentAnalytics(experiment_name=experiment_name)\n", "analytics_df = model_analytics.dataframe()\n", "\n", "if analytics_df.shape[0] == 0:\n", " raise (Exception(\"Please wait. No training or baseline jobs\"))\n", "\n", "pd.set_option(\"display.max_colwidth\", 100) # Increase column width to show full copmontent name\n", "cols = [\n", " \"TrialComponentName\",\n", " \"DisplayName\",\n", " \"SageMaker.InstanceType\",\n", " \"train:rmse - Last\",\n", " \"validation:rmse - Last\",\n", "] # return the last rmse for training and validation\n", "analytics_df[analytics_df.columns & cols].head(2)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["## Deploy Dev\n", "\n", "### Test Dev Deployment\n", "\n", "When the pipeline has finished training a model, it automatically moves to the next step, where the model is deployed as a SageMaker Endpoint. This endpoint is part of your dev deployment, therefore, in this section, you will run some tests on the endpoint to decide if you want to deploy this model into production.\n", "\n", "First, run the cell below to fetch the name of the SageMaker Endpoint."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["codepipeline = boto3.client(\"codepipeline\")\n", "\n", "deploy_dev = get_pipeline_stage(pipeline_name, \"DeployDev\")\n", "if not \"latestExecution\" in deploy_dev:\n", " raise (Exception(\"Please wait. Deploy dev not started\"))\n", "\n", "execution_id = deploy_dev[\"latestExecution\"][\"pipelineExecutionId\"]\n", "dev_endpoint_name = \"mlops-{}-dev-{}\".format(model_name, execution_id)\n", "\n", "print(\"endpoint name: {}\".format(dev_endpoint_name))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["If you moved through the previous section very quickly, you will need to wait until the dev endpoint has been successfully deployed and the pipeline is waiting for approval to deploy to production (see screenshot). It can take up to 10 minutes for SageMaker to create an endpoint.\n", "\n", "\n", "\n", "Alternatively, run the code below to check the status of your endpoint. Wait until the status of the endpoint is 'InService'."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["sm = boto3.client(\"sagemaker\")\n", "\n", "while True:\n", " try:\n", " response = sm.describe_endpoint(EndpointName=dev_endpoint_name)\n", " print(\"Endpoint status: {}\".format(response[\"EndpointStatus\"]))\n", " if response[\"EndpointStatus\"] == \"InService\":\n", " break\n", " except ClientError as e:\n", " print(e.response[\"Error\"][\"Message\"])\n", " time.sleep(10)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Now that your endpoint is ready, let's write some code to run the test data (which you split off from the dataset and saved to file at the start of this notebook) through the endpoint for inference. The code below supports both v1 and v2 of the SageMaker SDK, but we recommend using v2 of the SDK in all of your future projects."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["import numpy as np\n", "from tqdm import tqdm\n", "\n", "from sagemaker.predictor import Predictor\n", "from sagemaker.serializers import CSVSerializer\n", "\n", "\n", "def get_predictor(endpoint_name):\n", " xgb_predictor = Predictor(endpoint_name)\n", " xgb_predictor.serializer = CSVSerializer()\n", " return xgb_predictor\n", "\n", "\n", "def predict(predictor, data, rows=500):\n", " split_array = np.array_split(data, round(data.shape[0] / float(rows)))\n", " predictions = \"\"\n", " for array in tqdm(split_array):\n", " predictions = \",\".join([predictions, predictor.predict(array).decode(\"utf-8\")])\n", " return np.fromstring(predictions[1:], sep=\",\")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Now use the `predict` function, which was defined in the code above, to run the test data through the endpoint and generate the predictions."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["dev_predictor = get_predictor(dev_endpoint_name)\n", "predictions = predict(dev_predictor, test_df[test_df.columns[1:]].values)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Next, load the predictions into a data frame, and join it with your test data. Then, calculate absolute error as the difference between the actual taxi fare and the predicted taxi fare. Display the results in a table, sorted by the highest absolute error values."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["pred_df = pd.DataFrame({\"total_amount_predictions\": predictions})\n", "pred_df = test_df.join(pred_df) # Join on all\n", "pred_df[\"error\"] = abs(pred_df[\"total_amount\"] - pred_df[\"total_amount_predictions\"])\n", "\n", "pred_df.sort_values(\"error\", ascending=False).head()"]}, {"cell_type": "markdown", "metadata": {}, "source": ["From this table, we note that some short trip distances have large errors because the low predicted fare does not match the high actual fare. This could be the result of a generous tip which we haven't included in this dataset.\n", "\n", "You can also analyze the results by plotting the absolute error to visualize outliers. In this graph, we see that most of the outliers are cases where the model predicted a much lower fare than the actual fare. There are only a few outliers where the model predicted a higher fare than the actual fare."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["sns.scatterplot(data=pred_df, x=\"total_amount_predictions\", y=\"total_amount\", hue=\"error\")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["If you want one overall measure of quality for the model, you can calculate the root mean square error (RMSE) for the predicted fares compared to the actual fares. Compare this to the [results calculated on the validation set](#validation-results) at the end of the 'Inspect Training Job' section."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["from math import sqrt\n", "from sklearn.metrics import mean_squared_error\n", "\n", "\n", "def rmse(pred_df):\n", " return sqrt(mean_squared_error(pred_df[\"total_amount\"], pred_df[\"total_amount_predictions\"]))\n", "\n", "\n", "print(\"RMSE: {}\".format(rmse(pred_df)))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["## Deploy Prod\n", "\n", "### Approve Deployment to Production\n", "\n", "If you are happy with the results of the model, you can go ahead and approve the model to be deployed into production. You can do so by clicking the **Review** button in the CodePipeline UI, leaving a comment to explain why you approve this model, and clicking on **Approve**. \n", "\n", "Alternatively, you can create a Jupyter widget which (when enabled) allows you to comment and approve the model directly from this notebook. Run the cell below to see this in action."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["import ipywidgets as widgets\n", "\n", "\n", "def on_click(obj):\n", " result = {\"summary\": approval_text.value, \"status\": obj.description}\n", " response = codepipeline.put_approval_result(\n", " pipelineName=pipeline_name,\n", " stageName=\"DeployDev\",\n", " actionName=\"ApproveDeploy\",\n", " result=result,\n", " token=approval_action[\"token\"],\n", " )\n", " button_box.close()\n", " print(result)\n", "\n", "\n", "# Create the widget if we are ready for approval\n", "deploy_dev = get_pipeline_stage(pipeline_name, \"DeployDev\")\n", "if not \"latestExecution\" in deploy_dev[\"actionStates\"][-1]:\n", " raise (Exception(\"Please wait. Deploy dev not complete\"))\n", "\n", "approval_action = deploy_dev[\"actionStates\"][-1][\"latestExecution\"]\n", "if approval_action[\"status\"] == \"Succeeded\":\n", " print(\"Dev approved: {}\".format(approval_action[\"summary\"]))\n", "elif \"token\" in approval_action:\n", " approval_text = widgets.Text(placeholder=\"Optional approval message\")\n", " approve_btn = widgets.Button(description=\"Approved\", button_style=\"success\", icon=\"check\")\n", " reject_btn = widgets.Button(description=\"Rejected\", button_style=\"danger\", icon=\"close\")\n", " approve_btn.on_click(on_click)\n", " reject_btn.on_click(on_click)\n", " button_box = widgets.HBox([approval_text, approve_btn, reject_btn])\n", " display(button_box)\n", "else:\n", " raise (Exception(\"Please wait. No dev approval\"))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Test Production Deployment\n", "\n", "Within about a minute after approving the model deployment, you should see the pipeline start on the final step: deploying your model into production. In this section, you will check the deployment status and test the production endpoint after it has been deployed.\n", "\n", "\n", "\n", "This step of the pipeline uses CloudFormation to deploy a number of resources on your behalf. In particular, it creates:\n", "\n", "1. A production-ready SageMaker Endpoint for your model, with [data capture](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-data-capture.html)\u21d7 (used by SageMaker Model Monitor) and [autoscaling](https://docs.aws.amazon.com/sagemaker/latest/dg/endpoint-auto-scaling.html)\u21d7 enabled.\n", "1. A [model monitoring schedule](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-scheduling.html)\u21d7 which outputs the results to CloudWatch metrics, along with a [CloudWatch Alarm](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/AlarmThatSendsEmail.html)\u21d7 which will notify you when a violation occurs. \n", "1. A CodeDeploy instance which creates a simple app by deploying API Gateway, three Lambda functions, and an alarm to notify of the success or failure of this deployment. The code for the Lambda functions can be found in `api/app.py`, `api/pre_traffic_hook.py`, and `api/post_traffic_hook.py`. These functions update the endpoint to enable data capture, format and submit incoming traffic to the SageMaker endpoint, and capture the data logs.\n", "\n", "\n", "\n", "Let's check how the deployment is progressing. Use the code below to fetch the execution ID of the deployment step. Then generate a table which lists the resources created by the CloudFormation stack and their creation status. You can re-run the cell after a few minutes to see how the steps are progressing."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["deploy_prd = get_pipeline_stage(pipeline_name, \"DeployPrd\")\n", "if not \"latestExecution\" in deploy_prd or not \"latestExecution\" in deploy_prd[\"actionStates\"][0]:\n", " raise (Exception(\"Please wait. Deploy prd not started\"))\n", "\n", "execution_id = deploy_prd[\"latestExecution\"][\"pipelineExecutionId\"]"]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["from datetime import datetime, timedelta\n", "from dateutil.tz import tzlocal\n", "\n", "\n", "def get_event_dataframe(events):\n", " stack_cols = [\n", " \"LogicalResourceId\",\n", " \"ResourceStatus\",\n", " \"ResourceStatusReason\",\n", " \"Timestamp\",\n", " ]\n", " stack_event_df = pd.DataFrame(events)[stack_cols].fillna(\"\")\n", " stack_event_df[\"TimeAgo\"] = datetime.now(tzlocal()) - stack_event_df[\"Timestamp\"]\n", " return stack_event_df.drop(\"Timestamp\", axis=1)\n", "\n", "\n", "cfn = boto3.client(\"cloudformation\")\n", "\n", "stack_name = stack_name = \"{}-deploy-prd\".format(pipeline_name)\n", "print(\"stack name: {}\".format(stack_name))\n", "\n", "# Get latest stack events\n", "while True:\n", " try:\n", " response = cfn.describe_stack_events(StackName=stack_name)\n", " break\n", " except ClientError as e:\n", " print(e.response[\"Error\"][\"Message\"])\n", " time.sleep(10)\n", "\n", "get_event_dataframe(response[\"StackEvents\"]).head()"]}, {"cell_type": "markdown", "metadata": {}, "source": ["The resource of most interest to us is the endpoint. This takes on average 10 minutes to deploy. In the meantime, you can take a look at the Python code used for the application. \n", "\n", "The `app.py` is the main entry point invoking the Amazon SageMaker endpoint. It returns results along with a custom header for the endpoint we invoked."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["!pygmentize ../api/app.py"]}, {"cell_type": "markdown", "metadata": {}, "source": ["The `pre_traffic_hook.py` lambda is invoked prior to deployment and confirms the endpoint has data capture enabled."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["!pygmentize ../api/pre_traffic_hook.py"]}, {"cell_type": "markdown", "metadata": {}, "source": ["The `post_traffic_hook.py` lambda is invoked to perform any final checks, in this case to verify that we have received log data from data capature."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["!pygmentize ../api/post_traffic_hook.py"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Use the code below to fetch the name of the endpoint, then run a loop to wait for the endpoint to be fully deployed. You need the status to be 'InService'."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["prd_endpoint_name = \"mlops-{}-prd-{}\".format(model_name, execution_id)\n", "print(\"prod endpoint: {}\".format(prd_endpoint_name))"]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["sm = boto3.client(\"sagemaker\")\n", "\n", "while True:\n", " try:\n", " response = sm.describe_endpoint(EndpointName=prd_endpoint_name)\n", " print(\"Endpoint status: {}\".format(response[\"EndpointStatus\"]))\n", " # Wait until the endpoint is in service with data capture enabled\n", " if (\n", " response[\"EndpointStatus\"] == \"InService\"\n", " and \"DataCaptureConfig\" in response\n", " and response[\"DataCaptureConfig\"][\"EnableCapture\"]\n", " ):\n", " break\n", " except ClientError as e:\n", " print(e.response[\"Error\"][\"Message\"])\n", " time.sleep(10)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["When the endpoint status is 'InService', you can continue. Earlier in this notebook, you created some code to send data to the dev endpoint. Reuse this code now to send a sample of the test data to the production endpoint. Since data capture is enabled on this endpoint, you want to send single records at a time, so the model monitor can map these records to the baseline. \n", "\n", "You will [inspect the model monitor](#Inspect-Model-Monitor) later in this notebook. For now, just check if you can send data to the endpoint and receive predictions in return."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["prd_predictor = get_predictor(prd_endpoint_name)\n", "sample_values = test_df[test_df.columns[1:]].sample(100).values\n", "predictions = predict(prd_predictor, sample_values, rows=1)\n", "predictions"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Test REST API\n", "\n", "Although you already tested the SageMaker endpoint in the previous section, it is also a good idea to test the application created with API Gateway. \n", "\n", "\n", "\n", "Follow the link below to open the Lambda Deployment where you can see the in-progress and completed deployments. You can also click to expand the **SAM template** to see the packaged CloudFormation template used in the deployment."]}, {"cell_type": "code", "execution_count": null, "metadata": {"scrolled": true}, "outputs": [], "source": ["HTML(\n", " 'Lambda Deployment'.format(\n", " region, model_name\n", " )\n", ")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Run the code below to confirm that the endpoint is in service. It will complete once the REST API is available."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["def get_stack_status(stack_name):\n", " response = cfn.describe_stacks(StackName=stack_name)\n", " if response[\"Stacks\"]:\n", " stack = response[\"Stacks\"][0]\n", " outputs = None\n", " if \"Outputs\" in stack:\n", " outputs = dict([(o[\"OutputKey\"], o[\"OutputValue\"]) for o in stack[\"Outputs\"]])\n", " return stack[\"StackStatus\"], outputs\n", "\n", "\n", "outputs = None\n", "while True:\n", " try:\n", " status, outputs = get_stack_status(stack_name)\n", " response = sm.describe_endpoint(EndpointName=prd_endpoint_name)\n", " print(\"Endpoint status: {}\".format(response[\"EndpointStatus\"]))\n", " if outputs:\n", " break\n", " elif status.endswith(\"FAILED\"):\n", " raise (Exception(\"Stack status: {}\".format(status)))\n", " except ClientError as e:\n", " print(e.response[\"Error\"][\"Message\"])\n", " time.sleep(10)\n", "\n", "if outputs:\n", " print(\"deployment application: {}\".format(outputs[\"DeploymentApplication\"]))\n", " print(\"rest api: {}\".format(outputs[\"RestApi\"]))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["If you are performing an update on your production deployment as a result of running [Trigger Retraining](#Trigger-Retraining) you will then be able to expand the Lambda Deployment tab to reveal the resources. Click on the **ApiFunctionAliaslive** link to see the Lambda Deployment in progress. \n", "\n", "\n", "\n", "This page will be updated to list the deployment events. It also has a link to the Deployment Application which you can access in the output of the next cell."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["HTML(\n", " 'CodeDeploy application'.format(\n", " region, outputs[\"DeploymentApplication\"]\n", " )\n", ")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["CodeDeploy will perform a canary deployment and send 10% of the traffic to the new endpoint over a 5-minute period.\n", "\n", ""]}, {"cell_type": "markdown", "metadata": {}, "source": ["We can invoke the REST API and inspect the headers being returned to see which endpoint we are hitting. You will occasionally see the cell below show a different endpoint that settles to the new version once the stack is complete. "]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["%%time\n", "\n", "from urllib import request\n", "\n", "headers = {\"Content-type\": \"text/csv\"}\n", "payload = test_df[test_df.columns[1:]].head(1).to_csv(header=False, index=False).encode(\"utf-8\")\n", "rest_api = outputs[\"RestApi\"]\n", "\n", "while True:\n", " try:\n", " resp = request.urlopen(request.Request(rest_api, data=payload, headers=headers))\n", " print(\n", " \"Response code: %d: endpoint: %s\"\n", " % (resp.getcode(), resp.getheader(\"x-sagemaker-endpoint\"))\n", " )\n", " status, outputs = get_stack_status(stack_name)\n", " if status.endswith(\"COMPLETE\"):\n", " print(\"Deployment complete\\n\")\n", " break\n", " elif status.endswith(\"FAILED\"):\n", " raise (Exception(\"Stack status: {}\".format(status)))\n", " except ClientError as e:\n", " print(e.response[\"Error\"][\"Message\"])\n", " time.sleep(10)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["## Monitor\n", "\n", "### Inspect Model Monitor\n", "\n", "When you prepared the datasets for model training at the start of this notebook, you saved a baseline dataset (a copy of the train dataset). Then, when you approved the model for deployment into production, the pipeline set up an SageMaker Endpoint with data capture enabled and a model monitoring schedule. In this section, you will take a closer look at the model monitor results.\n", "\n", "To start off, fetch the latest production deployment execution ID."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["deploy_prd = get_pipeline_stage(pipeline_name, \"DeployPrd\")\n", "if not \"latestExecution\" in deploy_prd:\n", " raise (Exception(\"Please wait. Deploy prod not complete\"))\n", "\n", "execution_id = deploy_prd[\"latestExecution\"][\"pipelineExecutionId\"]"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Under the hood, SageMaker model monitor runs in SageMaker processing jobs. Use the execution ID to fetch the names of the processing job and the schedule."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["processing_job_name = \"mlops-{}-pbl-{}\".format(model_name, execution_id)\n", "schedule_name = \"mlops-{}-pms\".format(model_name)\n", "\n", "print(\"processing job name: {}\".format(processing_job_name))\n", "print(\"schedule name: {}\".format(schedule_name))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Explore Baseline\n", "\n", "Now fetch the baseline results from the processing job. This cell will throw an exception if the processing job is not complete - if that happens, just wait several minutes and try again. "]}, {"cell_type": "code", "execution_count": null, "metadata": {"scrolled": true}, "outputs": [], "source": ["import sagemaker\n", "from sagemaker.model_monitor import BaseliningJob, MonitoringExecution\n", "from sagemaker.s3 import S3Downloader\n", "\n", "sagemaker_session = sagemaker.Session()\n", "baseline_job = BaseliningJob.from_processing_name(sagemaker_session, processing_job_name)\n", "status = baseline_job.describe()[\"ProcessingJobStatus\"]\n", "if status != \"Completed\":\n", " raise (Exception(\"Please wait. Processing job not complete, status: {}\".format(status)))\n", "\n", "baseline_results_uri = baseline_job.outputs[0].destination"]}, {"cell_type": "markdown", "metadata": {}, "source": ["SageMaker model monitor generates two types of files. Take a look at the statistics file first. It calculates various statistics for each feature of the dataset, including the mean, standard deviation, minimum value, maximum value, and more. "]}, {"cell_type": "code", "execution_count": null, "metadata": {"scrolled": true}, "outputs": [], "source": ["import pandas as pd\n", "import json\n", "\n", "baseline_statistics = baseline_job.baseline_statistics().body_dict\n", "schema_df = pd.json_normalize(baseline_statistics[\"features\"])\n", "schema_df[\n", " [\n", " \"name\",\n", " \"numerical_statistics.mean\",\n", " \"numerical_statistics.std_dev\",\n", " \"numerical_statistics.min\",\n", " \"numerical_statistics.max\",\n", " ]\n", "].head()"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Now look at the suggested [constraints files](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-byoc-constraints.html)\u21d7. As the name implies, these are constraints which SageMaker model monitor recommends. If the live data which is sent to your production SageMaker Endpoint violates these constraints, this indicates data drift, and model monitor can raise an alert to trigger retraining. Of course, you can set different constraints based on the statistics which you viewed previously."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["baseline_constraints = baseline_job.suggested_constraints().body_dict\n", "constraints_df = pd.json_normalize(baseline_constraints[\"features\"])\n", "constraints_df.head()"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### View data capture\n", "\n", "When the \"Deploy Production\" stage of the MLOps pipeline deploys a SageMaker endpoint, it also enables data capture. This means the incoming requests to the endpoint, as well as the results from the ML model, are stored in an S3 location. Model monitor can analyze this data and compare it to the baseline to ensure that no constraints are violated. \n", "\n", "Use the code below to check how many files have been created by the data capture, and view the latest file in detail. Note, data capture relies on data being sent to the production endpoint. If you don't see any files yet, wait several minutes and try again."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["bucket = sagemaker_session.default_bucket()\n", "data_capture_logs_uri = \"s3://{}/mlops-{}/datacapture/{}\".format(\n", " bucket, model_name, prd_endpoint_name\n", ")\n", "\n", "capture_files = S3Downloader.list(data_capture_logs_uri)\n", "print(\"Found {} files\".format(len(capture_files)))\n", "\n", "if capture_files:\n", " # Get the first line of the most recent file\n", " event = json.loads(S3Downloader.read_file(capture_files[-1]).split(\"\\n\")[0])\n", " print(\"\\nLast file:\\n{}\".format(json.dumps(event, indent=2)))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### View monitoring schedule\n", "\n", "There are some useful functions for plotting and rendering distribution statistics or constraint violations provided in a `utils` file in the [SageMaker Examples GitHub](https://github.com/aws/amazon-sagemaker-examples/tree/master/sagemaker_model_monitor/visualization)\u21d7. Grab a copy of this code to use in this notebook. "]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["!wget -O utils.py --quiet https://raw.githubusercontent.com/awslabs/amazon-sagemaker-examples/master/sagemaker_model_monitor/visualization/utils.py\n", "import utils as mu"]}, {"cell_type": "markdown", "metadata": {}, "source": ["The [minimum scheduled run time](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-scheduling.html)\u21d7 for model monitor is one hour, which means you will need to wait at least an hour to see any results. Use the code below to check the schedule status and list the next run. If you are completing this notebook as part of a workshop, your host will have activities which you can complete while you wait. "]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["sm = boto3.client(\"sagemaker\")\n", "\n", "response = sm.describe_monitoring_schedule(MonitoringScheduleName=schedule_name)\n", "print(\"Schedule Status: {}\".format(response[\"MonitoringScheduleStatus\"]))\n", "\n", "now = datetime.now(tzlocal())\n", "next_hour = (now + timedelta(hours=1)).replace(minute=0)\n", "scheduled_diff = (next_hour - now).seconds // 60\n", "print(\"Next schedule in {} minutes\".format(scheduled_diff))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["While you wait, you can take a look at the CloudFormation template which is used as a base for the CloudFormation template built by CodeDeploy to deploy the production application. \n", "\n", "Alterntively, you can jump ahead to [Trigger Retraining](#Trigger-Retraining) which will kick off another run of the code pipeline whilst you wait."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["!cat ../assets/deploy-model-prd.yml"]}, {"cell_type": "markdown", "metadata": {}, "source": ["A couple of minutes after the model monitoring schedule has run, you can use the code below to fetch the latest schedule status. A completed schedule run may have found violations. "]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["processing_job_arn = None\n", "\n", "while processing_job_arn is None:\n", " try:\n", " response = sm.list_monitoring_executions(MonitoringScheduleName=schedule_name)\n", " except ClientError as e:\n", " print(e.response[\"Error\"][\"Message\"])\n", " for mon in response[\"MonitoringExecutionSummaries\"]:\n", " status = mon[\"MonitoringExecutionStatus\"]\n", " now = datetime.now(tzlocal())\n", " created_diff = (now - mon[\"CreationTime\"]).seconds // 60\n", " print(\"Schedule status: {}, Created: {} minutes ago\".format(status, created_diff))\n", " if status in [\"Completed\", \"CompletedWithViolations\"]:\n", " processing_job_arn = mon[\"ProcessingJobArn\"]\n", " break\n", " if status == \"InProgress\":\n", " break\n", " else:\n", " raise (Exception(\"Please wait. No Schedules executing\"))\n", " time.sleep(10)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### View monitoring results\n", "\n", "Once the model monitoring schedule has had a chance to run at least once, you can take a look at the results. First, load the monitoring execution results from the latest scheduled run."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["if processing_job_arn:\n", " execution = MonitoringExecution.from_processing_arn(\n", " sagemaker_session=sagemaker.Session(), processing_job_arn=processing_job_arn\n", " )\n", " exec_inputs = {inp[\"InputName\"]: inp for inp in execution.describe()[\"ProcessingInputs\"]}\n", " exec_results_uri = execution.output.destination\n", "\n", " print(\"Monitoring Execution results: {}\".format(exec_results_uri))"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Take a look at the files which have been saved in the S3 output location. If violations were found, you should see a constraint violations file in addition to the statistics and constraints file which you viewed before."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["!aws s3 ls $exec_results_uri/"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Now, fetch the monitoring statistics and violations. Then use the utils code to visualize the results in a table. It will highlight any baseline drift found by the model monitor. Drift can happen for categorical features (for inferred string styles) or for numerical features (e.g. total fare amount)."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["# Get the baseline and monitoring statistics & violations\n", "baseline_statistics = baseline_job.baseline_statistics().body_dict\n", "execution_statistics = execution.statistics().body_dict\n", "violations = execution.constraint_violations().body_dict[\"violations\"]"]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["mu.show_violation_df(\n", " baseline_statistics=baseline_statistics,\n", " latest_statistics=execution_statistics,\n", " violations=violations,\n", ")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Trigger Retraining\n", "\n", "The CodePipeline instance is configured with [CloudWatch Events](https://docs.aws.amazon.com/codepipeline/latest/userguide/create-cloudtrail-S3-source.html)\u21d7 to start the pipeline for retraining when the drift detection triggers specific metric alarms.\n", "\n", "You can simulate drift by putting a metric value above the threshold of `0.2` directly into CloudWatch. This will trigger the alarm, and start the code pipeline.\n", "\n", "\n", " Tip: This alarm is configured only for the latest production endpoint, so re-training will only occur if you are putting metrics against the latest endpoint.\n", "
\n", "\n", "\n", "\n", "Run the code below to trigger the metric alarm. The cell output will be a link to CloudWatch, where you can see the alarm (similar to the screenshot above), and a link to CodePipeline which you will see run again. Note that it can take a couple of minutes for everything to trigger."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["from datetime import datetime\n", "import random\n", "\n", "cloudwatch = boto3.client(\"cloudwatch\")\n", "\n", "# Define the metric name and threshold\n", "metric_name = \"feature_baseline_drift_total_amount\"\n", "metric_threshold = 0.2\n", "\n", "# Put a new metric to trigger an alaram\n", "def put_drift_metric(value):\n", " print(\"Putting metric: {}\".format(value))\n", " response = cloudwatch.put_metric_data(\n", " Namespace=\"aws/sagemaker/Endpoints/data-metrics\",\n", " MetricData=[\n", " {\n", " \"MetricName\": metric_name,\n", " \"Dimensions\": [\n", " {\"Name\": \"MonitoringSchedule\", \"Value\": schedule_name},\n", " {\"Name\": \"Endpoint\", \"Value\": prd_endpoint_name},\n", " ],\n", " \"Timestamp\": datetime.now(),\n", " \"Value\": value,\n", " \"Unit\": \"None\",\n", " },\n", " ],\n", " )\n", "\n", "\n", "def get_drift_stats():\n", " response = cloudwatch.get_metric_statistics(\n", " Namespace=\"aws/sagemaker/Endpoints/data-metrics\",\n", " MetricName=metric_name,\n", " Dimensions=[\n", " {\"Name\": \"MonitoringSchedule\", \"Value\": schedule_name},\n", " {\"Name\": \"Endpoint\", \"Value\": prd_endpoint_name},\n", " ],\n", " StartTime=datetime.now() - timedelta(minutes=2),\n", " EndTime=datetime.now(),\n", " Period=1,\n", " Statistics=[\"Average\"],\n", " Unit=\"None\",\n", " )\n", " if \"Datapoints\" in response and len(response[\"Datapoints\"]) > 0:\n", " return response[\"Datapoints\"][0][\"Average\"]\n", " return 0\n", "\n", "\n", "print(\"Simluate drift on endpoint: {}\".format(prd_endpoint_name))\n", "\n", "while True:\n", " put_drift_metric(round(random.uniform(metric_threshold, 1.0), 4))\n", " drift_stats = get_drift_stats()\n", " print(\"Average drift amount: {}\".format(get_drift_stats()))\n", " if drift_stats > metric_threshold:\n", " break\n", " time.sleep(1)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Click through to the Alarm and CodePipeline Execution history with the links below."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["# Output a html link to the cloudwatch dashboard\n", "metric_alarm_name = \"mlops-{}-metric-gt-threshold\".format(model_name)\n", "HTML(\n", " \"\"\"CloudWatch Alarm triggers\n", " Code Pipeline Execution\"\"\".format(\n", " region, metric_alarm_name, pipeline_name\n", " )\n", ")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Once the pipeline is running again you can jump back up to [Inspect Training Job](#Inspect-Training-Job)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["### Create a CloudWatch dashboard\n", "\n", "Finally, use the code below to create a CloudWatch dashboard to visualize the key performance metrics and alarms which you have created during this demo. The cell will output a link to the dashboard. This dashboard shows 9 charts in three rows, where the first row displays Lambda metrics, the second row displays SageMaker metrics, and the third row (shown in the screenshot below) displays the alarms set up for the pipeline.\n", "\n", ""]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["from string import Template\n", "\n", "sts = boto3.client(\"sts\")\n", "account_id = sts.get_caller_identity().get(\"Account\")\n", "dashboard_name = \"mlops-{0}-{1}\".format(model_name, config[\"SageMakerProjectId\"])\n", "\n", "with open(\"dashboard.json\") as f:\n", " dashboard_body = Template(f.read()).substitute(\n", " region=region, account_id=account_id, model_name=model_name\n", " )\n", " response = cloudwatch.put_dashboard(DashboardName=dashboard_name, DashboardBody=dashboard_body)\n", "\n", "# Output a html link to the cloudwatch dashboard\n", "HTML(\n", " 'CloudWatch Dashboard'.format(\n", " region, dashboard_name\n", " )\n", ")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Congratulations! You have made it to the end of this notebook, and have automated a safe MLOps pipeline using a wide range of AWS services. \n", "\n", "You can use the other notebook in this repository [workflow.ipynb](workflow.ipynb) to implement your own ML model and deploy it as part of this pipeline. Or, if you are finished with the content, follow the instructions in the next section to clean up the resources you have deployed."]}, {"cell_type": "markdown", "metadata": {}, "source": ["## Cleanup\n", "\n", "Execute the following cell to delete the stacks created in the pipeline. For a model name of **nyctaxi** these would be:\n", "\n", "1. *nyctaxi*-deploy-prd\n", "2. *nyctaxi*-deploy-dev\n", "3. *nyctaxi*-workflow\n", "4. sagemaker-custom-resource"]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["cfn = boto3.client(\"cloudformation\")\n", "\n", "# Delete the prod and then dev stack\n", "for stack_name in [\n", " f\"{pipeline_name}-deploy-prd\",\n", " f\"{pipeline_name}-deploy-dev\",\n", " f\"{pipeline_name}-workflow\",\n", " f\"mlops-{model_name}-{config['SageMakerProjectId']}-sagemaker-custom-resource\",\n", "]:\n", " print(\"Deleting stack: {}\".format(stack_name))\n", " cfn.delete_stack(StackName=stack_name)\n", " cfn.get_waiter(\"stack_delete_complete\").wait(StackName=stack_name)"]}, {"cell_type": "markdown", "metadata": {}, "source": ["The following code will delete the dashboard."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["cloudwatch.delete_dashboards(DashboardNames=[dashboard_name])\n", "print(\"Dashboard deleted\")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["The following code will clean up all objects in the artifact bucket and delete the SageMaker project."]}, {"cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": ["s3_resource = boto3.resource('s3')\n", "s3_artifact_bucket = s3_resource.Bucket(artifact_bucket)\n", "s3_artifact_bucket.object_versions.delete()\n", "print(\"Artifact bucket objects deleted\")\n", "\n", "sm.delete_project(\n", " ProjectName=PROJECT_NAME\n", ")\n", "print(\"SageMaker Project deleted\")"]}, {"cell_type": "markdown", "metadata": {}, "source": ["Finally, close this notebook."]}], "metadata": {"instance_type": "ml.t3.medium", "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.7.6"}}, "nbformat": 4, "nbformat_minor": 4}