batch_transform_with_lambdastep.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create a pipeline to perform batch inference utilizing a lambda step\n",
    "Amazon SageMaker Pipelines offers machine learning (ML) application developers and operations engineers the ability to orchestrate SageMaker jobs and author reproducible ML pipelines.\n",
    "\n",
    "This notebook shows how to create a pipeline that reads the latest model registered in a model registry and perform batch transformation on data. <BR>\n",
    "\n",
    "\n",
    "## Notebook Overview\n",
    "This notebook shows how to:\n",
    "- Define a LambdaStep that returns the latest model to be used by the transformation\n",
    "- Define a CreateModelStep that makes use of the model returned by the lambda\n",
    "- Define a TransformStep that creates a batch transform step using the latest model\n",
    "\n",
    "As a prerequisite, you need a trained model and a dataset. <BR>\n",
    "The pretrained model comes with this repository and is added to a new model registry as part of the Setup process. <BR>\n",
    "A sample dataset is also downloaded and is uploaded to Amazon s3 as part of the Setup process. The data used is the [UCI Machine Learning Abalone Dataset](https://archive.ics.uci.edu/ml/datasets/abalone) [1]. The aim for this task is to determine the age of an abalone from its physical measurements. At the core, this is a regression problem. However, we will use a pretrained model, as training the machine learning model is outside the scope of this notebook. If interested to see how you could use SageMaker Pipelines to train a model for this use-case, the example [Orchestrating Jobs with Amazon SageMaker Model Building Pipelines](https://github.com/aws/amazon-sagemaker-examples/blob/master/sagemaker-pipelines/tabular/abalone_build_train_deploy/sagemaker-pipelines-preprocess-train-evaluate-batch-transform.ipynb) shows exactly how to do this.\n",
    "\n",
    "\n",
    "[1] Dua, D. and Graff, C. (2019). UCI Machine Learning Repository. Irvine, CA: University of California, School of Information and Computer Science."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pre-Setup - IAM role setup\n",
    "Before you begin running this Notebook you need to set the right permissions for everything to work. Specifically in addition to the standard SageMaker permissions, what is needed is:\n",
    "- To have permissions to create a lambda function\n",
    "- To allow lambda to assume a role capable of loading data from the model registry\n",
    "\n",
    "To get set-up, head over to the IAM Management Console, locate the AmazonSagemaker-ExecutionRole and add the following permission by clicking on *Attach Policy*\n",
    "\n",
    "![iam_permissions](images/iam_permissions_lambdastep.png)\n",
    "\n",
    "Next, click on *Trust relationships* and *Edit trust relationship*. Edit the Service parameter to look like: \n",
    "\n",
    "\"Service\": [\"sagemaker.amazonaws.com\", \"lambda.amazonaws.com\"]\n",
    "\n",
    "\n",
    "\n",
    "*Please note that we are giving full access permissions to the SageMaker Execution Role as we will also be deleting the resources we create. In a real-world scenario we wouldn't need to delete the resources at the end and shouldn't therefore grant these permissions.*\n",
    "\n",
    "*Also note, that in a production system you may use different roles for the execution of the pipeline and the execution of the lambda. This example is only a demonstration of the SageMaker functionality and does not suggest how to setup a production system.*\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    },
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": [
    "!pip install \"sagemaker>=2.45.0\" --upgrade"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install \"boto3>=1.17.97\" --upgrade"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    },
    "pycharm": {
     "name": "#%%\n"
    }
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sagemaker\n",
    "import boto3\n",
    "from sagemaker import get_execution_role\n",
    "\n",
    "sm_client = boto3.client(service_name=\"sagemaker\")\n",
    "role = get_execution_role()\n",
    "sagemaker_session = sagemaker.Session()\n",
    "default_bucket = sagemaker_session.default_bucket()\n",
    "region = boto3.Session().region_name\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Execution role used by this notebook: \", role)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "prefix = \"batch-transform-with-lambda-step-pipeline-demo\"\n",
    "model_package_group_name = \"lambdaBatchTransformPipelineModelPackageGroup\"\n",
    "lambda_function_name = \"lambdaBatchTransformPipelineLambda\"\n",
    "pipeline_name = \"BatchTransformPipelineWithLambdaStep\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup\n",
    "There are certain preparatory steps we need to perform. \n",
    "\n",
    "\n",
    "First we assume that some other process/pipeline has trained a model, registered it in the Model Registry and someone has approved the model to be used for inference. <BR>\n",
    "For this we are going to use a pretrained model and create and register model to a new Model Registry.\n",
    "\n",
    "\n",
    "Secondly, we assume that some other process is generating the dataset on which we want to perform the predictions and saved this in some s3 location known to us. <BR>\n",
    "For this we are going to dowload a dataset from a public location and upload it to an s3 bucket that we will then provide as input to the pipeline. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a model registry and register model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Upload the model to Amazon s3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "base_uri = f\"s3://{default_bucket}/{prefix}/trained_model\"\n",
    "\n",
    "model_location = sagemaker.s3.S3Uploader.upload(\n",
    "    local_path=\"model.tar.gz\",\n",
    "    desired_s3_uri=base_uri,\n",
    ")\n",
    "\n",
    "image_uri = sagemaker.image_uris.retrieve(\n",
    "    framework=\"xgboost\", region=region, version=\"1.0-1\", py_version=\"py3\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a model package group and create a model package. Notice that we add the model directly with a status of `Approved`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "_ = sm_client.create_model_package_group(\n",
    "    ModelPackageGroupName=model_package_group_name,\n",
    "    ModelPackageGroupDescription=\"Demo model package group\",\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "_ = sm_client.create_model_package(\n",
    "    ModelPackageGroupName=model_package_group_name,\n",
    "    ModelApprovalStatus=\"Approved\",\n",
    "    InferenceSpecification={\n",
    "        \"Containers\": [\n",
    "            {\n",
    "                \"Image\": image_uri,\n",
    "                \"ModelDataUrl\": model_location,\n",
    "            },\n",
    "        ],\n",
    "        \"SupportedTransformInstanceTypes\": [\"ml.m5.large\"],\n",
    "        \"SupportedRealtimeInferenceInstanceTypes\": [\"ml.m5.large\"],\n",
    "        \"SupportedContentTypes\": [\n",
    "            \"text/csv\",\n",
    "        ],\n",
    "        \"SupportedResponseMIMETypes\": [\n",
    "            \"text/csv\",\n",
    "        ],\n",
    "    },\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Checking that the model was registered successfully. The function `get_approved_model` will also be used by the lambda function later to find the latest approved model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from utils import get_approved_package\n",
    "\n",
    "get_approved_package(model_package_group_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Collect data and upload to own Amazon s3 location"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "local_path = \"abalone-dataset-batch\"\n",
    "\n",
    "s3 = boto3.resource(\"s3\")\n",
    "s3.Bucket(f\"sagemaker-servicecatalog-seedcode-{region}\").download_file(\n",
    "    \"dataset/abalone-dataset-batch\", local_path\n",
    ")\n",
    "\n",
    "base_uri = f\"s3://{default_bucket}/{prefix}/batch_data\"\n",
    "batch_data_uri = sagemaker.s3.S3Uploader.upload(\n",
    "    local_path=local_path,\n",
    "    desired_s3_uri=base_uri,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Build pipeline\n",
    "We are now building the SageMaker Pipeline.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We begin by defining the Pipeline parameters. These parameters can change between consecutive pipeline runs,\n",
    "but if not set, the default values will be used. You can add any parameter that you want to be able to change\n",
    "amongst executions of the pipeline as a Pipeline Parameter."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.parameters import ParameterString\n",
    "\n",
    "batch_data = ParameterString(\n",
    "    name=\"BatchData\",\n",
    "    default_value=batch_data_uri,\n",
    ")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then, we define the lambda step that is responsible for fetching the latest model from the model registry"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.lambda_helper import Lambda\n",
    "\n",
    "# Lambda helper class can be used to create the Lambda function\n",
    "func = Lambda(\n",
    "    function_name=lambda_function_name,\n",
    "    execution_role_arn=role,\n",
    "    script=\"lambda_step_code.py\",\n",
    "    handler=\"lambda_step_code.handler\",\n",
    "    timeout=600,\n",
    "    memory_size=128,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.lambda_step import (\n",
    "    LambdaStep,\n",
    "    LambdaOutput,\n",
    "    LambdaOutputTypeEnum,\n",
    ")\n",
    "\n",
    "step_latest_model_fetch = LambdaStep(\n",
    "    name=\"fetchLatestModel\",\n",
    "    lambda_func=func,\n",
    "    inputs={\n",
    "        \"model_package_group_name\": model_package_group_name,\n",
    "    },\n",
    "    outputs=[\n",
    "        LambdaOutput(output_name=\"ModelUrl\", output_type=LambdaOutputTypeEnum.String), \n",
    "        LambdaOutput(output_name=\"ImageUri\", output_type=LambdaOutputTypeEnum.String), \n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At the next step we define the model.\n",
    "\n",
    "Note that we use the output of the previous step, which is the output of the lambda function,\n",
    "as the input for this step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.model import Model\n",
    "\n",
    "model = Model(\n",
    "    image_uri=step_latest_model_fetch.properties.Outputs[\"ImageUri\"],\n",
    "    model_data=step_latest_model_fetch.properties.Outputs[\"ModelUrl\"],\n",
    "    sagemaker_session=sagemaker_session,\n",
    "    role=role,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.inputs import CreateModelInput\n",
    "from sagemaker.workflow.steps import CreateModelStep\n",
    "\n",
    "\n",
    "inputs = CreateModelInput(\n",
    "    instance_type=\"ml.m5.large\",\n",
    ")\n",
    "step_create_model = CreateModelStep(\n",
    "    name=\"CreateModel\",\n",
    "    model=model,\n",
    "    inputs=inputs,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The final step in the pipeline is the Transformer Step, which is the step when the batch transformation\n",
    "of the data takes place."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.transformer import Transformer\n",
    "\n",
    "\n",
    "transformer = Transformer(\n",
    "    model_name=step_create_model.properties.ModelName,\n",
    "    instance_type=\"ml.m5.large\",\n",
    "    instance_count=1,\n",
    "    output_path=f\"s3://{default_bucket}/AbaloneTransform\",\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.inputs import TransformInput\n",
    "from sagemaker.workflow.steps import TransformStep\n",
    "\n",
    "\n",
    "step_transform = TransformStep(\n",
    "    name=\"Transform\", transformer=transformer, inputs=TransformInput(data=batch_data)\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To tie together the above steps, we define a Pipeline object as below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.pipeline import Pipeline\n",
    "\n",
    "\n",
    "pipeline = Pipeline(\n",
    "    name=pipeline_name,\n",
    "    parameters=[\n",
    "        batch_data,\n",
    "    ],\n",
    "    steps=[step_latest_model_fetch, step_create_model, step_transform],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "\n",
    "definition = json.loads(pipeline.definition())\n",
    "# definition"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Execute pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline.upsert(role_arn=role)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution = pipeline.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution.wait()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once you start the execution of the pipeline, head over to the Pipelines SageMaker Resources (shown in the picture below)\n",
    "\n",
    "![pipelineexecution](images/pipelineexecution.png)\n",
    "\n",
    "Feel free to double-click on the executing pipeline to get more details on the progress of the execution.\n",
    "\n",
    "Double-clicking on the pipeline execution (or on the tab named \"Graph\") will open a page where SageMaker constracts\n",
    "the DAG that represents the Pipeline you defined. If the pipeline execution is opened, the progress of the execution\n",
    "of the pipeline will be represented using a standard colouring format (green for succeeded, blue for running,\n",
    "red for failed), like in the image below:\n",
    "\n",
    "![pipelinegraph](images/pipelinegraph.png)\n",
    "\n",
    "\n",
    "The pipeline will be finished and marked green, in about 4 minutes. You may need to refresh the page to see all\n",
    "these steps being marked green. Lastly, you can click on the \"Transform\" step which will open the details of the step,\n",
    "including the data output location on Amazon s3 where the transformed data have been saved."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Clean-up\n",
    "\n",
    "You have successfully run this notebook.\n",
    "\n",
    "Run the following steps to delete all resources created by this example. <BR>\n",
    "Please note that the following are not deleting data from your Amazon s3 bucket, including the original trained model, the original data, the transformed data as well as any intermediate data created by the Pipeline Steps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sm_client.delete_pipeline(PipelineName=pipeline_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from utils import get_approved_package\n",
    "\n",
    "arn_of_model_to_delete = get_approved_package(model_package_group_name)[\"ModelPackageArn\"]\n",
    "sm_client.delete_model_package(\n",
    "    ModelPackageName=arn_of_model_to_delete,\n",
    ")\n",
    "sm_client.delete_model_package_group(\n",
    "    ModelPackageGroupName=model_package_group_name,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lambda_client = boto3.client(\"lambda\")\n",
    "lambda_client.delete_function(FunctionName=lambda_function_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (Data Science)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:eu-west-1:470317259841:image/datascience-1.0"
  },
  "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}