   ml_ops/sm-pipelines_callback_step/sm-pipelines_callback_step.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "6b68cbe5",
   "metadata": {},
   "source": [
    "# Glue ETL as part of a SageMaker pipeline\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a76253d7",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "This notebook's CI test result for us-west-2 is as follows. CI test results in other regions can be found at the end of the notebook. \n",
    "\n",
    "![This us-west-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-west-2/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "835e77af",
   "metadata": {},
   "source": [
    "\n",
    "This notebook will show how to use the [Callback Step](https://docs.aws.amazon.com/sagemaker/latest/dg/build-and-manage-steps.html#step-type-callback) to extend your SageMaker Pipeline steps to include tasks performed by other AWS services or custom integrations.  For this notebook, you'll learn how to include a Glue ETL job as part of a SageMaker ML pipeline.  The overall flow will be:\n",
    "\n",
    "* Define Glue ETL job\n",
    "* Run Spark data preparation job in Glue\n",
    "* Run ML training job on SageMaker\n",
    "* Evaluate ML model performance \n",
    "\n",
    "The pipeline sends a message to an SQS queue.  A Lambda function responds to SQS and invokes an ECS Fargate task.  The task will handle running the Spark job and monitoring for progress.  It'll then send the callback token back to the pipeline.\n",
    "\n",
    "![CustomStepPipeline](./images/pipelinescustom.png)\n",
    "\n",
    "## Data set\n",
    "\n",
    "We'll use the Yellow Taxi records from [NYC](https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page) in 2020.  In this [blog](https://aws.amazon.com/blogs/machine-learning/use-the-built-in-amazon-sagemaker-random-cut-forest-algorithm-for-anomaly-detection/), we used a prepared version of the data that had passenger counts per half hour.  In this notebook we'll take the raw NYC data and prepare the half-hour totals.\n",
    "\n",
    "## One-time setup\n",
    "\n",
    "This notebook needs permissions to:\n",
    "\n",
    "* Create Lambda functions\n",
    "* Create an ECS cluster\n",
    "* Upload images to ECR\n",
    "* Create IAM roles\n",
    "* Invoke SageMaker API for pipelines\n",
    "* Create security groups\n",
    "* Write data into S3\n",
    "* Create security groups\n",
    "* Describe VPC information\n",
    "\n",
    "In a production setting, we would deploy a lot of these resources using an infrastructure-as-code tool like CloudFormation or the CDK.  But for simplicity in this demo we'll create everything in this notebook."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1eaef884",
   "metadata": {},
   "source": [
    "## Setup prerequisite IAM roles"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "035d7940",
   "metadata": {},
   "source": [
    "First we need to create the following IAM roles:\n",
    "\n",
    "* A role for the ECS Fargate task and task runner. Besides the usual policies that allow pulling images and creating logs, the task needs permission to start and monitor a Glue job, and send the callback token to SageMaker. Because the specific SageMaker action isn't visible in IAM yet, for now we give the task full SageMaker permissions.\n",
    "* A role for Glue with permissions to read and write from our S3 bucket.\n",
    "* A role for Lambda with permissions to run an ECS task, send the failure callback if something goes wrong, and poll SQS.\n",
    "\n",
    "For your convenience, we have prepared the setup_iam_roles.py script to help create the IAM roles and respective policies. In most cases, this script will be run by administrator teams, on behalf of data scientists."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9fea165f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import sagemaker\n",
    "from setup_iam_roles import create_glue_pipeline_role\n",
    "from setup_iam_roles import create_lambda_sm_pipeline_role\n",
    "from setup_iam_roles import create_ecs_task_role, create_task_runner_role\n",
    "\n",
    "sagemaker_session = sagemaker.session.Session()\n",
    "default_bucket = sagemaker_session.default_bucket()\n",
    "default_bucket_prefix = sagemaker_session.default_bucket_prefix\n",
    "default_bucket_prefix_path = \"\"\n",
    "\n",
    "# If a default bucket prefix is specified, append it to the s3 path\n",
    "if default_bucket_prefix:\n",
    "    default_bucket_prefix_path = f\"{default_bucket_prefix}/\"\n",
    "\n",
    "ecs_role_arn = create_ecs_task_role(role_name=\"fg_task_pipeline_role\")\n",
    "task_role_arn = create_task_runner_role(role_name=\"fg_task_runner_pipeline_role\")\n",
    "glue_role_arn = create_glue_pipeline_role(role_name=\"glue_pipeline_role\", bucket=default_bucket)\n",
    "\n",
    "lambda_role_arn = create_lambda_sm_pipeline_role(\n",
    "    role_name=\"lambda_sm_pipeline_role\", ecs_role_arn=ecs_role_arn, task_role_arn=task_role_arn\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "405b6d29",
   "metadata": {},
   "source": [
    "## Processing\n",
    "\n",
    "Setup the configurations & tasks that will be used to process data in the pipeline. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cc66a4ad",
   "metadata": {},
   "source": [
    "### Set up ECS Fargate cluster\n",
    "\n",
    "The ECS Fargate cluster will be used to execute a Fargate task that will handle running the Spark data pre-processing in Glue and monitoring for progress.  This task is invoked by a Lambda function that gets called whenever the CallbackStep puts a message to SQS.\n",
    "\n",
    "**Pipeline Step Tasks:** *CallbackStep -> SQS -> Lambda -> Fargate Task -> Glue Job*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "67f8716e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import boto3\n",
    "\n",
    "ecs = boto3.client(\"ecs\")\n",
    "\n",
    "response = ecs.create_cluster(clusterName=\"FargateTaskRunner\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9ec51d60",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "print(f\"Cluster Name: {response['cluster']['clusterName']}\")\n",
    "print(f\"Cluster ARN: {response['cluster']['clusterArn']}\")\n",
    "print(f\"Cluster Status: {response['cluster']['status']}\")\n",
    "cluster_arn = response[\"cluster\"][\"clusterArn\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "08e5560a",
   "metadata": {},
   "source": [
    "### Build container image for Fargate task\n",
    "\n",
    "First, install the Amazon SageMaker Studio Build CLI convenience package that allows you to build docker images from your Studio environment.  Please ensure you have the pre-requisites in place as outlined in this [blog](https://aws.amazon.com/blogs/machine-learning/using-the-amazon-sagemaker-studio-image-build-cli-to-build-container-images-from-your-studio-notebooks/)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2f752c57",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import sys\n",
    "\n",
    "!{sys.executable} -m pip install sagemaker_studio_image_build"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "20fa719b",
   "metadata": {},
   "source": [
    "Next, write the code to your local environment that will be used to build the docker image.  \n",
    "\n",
    "**task.py:** This code will be used by the task runner to start and monitor the Glue job then report status back to SageMaker Pipelines via *send_pipeline_execution_step_success* or *send_pipeline_execution_step_failure*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "11af97b4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "!mkdir container"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "94543f8f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%writefile container/task.py\n",
    "\n",
    "import boto3\n",
    "import os\n",
    "import sys\n",
    "import traceback\n",
    "import time\n",
    "\n",
    "if \"inputLocation\" in os.environ:\n",
    "    input_uri = os.environ[\"inputLocation\"]\n",
    "else:\n",
    "    print(\"inputLocation not found in environment\")\n",
    "    sys.exit(1)\n",
    "if \"outputLocation\" in os.environ:\n",
    "    output_uri = os.environ[\"outputLocation\"]\n",
    "else:\n",
    "    print(\"outputLocation not found in environment\")\n",
    "    sys.exit(1)\n",
    "if \"token\" in os.environ:\n",
    "    token = os.environ[\"token\"]\n",
    "else:\n",
    "    print(\"token not found in environment\")\n",
    "    sys.exit(1)\n",
    "if \"glue_job_name\" in os.environ:\n",
    "    glue_job_name = os.environ[\"glue_job_name\"]\n",
    "else:\n",
    "    print(\"glue_job_name not found in environment\")\n",
    "    sys.exit(1)\n",
    "\n",
    "print(f\"Processing from {input_uri} to {output_uri} using callback token {token}\")\n",
    "sagemaker = boto3.client(\"sagemaker\")\n",
    "glue = boto3.client(\"glue\")\n",
    "\n",
    "poll_interval = 60\n",
    "\n",
    "try:\n",
    "    t1 = time.time()\n",
    "    response = glue.start_job_run(\n",
    "        JobName=glue_job_name, Arguments={\"--output_uri\": output_uri, \"--input_uri\": input_uri}\n",
    "    )\n",
    "    job_run_id = response[\"JobRunId\"]\n",
    "    print(f\"Starting job {job_run_id}\")\n",
    "\n",
    "    job_status = \"STARTING\"\n",
    "    job_error = \"\"\n",
    "    while job_status in [\"STARTING\", \"RUNNING\", \"STOPPING\"]:\n",
    "        time.sleep(poll_interval)\n",
    "        response = glue.get_job_run(\n",
    "            JobName=glue_job_name, RunId=job_run_id, PredecessorsIncluded=False\n",
    "        )\n",
    "        job_status = response[\"JobRun\"][\"JobRunState\"]\n",
    "        if \"ErrorMessage\" in response[\"JobRun\"]:\n",
    "            job_error = response[\"JobRun\"][\"ErrorMessage\"]\n",
    "        print(f\"Job is in state {job_status}\")\n",
    "\n",
    "    t2 = time.time()\n",
    "    total_time = (t2 - t1) / 60.0\n",
    "    if job_status == \"SUCCEEDED\":\n",
    "        print(\"Job succeeded\")\n",
    "        sagemaker.send_pipeline_execution_step_success(\n",
    "            CallbackToken=token,\n",
    "            OutputParameters=[\n",
    "                {\"Name\": \"minutes\", \"Value\": str(total_time)},\n",
    "                {\n",
    "                    \"Name\": \"s3_data_out\",\n",
    "                    \"Value\": str(output_uri),\n",
    "                },\n",
    "            ],\n",
    "        )\n",
    "    else:\n",
    "        print(f\"Job failed: {job_error}\")\n",
    "        sagemaker.send_pipeline_execution_step_failure(CallbackToken=token, FailureReason=job_error)\n",
    "except Exception as e:\n",
    "    trc = traceback.format_exc()\n",
    "    print(f\"Error running ETL job: {str(e)}:\\m {trc}\")\n",
    "    sagemaker.send_pipeline_execution_step_failure(CallbackToken=token, FailureReason=str(e))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e8885798",
   "metadata": {},
   "source": [
    "Next, write the code for your Dockerfile..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "274ee252",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%writefile container/Dockerfile\n",
    "\n",
    "#FROM ubuntu:18.04\n",
    "FROM public.ecr.aws/ubuntu/ubuntu:latest\n",
    "\n",
    "RUN apt-get -y update && apt-get install -y --no-install-recommends \\\n",
    "         python3-pip \\\n",
    "         python3-setuptools \\\n",
    "         curl \\ \n",
    "         unzip\n",
    "\n",
    "RUN /usr/bin/pip3 install boto3\n",
    "\n",
    "RUN curl \"https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip\" -o \"awscliv2.zip\"\n",
    "RUN unzip awscliv2.zip\n",
    "RUN ./aws/install\n",
    "\n",
    "COPY task.py /opt\n",
    "CMD /usr/bin/python3 /opt/task.py"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "710327b7",
   "metadata": {},
   "source": [
    "Finally, use the studio image build CLI to build and push your image to ECR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f66c0197",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%sh\n",
    "\n",
    "cd container\n",
    "\n",
    "sm-docker build . --repository ecs-fargate-task:latest"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "17f0aa2c",
   "metadata": {},
   "source": [
    "After building the image, you have to grab the ECR URI and define a local notebook variable that holds it in the last cell in this section."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "676c7da0",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import sagemaker as sage\n",
    "\n",
    "sess = sage.Session()\n",
    "\n",
    "account = sess.boto_session.client(\"sts\").get_caller_identity()[\"Account\"]\n",
    "region = boto3.session.Session().region_name\n",
    "\n",
    "task_uri = \"{}.dkr.ecr.{}.amazonaws.com/ecs-fargate-task\".format(account, region)\n",
    "print(\"URI:\", task_uri)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cb4cc2ab",
   "metadata": {},
   "source": [
    "### Set up ECS Fargate task\n",
    "\n",
    "Now we will create and register the task using the roles we create above... "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bfcb56c3",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "region = boto3.Session().region_name\n",
    "response = ecs.register_task_definition(\n",
    "    family=\"FargateTaskRunner\",\n",
    "    taskRoleArn=task_role_arn,\n",
    "    executionRoleArn=ecs_role_arn,\n",
    "    networkMode=\"awsvpc\",\n",
    "    containerDefinitions=[\n",
    "        {\n",
    "            \"name\": \"FargateTask\",\n",
    "            \"image\": task_uri,\n",
    "            \"cpu\": 512,\n",
    "            \"memory\": 1024,\n",
    "            \"essential\": True,\n",
    "            \"environment\": [\n",
    "                {\"name\": \"inputLocation\", \"value\": \"temp\"},\n",
    "                {\"name\": \"outputLocation\", \"value\": \"temp\"},\n",
    "            ],\n",
    "            \"logConfiguration\": {\n",
    "                \"logDriver\": \"awslogs\",\n",
    "                \"options\": {\n",
    "                    \"awslogs-create-group\": \"true\",\n",
    "                    \"awslogs-group\": \"glue_sg_pipeline\",\n",
    "                    \"awslogs-region\": region,\n",
    "                    \"awslogs-stream-prefix\": \"task\",\n",
    "                },\n",
    "            },\n",
    "        },\n",
    "    ],\n",
    "    requiresCompatibilities=[\n",
    "        \"FARGATE\",\n",
    "    ],\n",
    "    cpu=\"512\",\n",
    "    memory=\"1024\",\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eda28198",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "print(f\"Task definition ARN: {response['taskDefinition']['taskDefinitionArn']}\")\n",
    "task_arn = response[\"taskDefinition\"][\"taskDefinitionArn\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c068185a",
   "metadata": {},
   "source": [
    "### Copy data to our bucket\n",
    "\n",
    "Next, we'll copy the 2020 taxi data to the sagemaker session default bucket breaking up the data per month."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7e510f97",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "s3 = boto3.client(\"s3\")\n",
    "taxi_bucket = \"nyc-tlc\"\n",
    "taxi_prefix = \"taxi\"\n",
    "\n",
    "# If a default bucket prefix is specified, append it to the s3 path\n",
    "if default_bucket_prefix:\n",
    "    taxi_prefix = f\"{default_bucket_prefix}/{taxi_prefix}\"\n",
    "\n",
    "for month in [\"01\", \"02\", \"03\", \"04\", \"05\", \"06\", \"07\", \"08\", \"09\", \"10\", \"11\", \"12\"]:\n",
    "    copy_source = {\"Bucket\": taxi_bucket, \"Key\": f\"csv_backup/yellow_tripdata_2020-{month}.csv\"}\n",
    "    s3.copy(copy_source, default_bucket, f\"{taxi_prefix}/yellow_tripdata_2020-{month}.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0192d933",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "default_bucket"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "38de53d6",
   "metadata": {},
   "source": [
    "### Create SQS queue for pipeline\n",
    "\n",
    "In this step, we'll create the SQS queue that will be used by the CallbackStep inside SageMaker Pipeline steps.  SageMaker Pipelines will put a token to this queue that will serve as a trigger for your Lambda function which will initiate the Fargate task to process your data.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "75502d6c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "sqs_client = boto3.client(\"sqs\")\n",
    "queue_url = \"\"\n",
    "queue_name = \"pipeline_callbacks_glue_prep\"\n",
    "try:\n",
    "    response = sqs_client.create_queue(QueueName=queue_name)\n",
    "except Exception as e:\n",
    "    print(f\"Failed to create queue:\", e)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4daf9364",
   "metadata": {},
   "source": [
    "Format the queue URL to the same format we will need later on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d4be3a3a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "queue_url = f\"https://sqs.{region}.amazonaws.com/{account}/{queue_name}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4437db85",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "queue_url"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "108e9a87",
   "metadata": {},
   "source": [
    "### VPC and security settings\n",
    "\n",
    "For this setup, we'll use the default VPC and all of its subnets for the fargate task.  However, we'll create a new security group for the tasks that allows egress but no ingress.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "092023b9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ec2 = boto3.client(\"ec2\")\n",
    "response = ec2.describe_vpcs(Filters=[{\"Name\": \"isDefault\", \"Values\": [\"true\"]}])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "03ed560f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "default_vpc_id = response[\"Vpcs\"][0][\"VpcId\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9882488a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "response = ec2.describe_subnets(Filters=[{\"Name\": \"vpc-id\", \"Values\": [default_vpc_id]}])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2ddb728f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "task_subnets = []\n",
    "for r in response[\"Subnets\"]:\n",
    "    task_subnets.append(r[\"SubnetId\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "be02f3dc",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "response = ec2.create_security_group(\n",
    "    Description=\"Security group for Fargate tasks\", GroupName=\"fg_task_sg\", VpcId=default_vpc_id\n",
    ")\n",
    "sg_id = response[\"GroupId\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "88068114",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "response = ec2.authorize_security_group_ingress(\n",
    "    GroupId=sg_id,\n",
    "    IpPermissions=[\n",
    "        {\n",
    "            \"FromPort\": 0,\n",
    "            \"IpProtocol\": \"-1\",\n",
    "            \"UserIdGroupPairs\": [\n",
    "                {\"GroupId\": sg_id, \"Description\": \"local SG ingress\"},\n",
    "            ],\n",
    "            \"ToPort\": 65535,\n",
    "        },\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "927cef06",
   "metadata": {},
   "source": [
    "### Create ETL script\n",
    "\n",
    "The ETL job will take two arguments, the location of the input data in S3 and the output path in S3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1ab6cfbe",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%writefile etl.py\n",
    "import sys\n",
    "from awsglue.transforms import *\n",
    "from awsglue.utils import getResolvedOptions\n",
    "from pyspark.context import SparkContext\n",
    "from awsglue.context import GlueContext\n",
    "from awsglue.job import Job\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.sql import functions as F\n",
    "\n",
    "## @params: [JOB_NAME]\n",
    "args = getResolvedOptions(sys.argv, [\"JOB_NAME\", \"input_uri\", \"output_uri\"])\n",
    "\n",
    "sc = SparkContext()\n",
    "glueContext = GlueContext(sc)\n",
    "spark = glueContext.spark_session\n",
    "job = Job(glueContext)\n",
    "job.init(args[\"JOB_NAME\"], args)\n",
    "\n",
    "df = spark.read.format(\"csv\").option(\"header\", \"true\").load(\"{0}*.csv\".format(args[\"input_uri\"]))\n",
    "df = df.withColumn(\"Passengers\", df[\"passenger_count\"].cast(IntegerType()))\n",
    "df = df.withColumn(\n",
    "    \"pickup_time\",\n",
    "    F.to_timestamp(\n",
    "        F.unix_timestamp(\"tpep_pickup_datetime\", \"yyyy-MM-dd HH:mm:ss\").cast(\"timestamp\")\n",
    "    ),\n",
    ")\n",
    "\n",
    "dfW = df.groupBy(F.window(\"pickup_time\", \"30 minutes\")).agg(F.sum(\"Passengers\").alias(\"passenger\"))\n",
    "dfOut = dfW.drop(\"window\")\n",
    "dfOut.repartition(1).write.option(\"timestampFormat\", \"yyyy-MM-dd HH:mm:ss\").csv(args[\"output_uri\"])\n",
    "\n",
    "job.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "484266df",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "s3.upload_file(\"etl.py\", default_bucket, f\"{default_bucket_prefix_path}pipeline/etl.py\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "31051598",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "glue_script_location = f\"s3://{default_bucket}/{default_bucket_prefix_path}pipeline/etl.py\"\n",
    "glue_script_location"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5aae23a3",
   "metadata": {},
   "source": [
    "### Create ETL job\n",
    "Next, we'll create the glue job using the script and roles creates in the prevous steps..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d85e7d06",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "glue = boto3.client(\"glue\")\n",
    "response = glue.create_job(\n",
    "    Name=\"GlueDataPrepForPipeline\",\n",
    "    Description=\"Prepare data for SageMaker training\",\n",
    "    Role=glue_role_arn,\n",
    "    ExecutionProperty={\"MaxConcurrentRuns\": 1},\n",
    "    Command={\n",
    "        \"Name\": \"glueetl\",\n",
    "        \"ScriptLocation\": glue_script_location,\n",
    "    },\n",
    "    MaxRetries=0,\n",
    "    Timeout=60,\n",
    "    MaxCapacity=10.0,\n",
    "    GlueVersion=\"2.0\",\n",
    ")\n",
    "glue_job_name = response[\"Name\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "59942376",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "glue_job_name"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e98beea3",
   "metadata": {},
   "source": [
    "### Create Lambda function\n",
    "\n",
    "The Lambda function will be triggered on new messages to the SQS queue create by the CallbackStep in SageMaker Pipelines.  The Lambda function is responsible for initiating the run of your Fargate task. Now, write the code that will be used in the Lambda function.."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "35ffc6c8",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%writefile queue_handler.py\n",
    "import json\n",
    "import boto3\n",
    "import os\n",
    "import traceback\n",
    "\n",
    "ecs = boto3.client(\"ecs\")\n",
    "sagemaker = boto3.client(\"sagemaker\")\n",
    "\n",
    "\n",
    "def handler(event, context):\n",
    "    print(f\"Got event: {json.dumps(event)}\")\n",
    "\n",
    "    cluster_arn = os.environ[\"cluster_arn\"]\n",
    "    task_arn = os.environ[\"task_arn\"]\n",
    "    task_subnets = os.environ[\"task_subnets\"]\n",
    "    task_sgs = os.environ[\"task_sgs\"]\n",
    "    glue_job_name = os.environ[\"glue_job_name\"]\n",
    "    print(f\"Cluster ARN: {cluster_arn}\")\n",
    "    print(f\"Task ARN: {task_arn}\")\n",
    "    print(f\"Task Subnets: {task_subnets}\")\n",
    "    print(f\"Task SG: {task_sgs}\")\n",
    "    print(f\"Glue job name: {glue_job_name}\")\n",
    "\n",
    "    for record in event[\"Records\"]:\n",
    "        payload = json.loads(record[\"body\"])\n",
    "        print(f\"Processing record {payload}\")\n",
    "\n",
    "        token = payload[\"token\"]\n",
    "        print(f\"Got token {token}\")\n",
    "\n",
    "        try:\n",
    "            input_data_s3_uri = payload[\"arguments\"][\"input_location\"]\n",
    "            output_data_s3_uri = payload[\"arguments\"][\"output_location\"]\n",
    "            print(f\"Got input_data_s3_uri {input_data_s3_uri}\")\n",
    "            print(f\"Got output_data_s3_uri {output_data_s3_uri}\")\n",
    "\n",
    "            response = ecs.run_task(\n",
    "                cluster=cluster_arn,\n",
    "                count=1,\n",
    "                launchType=\"FARGATE\",\n",
    "                taskDefinition=task_arn,\n",
    "                networkConfiguration={\n",
    "                    \"awsvpcConfiguration\": {\n",
    "                        \"subnets\": task_subnets.split(\",\"),\n",
    "                        \"securityGroups\": task_sgs.split(\",\"),\n",
    "                        \"assignPublicIp\": \"ENABLED\",\n",
    "                    }\n",
    "                },\n",
    "                overrides={\n",
    "                    \"containerOverrides\": [\n",
    "                        {\n",
    "                            \"name\": \"FargateTask\",\n",
    "                            \"environment\": [\n",
    "                                {\"name\": \"inputLocation\", \"value\": input_data_s3_uri},\n",
    "                                {\"name\": \"outputLocation\", \"value\": output_data_s3_uri},\n",
    "                                {\"name\": \"token\", \"value\": token},\n",
    "                                {\"name\": \"glue_job_name\", \"value\": glue_job_name},\n",
    "                            ],\n",
    "                        }\n",
    "                    ]\n",
    "                },\n",
    "            )\n",
    "            if \"failures\" in response and len(response[\"failures\"]) > 0:\n",
    "                f = response[\"failures\"][0]\n",
    "                print(f\"Failed to launch task for token {token}: {f['reason']}\")\n",
    "                sagemaker.send_step_failure(CallbackToken=token, FailureReason=f[\"reason\"])\n",
    "            else:\n",
    "                print(f\"Launched task {response['tasks'][0]['taskArn']}\")\n",
    "        except Exception as e:\n",
    "            trc = traceback.format_exc()\n",
    "            print(f\"Error handling record: {str(e)}:\\m {trc}\")\n",
    "            sagemaker.send_step_failure(CallbackToken=token, FailureReason=e)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "72357868",
   "metadata": {},
   "source": [
    "Finally, bundle the code and upload it to S3 then create the Lambda function..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "394ecaf1",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import zipfile\n",
    "\n",
    "archive = zipfile.ZipFile(\"queue_handler.zip\", \"w\")\n",
    "archive.write(\"queue_handler.py\")\n",
    "\n",
    "s3 = boto3.client(\"s3\")\n",
    "s3.upload_file(\n",
    "    \"queue_handler.zip\", default_bucket, f\"{default_bucket_prefix_path}pipeline/queue_handler.zip\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "25939c7c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "lambda_client = boto3.client(\"lambda\")\n",
    "lambda_client.create_function(\n",
    "    Code={\n",
    "        \"S3Bucket\": default_bucket,\n",
    "        \"S3Key\": f\"{default_bucket_prefix_path}pipeline/queue_handler.zip\",\n",
    "    },\n",
    "    FunctionName=\"SMPipelineQueueHandler\",\n",
    "    Description=\"Process Glue callback messages from SageMaker Pipelines\",\n",
    "    Handler=\"queue_handler.handler\",\n",
    "    Publish=True,\n",
    "    Role=lambda_role_arn,\n",
    "    Runtime=\"python3.7\",\n",
    "    Timeout=20,\n",
    "    MemorySize=128,\n",
    "    PackageType=\"Zip\",\n",
    "    Environment={\n",
    "        \"Variables\": {\n",
    "            \"cluster_arn\": cluster_arn,\n",
    "            \"task_arn\": task_arn,\n",
    "            \"task_subnets\": \",\".join(task_subnets),\n",
    "            \"task_sgs\": sg_id,\n",
    "            \"glue_job_name\": glue_job_name,\n",
    "        }\n",
    "    },\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "88e427cb",
   "metadata": {},
   "source": [
    "### Set up Lambda as SQS target\n",
    "\n",
    "Next, we'll attach the lambda function created above to the SQS queue we previously created.   This ensures that your Lambda will be triggered when new messages are put to your SQS queue. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "068fcce0",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "lambda_client.create_event_source_mapping(\n",
    "    EventSourceArn=f\"arn:aws:sqs:{region}:{account}:{queue_name}\",\n",
    "    FunctionName=\"SMPipelineQueueHandler\",\n",
    "    Enabled=True,\n",
    "    BatchSize=10,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a671ba5f",
   "metadata": {},
   "source": [
    "## Build & Execute SageMaker Pipeline\n",
    "\n",
    "Now that all of the components are created and configured that support the tasks within your pipeline steps, we're now ready to bring it all together and setup the pipeline.  \n",
    "\n",
    "First, install the SageMaker Python SDK."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bcf1d0ab",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "!pip install \"sagemaker>=2.99.0\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6fc56f1a",
   "metadata": {},
   "source": [
    "### Pipeline Initialization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4442658f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "timestamp = int(time.time())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ebbdbb15",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.parameters import (\n",
    "    ParameterInteger,\n",
    "    ParameterString,\n",
    ")\n",
    "\n",
    "input_data = ParameterString(\n",
    "    name=\"InputData\", default_value=f\"s3://{default_bucket}/{taxi_prefix}/\"\n",
    ")\n",
    "id_out = ParameterString(name=\"IdOut\", default_value=\"taxiout\" + str(timestamp))\n",
    "output_data = ParameterString(\n",
    "    name=\"OutputData\", default_value=f\"s3://{default_bucket}/{taxi_prefix}_output/\"\n",
    ")\n",
    "training_instance_count = ParameterInteger(name=\"TrainingInstanceCount\", default_value=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a6d54730",
   "metadata": {},
   "source": [
    "### Pipeline Steps"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18bcbab9",
   "metadata": {},
   "source": [
    "#### 1 - Call Back Step \n",
    "\n",
    "First, we'll configure the callback step.  \n",
    "\n",
    "The callback step will accept the following **inputs**: \n",
    "  * S3 location of our raw taxi data\n",
    "  * SQS queue \n",
    "  \n",
    "The callback step will return the following **outputs**:\n",
    "  * S3 location of processed data to be used for model training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6b952d28",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.callback_step import CallbackStep, CallbackOutput, CallbackOutputTypeEnum\n",
    "from sagemaker.workflow.functions import Join\n",
    "\n",
    "callback1_output = CallbackOutput(\n",
    "    output_name=\"s3_data_out\", output_type=CallbackOutputTypeEnum.String\n",
    ")\n",
    "\n",
    "step_callback_data = CallbackStep(\n",
    "    name=\"GluePrepCallbackStep\",\n",
    "    sqs_queue_url=queue_url,\n",
    "    inputs={\n",
    "        \"input_location\": f\"s3://{default_bucket}/{taxi_prefix}/\",\n",
    "        \"output_location\": Join(\n",
    "            on=\"/\", values=[\"s3:/\", default_bucket, f\"{taxi_prefix}_output\", id_out]\n",
    "        ),\n",
    "    },\n",
    "    outputs=[callback1_output],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9d28c619",
   "metadata": {},
   "source": [
    "#### 2 - Training Step \n",
    "\n",
    "Next, we'll configure the training step by first configuring the estimator for random cut forest. Then, we use the output of the estimator's .fit() method as arguments to the TrainingStep. By passing the pipeline_session to the sagemaker_session, calling .fit() does not launch the training job. Instead, it returns the arguments needed to run the job as a step in the pipeline.\n",
    "\n",
    "To generate the step arguments for the training step, it will accept the following **inputs**: \n",
    "  * S3 location of processed data to be used for model training\n",
    "  * ECR containing the training image for rcf\n",
    "  * Estimator configuration\n",
    "  \n",
    "The training step will return the following **outputs**:\n",
    "  * S3 location of the trained model artifact"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a846a735",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.pipeline_context import PipelineSession\n",
    "\n",
    "containers = {\n",
    "    \"us-west-2\": \"174872318107.dkr.ecr.us-west-2.amazonaws.com/randomcutforest:latest\",\n",
    "    \"us-east-1\": \"382416733822.dkr.ecr.us-east-1.amazonaws.com/randomcutforest:latest\",\n",
    "    \"us-east-2\": \"404615174143.dkr.ecr.us-east-2.amazonaws.com/randomcutforest:latest\",\n",
    "    \"eu-west-1\": \"438346466558.dkr.ecr.eu-west-1.amazonaws.com/randomcutforest:latest\",\n",
    "}\n",
    "region_name = boto3.Session().region_name\n",
    "container = containers[region_name]\n",
    "model_prefix = \"model\"\n",
    "\n",
    "# If a default bucket prefix is specified, append it to the s3 path\n",
    "if default_bucket_prefix:\n",
    "    model_prefix = f\"{default_bucket_prefix}/{model_prefix}\"\n",
    "\n",
    "pipeline_session = PipelineSession()\n",
    "\n",
    "rcf = sagemaker.estimator.Estimator(\n",
    "    container,\n",
    "    sagemaker.get_execution_role(),\n",
    "    output_path=\"s3://{}/{}/output\".format(default_bucket, model_prefix),\n",
    "    instance_count=training_instance_count,\n",
    "    instance_type=\"ml.c5.xlarge\",\n",
    "    sagemaker_session=pipeline_session,\n",
    ")\n",
    "\n",
    "rcf.set_hyperparameters(num_samples_per_tree=200, num_trees=50, feature_dim=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9f0e02e5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.inputs import TrainingInput\n",
    "from sagemaker.workflow.steps import TrainingStep\n",
    "\n",
    "train_step_args = rcf.fit(\n",
    "    inputs={\n",
    "        \"train\": TrainingInput(\n",
    "            # s3_data = Output of the previous call back step\n",
    "            s3_data=step_callback_data.properties.Outputs[\"s3_data_out\"],\n",
    "            content_type=\"text/csv;label_size=0\",\n",
    "            distribution=\"ShardedByS3Key\",\n",
    "        ),\n",
    "    },\n",
    ")\n",
    "step_train = TrainingStep(\n",
    "    name=\"TrainModel\",\n",
    "    step_args=train_step_args,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2d99418f",
   "metadata": {},
   "source": [
    "#### 3 - Create Model\n",
    "\n",
    "Next, we'll package the trained model for deployment. To achieve this, we define the ModelStep by providing the return values from `model.create()` as the step arguments. Similarly, the `pipeline_session` is required when defining the model, which puts off the model creation to the pipeline execution time.\n",
    "\n",
    "To generate the step arguments for the model step, it will accept the following **inputs**: \n",
    "  * S3 location of the trained model artifact\n",
    "  * ECR containing the inference image for rcf\n",
    "  \n",
    "The create model step will return the following **outputs**: \n",
    "  * SageMaker packaged model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "56103885",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.model import Model\n",
    "from sagemaker import get_execution_role\n",
    "\n",
    "role = get_execution_role()\n",
    "\n",
    "image_uri = sagemaker.image_uris.retrieve(\"randomcutforest\", region)\n",
    "\n",
    "model = Model(\n",
    "    image_uri=image_uri,\n",
    "    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,\n",
    "    sagemaker_session=pipeline_session,\n",
    "    role=role,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b52fb01a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.model_step import ModelStep\n",
    "\n",
    "\n",
    "model_step_args = model.create(\n",
    "    instance_type=\"ml.m5.large\",\n",
    ")\n",
    "\n",
    "create_model = ModelStep(name=\"TaxiModel\", step_args=model_step_args)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6d29e348",
   "metadata": {},
   "source": [
    "#### 4 - Batch Transform\n",
    "\n",
    "Next, we'll deploy the model using batch transform then do a quick evaluation with our data to compute anomaly scores for each of our data points on input.\n",
    "\n",
    "To generate the step arguments for the batch transform step, it will accept the following **inputs**: \n",
    "  * SageMaker packaged model\n",
    "  * S3 location of the input data\n",
    "  * ECR containing the inference image for rcf\n",
    "  \n",
    "The batch transform step will return the following **outputs**: \n",
    "  * S3 location of the output data (anomaly scores)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3611a03d",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "base_uri = step_callback_data.properties.Outputs[\"s3_data_out\"]\n",
    "output_prefix = \"batch-out\"\n",
    "\n",
    "# If a default bucket prefix is specified, append it to the s3 path\n",
    "if default_bucket_prefix:\n",
    "    output_prefix = f\"{default_bucket_prefix}/{output_prefix}\"\n",
    "\n",
    "from sagemaker.transformer import Transformer\n",
    "\n",
    "transformer = Transformer(\n",
    "    model_name=create_model.properties.ModelName,\n",
    "    instance_type=\"ml.m5.xlarge\",\n",
    "    assemble_with=\"Line\",\n",
    "    accept=\"text/csv\",\n",
    "    instance_count=1,\n",
    "    output_path=f\"s3://{default_bucket}/{output_prefix}/\",\n",
    "    sagemaker_session=pipeline_session,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3a3efefe",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.inputs import TransformInput\n",
    "from sagemaker.workflow.steps import TransformStep\n",
    "\n",
    "batch_data = step_callback_data.properties.Outputs[\"s3_data_out\"]\n",
    "\n",
    "transform_step_args = transformer.transform(\n",
    "    data=batch_data,\n",
    "    content_type=\"text/csv\",\n",
    "    split_type=\"Line\",\n",
    "    input_filter=\"$[0]\",\n",
    "    join_source=\"Input\",\n",
    "    output_filter=\"$[0,-1]\",\n",
    ")\n",
    "\n",
    "step_transform = TransformStep(\n",
    "    name=\"TaxiTransform\",\n",
    "    step_args=transform_step_args,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8d8e29b0",
   "metadata": {},
   "source": [
    "### Configure Pipeline Using Created Steps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "54b59128",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.pipeline import Pipeline\n",
    "from sagemaker.utils import unique_name_from_base\n",
    "\n",
    "pipeline_name = unique_name_from_base(\"GluePipeline\")\n",
    "pipeline = Pipeline(\n",
    "    name=pipeline_name,\n",
    "    parameters=[\n",
    "        input_data,\n",
    "        training_instance_count,\n",
    "        id_out,\n",
    "    ],\n",
    "    steps=[step_callback_data, step_train, create_model, step_transform],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bfaa63e5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker import get_execution_role\n",
    "\n",
    "pipeline.upsert(role_arn=get_execution_role())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ec3a8744",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "definition = json.loads(pipeline.definition())\n",
    "definition"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d497be66",
   "metadata": {},
   "source": [
    "### Execute Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "87383885",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "execution = pipeline.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cad74b43",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "execution.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98238b16",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "execution.list_steps()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fa479974",
   "metadata": {},
   "source": [
    "You can continue to check the status of your pipeline to sure it completes successfuly.  When your pipeline completes, you will see all steps in a completed status.  You can also check the status of your pipeline and each pipeline step directy in the Studio console under *Components and registries* --> Pipelines as shown below:"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6a55f1bf",
   "metadata": {},
   "source": [
    "![CustomPipeline](./images/Custom-Pipeline-Studio.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fd9c1a3d",
   "metadata": {},
   "source": [
    "## Notebook CI Test Results\n",
    "\n",
    "This notebook was tested in multiple regions. The test results are as follows, except for us-west-2 which is shown at the top of the notebook.\n",
    "\n",
    "![This us-east-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-east-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This us-east-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-east-2/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This us-west-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/us-west-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This ca-central-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ca-central-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This sa-east-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/sa-east-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This eu-west-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This eu-west-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-2/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This eu-west-3 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-west-3/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This eu-central-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-central-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This eu-north-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/eu-north-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This ap-southeast-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-southeast-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This ap-southeast-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-southeast-2/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This ap-northeast-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-northeast-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This ap-northeast-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-northeast-2/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n",
    "\n",
    "![This ap-south-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://prod.us-west-2.tcx-beacon.docs.aws.dev/sagemaker-nb/ap-south-1/ml_ops|sm-pipelines_callback_step|sm-pipelines_callback_step.ipynb)\n"
   ]
  }
 ],
 "metadata": {
  "availableInstances": [
   {
    "_defaultOrder": 0,
    "_isFastLaunch": true,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 4,
    "name": "ml.t3.medium",
    "vcpuNum": 2
   },
   {
    "_defaultOrder": 1,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 8,
    "name": "ml.t3.large",
    "vcpuNum": 2
   },
   {
    "_defaultOrder": 2,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 16,
    "name": "ml.t3.xlarge",
    "vcpuNum": 4
   },
   {
    "_defaultOrder": 3,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 32,
    "name": "ml.t3.2xlarge",
    "vcpuNum": 8
   },
   {
    "_defaultOrder": 4,
    "_isFastLaunch": true,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 8,
    "name": "ml.m5.large",
    "vcpuNum": 2
   },
   {
    "_defaultOrder": 5,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 16,
    "name": "ml.m5.xlarge",
    "vcpuNum": 4
   },
   {
    "_defaultOrder": 6,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 32,
    "name": "ml.m5.2xlarge",
    "vcpuNum": 8
   },
   {
    "_defaultOrder": 7,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 64,
    "name": "ml.m5.4xlarge",
    "vcpuNum": 16
   },
   {
    "_defaultOrder": 8,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 128,
    "name": "ml.m5.8xlarge",
    "vcpuNum": 32
   },
   {
    "_defaultOrder": 9,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 192,
    "name": "ml.m5.12xlarge",
    "vcpuNum": 48
   },
   {
    "_defaultOrder": 10,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 256,
    "name": "ml.m5.16xlarge",
    "vcpuNum": 64
   },
   {
    "_defaultOrder": 11,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 384,
    "name": "ml.m5.24xlarge",
    "vcpuNum": 96
   },
   {
    "_defaultOrder": 12,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 8,
    "name": "ml.m5d.large",
    "vcpuNum": 2
   },
   {
    "_defaultOrder": 13,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 16,
    "name": "ml.m5d.xlarge",
    "vcpuNum": 4
   },
   {
    "_defaultOrder": 14,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 32,
    "name": "ml.m5d.2xlarge",
    "vcpuNum": 8
   },
   {
    "_defaultOrder": 15,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 64,
    "name": "ml.m5d.4xlarge",
    "vcpuNum": 16
   },
   {
    "_defaultOrder": 16,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 128,
    "name": "ml.m5d.8xlarge",
    "vcpuNum": 32
   },
   {
    "_defaultOrder": 17,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 192,
    "name": "ml.m5d.12xlarge",
    "vcpuNum": 48
   },
   {
    "_defaultOrder": 18,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 256,
    "name": "ml.m5d.16xlarge",
    "vcpuNum": 64
   },
   {
    "_defaultOrder": 19,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 384,
    "name": "ml.m5d.24xlarge",
    "vcpuNum": 96
   },
   {
    "_defaultOrder": 20,
    "_isFastLaunch": false,
    "category": "General purpose",
    "gpuNum": 0,
    "hideHardwareSpecs": true,
    "memoryGiB": 0,
    "name": "ml.geospatial.interactive",
    "supportedImageNames": [
     "sagemaker-geospatial-v1-0"
    ],
    "vcpuNum": 0
   },
   {
    "_defaultOrder": 21,
    "_isFastLaunch": true,
    "category": "Compute optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 4,
    "name": "ml.c5.large",
    "vcpuNum": 2
   },
   {
    "_defaultOrder": 22,
    "_isFastLaunch": false,
    "category": "Compute optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 8,
    "name": "ml.c5.xlarge",
    "vcpuNum": 4
   },
   {
    "_defaultOrder": 23,
    "_isFastLaunch": false,
    "category": "Compute optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 16,
    "name": "ml.c5.2xlarge",
    "vcpuNum": 8
   },
   {
    "_defaultOrder": 24,
    "_isFastLaunch": false,
    "category": "Compute optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 32,
    "name": "ml.c5.4xlarge",
    "vcpuNum": 16
   },
   {
    "_defaultOrder": 25,
    "_isFastLaunch": false,
    "category": "Compute optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 72,
    "name": "ml.c5.9xlarge",
    "vcpuNum": 36
   },
   {
    "_defaultOrder": 26,
    "_isFastLaunch": false,
    "category": "Compute optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 96,
    "name": "ml.c5.12xlarge",
    "vcpuNum": 48
   },
   {
    "_defaultOrder": 27,
    "_isFastLaunch": false,
    "category": "Compute optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 144,
    "name": "ml.c5.18xlarge",
    "vcpuNum": 72
   },
   {
    "_defaultOrder": 28,
    "_isFastLaunch": false,
    "category": "Compute optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 192,
    "name": "ml.c5.24xlarge",
    "vcpuNum": 96
   },
   {
    "_defaultOrder": 29,
    "_isFastLaunch": true,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 16,
    "name": "ml.g4dn.xlarge",
    "vcpuNum": 4
   },
   {
    "_defaultOrder": 30,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 32,
    "name": "ml.g4dn.2xlarge",
    "vcpuNum": 8
   },
   {
    "_defaultOrder": 31,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 64,
    "name": "ml.g4dn.4xlarge",
    "vcpuNum": 16
   },
   {
    "_defaultOrder": 32,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 128,
    "name": "ml.g4dn.8xlarge",
    "vcpuNum": 32
   },
   {
    "_defaultOrder": 33,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 4,
    "hideHardwareSpecs": false,
    "memoryGiB": 192,
    "name": "ml.g4dn.12xlarge",
    "vcpuNum": 48
   },
   {
    "_defaultOrder": 34,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 256,
    "name": "ml.g4dn.16xlarge",
    "vcpuNum": 64
   },
   {
    "_defaultOrder": 35,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 61,
    "name": "ml.p3.2xlarge",
    "vcpuNum": 8
   },
   {
    "_defaultOrder": 36,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 4,
    "hideHardwareSpecs": false,
    "memoryGiB": 244,
    "name": "ml.p3.8xlarge",
    "vcpuNum": 32
   },
   {
    "_defaultOrder": 37,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 8,
    "hideHardwareSpecs": false,
    "memoryGiB": 488,
    "name": "ml.p3.16xlarge",
    "vcpuNum": 64
   },
   {
    "_defaultOrder": 38,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 8,
    "hideHardwareSpecs": false,
    "memoryGiB": 768,
    "name": "ml.p3dn.24xlarge",
    "vcpuNum": 96
   },
   {
    "_defaultOrder": 39,
    "_isFastLaunch": false,
    "category": "Memory Optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 16,
    "name": "ml.r5.large",
    "vcpuNum": 2
   },
   {
    "_defaultOrder": 40,
    "_isFastLaunch": false,
    "category": "Memory Optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 32,
    "name": "ml.r5.xlarge",
    "vcpuNum": 4
   },
   {
    "_defaultOrder": 41,
    "_isFastLaunch": false,
    "category": "Memory Optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 64,
    "name": "ml.r5.2xlarge",
    "vcpuNum": 8
   },
   {
    "_defaultOrder": 42,
    "_isFastLaunch": false,
    "category": "Memory Optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 128,
    "name": "ml.r5.4xlarge",
    "vcpuNum": 16
   },
   {
    "_defaultOrder": 43,
    "_isFastLaunch": false,
    "category": "Memory Optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 256,
    "name": "ml.r5.8xlarge",
    "vcpuNum": 32
   },
   {
    "_defaultOrder": 44,
    "_isFastLaunch": false,
    "category": "Memory Optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 384,
    "name": "ml.r5.12xlarge",
    "vcpuNum": 48
   },
   {
    "_defaultOrder": 45,
    "_isFastLaunch": false,
    "category": "Memory Optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 512,
    "name": "ml.r5.16xlarge",
    "vcpuNum": 64
   },
   {
    "_defaultOrder": 46,
    "_isFastLaunch": false,
    "category": "Memory Optimized",
    "gpuNum": 0,
    "hideHardwareSpecs": false,
    "memoryGiB": 768,
    "name": "ml.r5.24xlarge",
    "vcpuNum": 96
   },
   {
    "_defaultOrder": 47,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 16,
    "name": "ml.g5.xlarge",
    "vcpuNum": 4
   },
   {
    "_defaultOrder": 48,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 32,
    "name": "ml.g5.2xlarge",
    "vcpuNum": 8
   },
   {
    "_defaultOrder": 49,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 64,
    "name": "ml.g5.4xlarge",
    "vcpuNum": 16
   },
   {
    "_defaultOrder": 50,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 128,
    "name": "ml.g5.8xlarge",
    "vcpuNum": 32
   },
   {
    "_defaultOrder": 51,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 1,
    "hideHardwareSpecs": false,
    "memoryGiB": 256,
    "name": "ml.g5.16xlarge",
    "vcpuNum": 64
   },
   {
    "_defaultOrder": 52,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 4,
    "hideHardwareSpecs": false,
    "memoryGiB": 192,
    "name": "ml.g5.12xlarge",
    "vcpuNum": 48
   },
   {
    "_defaultOrder": 53,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 4,
    "hideHardwareSpecs": false,
    "memoryGiB": 384,
    "name": "ml.g5.24xlarge",
    "vcpuNum": 96
   },
   {
    "_defaultOrder": 54,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 8,
    "hideHardwareSpecs": false,
    "memoryGiB": 768,
    "name": "ml.g5.48xlarge",
    "vcpuNum": 192
   },
   {
    "_defaultOrder": 55,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 8,
    "hideHardwareSpecs": false,
    "memoryGiB": 1152,
    "name": "ml.p4d.24xlarge",
    "vcpuNum": 96
   },
   {
    "_defaultOrder": 56,
    "_isFastLaunch": false,
    "category": "Accelerated computing",
    "gpuNum": 8,
    "hideHardwareSpecs": false,
    "memoryGiB": 1152,
    "name": "ml.p4de.24xlarge",
    "vcpuNum": 96
   }
  ],
  "instance_type": "ml.m5.large",
  "kernelspec": {
   "display_name": "Python 3 (Data Science 3.0)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-west-2:236514542706:image/sagemaker-data-science-310-v1"
  },
  "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.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}