<jupyter_start><jupyter_text>Deployment Pipeline NotebookThis notebook will exercise the drift detection MLOps `deployment pipeline`  SetupRetrieve the project name from your build pipeline<jupyter_code>%store -r project_name<jupyter_output><empty_output><jupyter_text>Get back the project id and region<jupyter_code>import sagemaker
import json

sess = sagemaker.session.Session()
region_name = sess._region_name
sm_client = sess.sagemaker_client
project_id = sm_client.describe_project(ProjectName=project_name)["ProjectId"]
artifact_bucket = f"sagemaker-project-{project_id}-{region_name}"

print(f"Project: {project_name} ({project_id})")<jupyter_output><empty_output><jupyter_text>Your batch pipeline should now be running, click the link below to open the AWS CodePipeline in a new window.<jupyter_code>from IPython.core.display import HTML

HTML(
    f'Open <a target="_blank" href="https://{region_name}.console.aws.amazon.com/codesuite/codepipeline/pipelines/sagemaker-{project_name}-deploy/view?region={region_name}">Code Pipeline</a> in a new window'
)<jupyter_output><empty_output><jupyter_text>Data PrepDownload the test dataset output from the pre-processing job in our build pipeline, which we will use for input to batch scoring.<jupyter_code>import boto3
import pandas as pd
import random
from sagemaker.s3 import S3Downloader, S3Uploader


def get_latest_processed_data(pipeline_name, step_name, output_name):
    execution_arn = sm_client.list_pipeline_executions(
        PipelineName=pipeline_name, SortBy="CreationTime"
    )["PipelineExecutionSummaries"][0]["PipelineExecutionArn"]
    steps = sm_client.list_pipeline_execution_steps(
        PipelineExecutionArn=execution_arn, SortOrder="Ascending"
    )["PipelineExecutionSteps"]
    preprocess_arn = next(
        item["Metadata"]["ProcessingJob"]["Arn"]
        for item in steps
        if item["StepName"] == step_name
    )
    job_outputs = sm_client.describe_processing_job(
        ProcessingJobName=preprocess_arn.split("/")[1]
    )["ProcessingOutputConfig"]["Outputs"]
    return next(
        item["S3Output"]["S3Uri"]
        for item in job_outputs
        if item["OutputName"] == output_name
    )


pipeline_name = f"{project_name}-build"
test_uri = get_latest_processed_data(pipeline_name, "PreprocessData", "test")
S3Downloader().download(test_uri, "preprocessed")

# Load the test scores into a dataframe
test_df = pd.read_csv("preprocessed/test.csv")
print(test_df.shape)
test_df.head()<jupyter_output><empty_output><jupyter_text>Test StagingThe staging SageMaker endpoint is created by AWS CloudFormation in the `Batch_CFN_Staging` stage of the AWS CodePipelineOnce its created, run the next cell to wait for the staging endpoint to be in service.<jupyter_code>from botocore.exceptions import WaiterError
from sagemaker.predictor import Predictor
from sagemaker.serializers import CSVSerializer
from sagemaker.deserializers import JSONDeserializer

# Define the predictor for staging
def wait_for_predictor(stage_name):
    try:
        endpoint_name = f"sagemaker-{project_name}-{stage_name}"
        predictor = Predictor(
            endpoint_name, serializer=CSVSerializer(), deserializer=JSONDeserializer()
        )
        print(
            f"Waiting for {stage_name} endpoint: {predictor.endpoint_name} to be deployed..."
        )
        sm_client.get_waiter("endpoint_in_service").wait(
            EndpointName=predictor.endpoint_name
        )
        print("Ready")
        return predictor
    except WaiterError as err:
        error_message = err.last_response["Error"]["Message"]
        if error_message.startswith("Could not find endpoint"):
            err = Exception(f"Endpoint {endpoint_name} not found.")
        raise err


predictor = wait_for_predictor("staging")<jupyter_output><empty_output><jupyter_text>Let's send some traffic to the staging endpoint with the following payload:| passenger_count	| pickup_latitude	| pickup_longitude	| dropoff_latitude	| dropoff_longitude	| geo_distance	| hour	| weekday	| month || -| - | - | - | - | - | - | - | - || 1	| -73.986114	| 40.685634	| -73.936794	| 40.715370	| 5.318025	| 7	| 0	| 2 |We expect approximately a $20 fare:<jupyter_code>payload = "1,-73.986114,40.685634,-73.936794,40.715370,5.318025,7,0,2"
predictor.predict(data=payload)<jupyter_output><empty_output><jupyter_text>Approve Staging🛑  Head back to the AWS Code Pipeline and approve the staging deployment to kick off the production deployment.  Test ProductionAfter a few minutes our production endpoint will start to be deployed.<jupyter_code>predictor = wait_for_predictor("prod")<jupyter_output><empty_output><jupyter_text>And confirm that data capture is enabled.<jupyter_code>data_capture = sm_client.describe_endpoint(EndpointName=predictor.endpoint_name)[
    "DataCaptureConfig"
]
print(f"Data capture is: {data_capture['CaptureStatus']}")<jupyter_output><empty_output><jupyter_text>Inspect Data CaptureLet's send some traffic to the producition endpoint, which our [Data Quality Monitor](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-data-quality.html) should detect as drifting from the baseline.<jupyter_code>%%time
for n in range(100):
    predictor.predict(data=payload)<jupyter_output><empty_output><jupyter_text>Let's see if we have received some outputs to our data capture<jupyter_code>data_capture_uri = data_capture["DestinationS3Uri"]
data_capture_files = S3Downloader.list(data_capture_uri)

print("Found {} files".format(len(data_capture_files)))

if data_capture["EnableCapture"] and len(data_capture_files) > 0:
    # Get the first line of the most recent file
    event = json.loads(S3Downloader.read_file(data_capture_files[-1]).split("\n")[0])
    print("\nLast file:\n{}".format(json.dumps(event, indent=2)))
elif len(data_capture_files) == 0:
    print("No files yet, please rerun this cell in a few seconds")<jupyter_output><empty_output><jupyter_text>Before we test production, let's tweak some of the columns to change the distribution of the data. This represents a simulation of reality where the distribution of the incoming data has changed due to changes in the environment.<jupyter_code>test_df["passenger_count"] = random.choices(
    [1, 2, 3, 4, 5, 6], weights=[2, 1, 2, 5, 2, 1], k=test_df.shape[0]
)
test_df["geo_distance"] = test_df["passenger_count"].apply(
    lambda x: 70 * random.betavariate(2.5, 2)
)

tweaked_rows = (
    test_df.drop("fare_amount", axis=1).to_csv(header=False, index=False).split("\n")
)<jupyter_output><empty_output><jupyter_text>Then make a series of prediction requests in the background every 10 minutes with this  data to cause an artificial model monitoring alarm to be triggered.<jupyter_code>from threading import Thread
import time


def invoke_endpoint_forever():
    while True:
        for i in range(10000):
            predictor.predict(data=tweaked_rows[i % len(tweaked_rows)])
        time.sleep(10 * 60)


Thread(target=invoke_endpoint_forever).start()<jupyter_output><empty_output><jupyter_text>In the above code can change the sleep time, the number of requests per batch, or the randomly generated data, to different values. This will allow you to test your endpoint with more requests per second or to see how different changes in data would affect your monitoring.  You can even completely remove the sleep time so that the kernel will be hitting the endpoint as fast as it can. However, this will cause the endpoint to work harder and trigger the automatic scaling to increase the underlying infrastructure used by the endpoint, which might incur higher costs.  MonitorLet's check that we have a monitor configured and that its schedule.<jupyter_code>from datetime import datetime, timedelta
from dateutil.tz import tzlocal

model_monitor = predictor.list_monitors()[0]
model_monitor_status = model_monitor.describe_schedule()["MonitoringScheduleStatus"]
print(f"Model Monitoring: {model_monitor_status}")

now = datetime.now(tzlocal())
next_hour = (now + timedelta(hours=1)).replace(minute=0)
scheduled_diff = (next_hour - now).seconds // 60
print("Next schedule in {} minutes".format(scheduled_diff))<jupyter_output><empty_output><jupyter_text>List the latest execution and output the status<jupyter_code>monitor_executions = model_monitor.list_executions()
if len(monitor_executions) == 0:
    raise (Exception("Please wait, no monitor executions available yet"))

# Get the latest monitor status
monitor_status = monitor_executions[0].describe()["ProcessingJobStatus"]
if monitor_status == "Completed":
    monitor_message = monitor_executions[0].describe()["ExitMessage"]
    print(f"Latest execution: {monitor_message}")
else:
    print(f"Latest execution: {monitor_status}")<jupyter_output><empty_output><jupyter_text>Inspect Model Monitor report🛑 Browse to the model monitoring results in SageMaker Studio to download and run a report  RetrainWhen the model monitoring schedule runs it will publish Amazon [CloudWatch Metrics](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-interpreting-cloudwatch.html).  If drift is detected for a metric above the threshold defined in the `prod-config.json` in the deployment pipeline, then the Amazon CloudWatch will Alarm resulting in the SageMaker pipeline to be re-trained.You can simulate drift by putting a metric value above the threshold of `0.5` directly into CloudWatch.<jupyter_code>import boto3
from datetime import datetime, timedelta
from dateutil.tz import tzlocal
import random


cloudwatch = boto3.client("cloudwatch")

# Define the metric name and threshold
endpoint_name = predictor.endpoint_name
schedule_name = f"{endpoint_name}-threshold"
metric_name = "feature_baseline_drift_fare_amount"
metric_threshold = 0.5

# Put a new metric to trigger an alaram
def put_drift_metric(value):
    print("Putting metric: {}".format(value))
    response = cloudwatch.put_metric_data(
        Namespace="aws/sagemaker/Endpoints/data-metrics",
        MetricData=[
            {
                "MetricName": metric_name,
                "Dimensions": [
                    {"Name": "MonitoringSchedule", "Value": schedule_name},
                    {"Name": "Endpoint", "Value": endpoint_name},
                ],
                "Timestamp": datetime.now(),
                "Value": value,
                "Unit": "None",
            },
        ],
    )


def get_drift_stats():
    response = cloudwatch.get_metric_statistics(
        Namespace="aws/sagemaker/Endpoints/data-metrics",
        MetricName=metric_name,
        Dimensions=[
            {"Name": "MonitoringSchedule", "Value": schedule_name},
            {"Name": "Endpoint", "Value": endpoint_name},
        ],
        StartTime=datetime.now() - timedelta(minutes=2),
        EndTime=datetime.now(),
        Period=1,
        Statistics=["Average"],
        Unit="None",
    )
    if "Datapoints" in response and len(response["Datapoints"]) > 0:
        return response["Datapoints"][0]["Average"]
    return 0


print("Simluate drift on endpoint: {}".format(endpoint_name))

while True:
    put_drift_metric(round(random.uniform(metric_threshold, 1.0), 4))
    drift_stats = get_drift_stats()
    print("Average drift amount: {}".format(get_drift_stats()))
    if drift_stats > metric_threshold:
        break
    time.sleep(1)<jupyter_output><empty_output><jupyter_text>To see the CloudWatch metric Alarm click on the link below.<jupyter_code>HTML(
    f'Open <a target="_blank" href="https://{region_name}.console.aws.amazon.com/cloudwatch/home?region={region_name}#alarmsV2:alarm/{schedule_name}">CloudWatch Alarm</a> in new window'
)<jupyter_output><empty_output><jupyter_text>This will result in a new SageMaker pipeline execution starting.<jupyter_code>latest_pipeline_execution = sm_client.list_pipeline_executions(
    PipelineName=pipeline_name,
)["PipelineExecutionSummaries"][0]
latest_execution_status = latest_pipeline_execution["PipelineExecutionStatus"]
time_ago = datetime.now(tzlocal()) - latest_pipeline_execution["StartTime"]

print(
    f"Latest pipeline: {pipeline_name} execution: {latest_execution_status} started {time_ago.total_seconds()/60:0.2f} mins ago"
)<jupyter_output><empty_output><jupyter_text>We can verify that this was triggered by Drift by inspecting the InputSource:<jupyter_code>params = sm_client.list_pipeline_parameters_for_execution(
    PipelineExecutionArn=latest_pipeline_execution["PipelineExecutionArn"],
)
input_source = [
    p["Value"] for p in params["PipelineParameters"] if p["Name"] == "InputSource"
][0]
print(f"Pipeline execution started with InputSource: {input_source}")<jupyter_output><empty_output><jupyter_text>And let's list the steps of that execution.<jupyter_code>execution_steps = sm_client.list_pipeline_execution_steps(
    PipelineExecutionArn=latest_pipeline_execution["PipelineExecutionArn"],
)["PipelineExecutionSteps"]
for step in execution_steps:
    print("Step: {}, Status: {}".format(step["StepName"], step["StepStatus"]))<jupyter_output><empty_output><jupyter_text>✅ Great now you have completed all the steps.  Clean upExecute the following cell to delete cloudformation stacks1. SageMaker prod endpoint2. SageMaker staging endpoint<jupyter_code>import boto3

cfn = boto3.client("cloudformation")

for stack_name in [
    f"sagemaker-{project_name}-deploy-prod",
    f"sagemaker-{project_name}-deploy-staging",
]:
    print("Deleting stack: {}".format(stack_name))
    cfn.delete_stack(StackName=stack_name)
    cfn.get_waiter("stack_delete_complete").wait(StackName=stack_name)<jupyter_output><empty_output>