4-afd-pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Part 4: Create an End to End Pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-warning\"> <h4><strong>🛑 PRE-REQUISITE</strong></h4>\n",
    "In order to be able to execute this notebook, you must first execute the first three notebooks included in this project\n",
    "    <ul>\n",
    "        <li><a href=\"./1-data-analysis-prep.ipynb\">1-data-analysis-prep.ipynb</a></li>\n",
    "        <li><a href=\"./2-afd-model-setup.ipynb\">2-afd-model-setup.ipynb</a></li>\n",
    "        <li><a href=\"./3-afd-model-train-deploy.ipynb\">3-afd-model-train-deploy.ipynb</a></li>\n",
    "        <li><a href=\"./4-0-custom-container.ipynb\">4-0-custom-container.ipynb</a></li>\n",
    "    </ul>\n",
    "    Also ensure that you have the latest version of SageMaker Python SDK before proceeding, by running the code cell below. Once Sagemaker SDK is updated, please restart the kernel using menu \"Kernel\">\"Restart Kernel\".\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install --upgrade sagemaker"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Overview <a id='overview'></a>\n",
    "\n",
    "* [Notebook 1: Data Preparation, Process, and Store Features](./1-data-analysis-prep.ipynb)\n",
    "* [Notebook 2: Amazon Fraud Detector Model Setup](./2-afd-model-setup.ipynb)\n",
    "* [Notebook 3: Model training, deployment, real-time and batch inference](./3-afd-model-train-deploy.ipynb)\n",
    "* **[Notebook 4: Create an end-to-end pipeline](./4-afd-pipeline.ipynb)**\n",
    "    * [Introduction](#intro)\n",
    "    * [Setup notebook](#setup)\n",
    "    * [Setup Pipeline Parameters & Steps](#pipeline)\n",
    "        * **Step 1:** [Signup attempts Data Wrangler Preprocessing Step](#step1)\n",
    "        * **Step 2:** [Outcomes Data Wrangler Preprocessing Step](#step2)\n",
    "        * **Step 3:** [Create Training Data Set Step](#step3)\n",
    "        * **Step 4:** [Train Amazon Fraud Detector Model Step](#step4)\n",
    "        * **Step 5:** [Check AUC Metric (Area Under the ROC Curve) Condition](#step5)\n",
    "        * **Step 6:** [Activate Amazon Fraud Detector Model Step](#step6)        \n",
    "        * **Step 7:** [Setup Amazon Fraud Detector Model detector Step](#step7)\n",
    "    * [Combine the Pipeline Steps and Run](#define-pipeline)\n",
    "    * [Delete Pipeline and Cleanup (Optional)](#delete)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Introduction <a id=\"intro\"></a>\n",
    "___\n",
    "<a href=\"#overview\">overview</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook, we will build a [SageMaker Pipeline](https://docs.aws.amazon.com/sagemaker/latest/dg/pipelines-sdk.html) that automates the entire end to end process. Recall that we initially did all the steps in a manual way, and experimented as a data scientist: testing each segment, hands on, and determine for example, which transformations should be applied to the features, which features should be added to the training data file etc.  Now we will automate these steps, and perhaps pass on the responsibility to an ML Engineer or MLOps role.\n",
    "\n",
    "<img src=\"images/nb4.png\" width=\"800\" height=\"800\"/>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Setup <a id=\"setup\"></a>\n",
    "----\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "As part of setup, we will import necessary libraries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import boto3\n",
    "import pathlib\n",
    "import sagemaker\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import awswrangler as wr\n",
    "\n",
    "from IPython.core.display import display, HTML\n",
    "from IPython.display import clear_output, JSON\n",
    "\n",
    "from sagemaker.workflow.pipeline import Pipeline\n",
    "from sagemaker.workflow.steps import CreateModelStep\n",
    "from sagemaker.sklearn.processing import SKLearnProcessor, ScriptProcessor\n",
    "from sagemaker.workflow.step_collections import RegisterModel\n",
    "from sagemaker.workflow.steps import ProcessingStep, TrainingStep\n",
    "from sagemaker.workflow.parameters import ParameterInteger, ParameterFloat, ParameterString\n",
    "from sagemaker.workflow.properties import PropertyFile"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1 Set region and boto3 config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Using AWS Region: us-east-2\n"
     ]
    }
   ],
   "source": [
    "#You can change this to a region of your choice\n",
    "import sagemaker\n",
    "region = sagemaker.Session().boto_region_name\n",
    "print(\"Using AWS Region: {}\".format(region))\n",
    "\n",
    "boto3.setup_default_session(region_name=region)\n",
    "boto_session = boto3.Session(region_name=region)\n",
    "\n",
    "s3_client = boto3.client('s3', region_name=region)\n",
    "\n",
    "sagemaker_boto_client = boto_session.client('sagemaker')\n",
    "sagemaker_session = sagemaker.session.Session(\n",
    "    boto_session=boto_session,\n",
    "    sagemaker_client=sagemaker_boto_client)\n",
    "sagemaker_role = sagemaker.get_execution_role()\n",
    "\n",
    "account_id = boto3.client('sts').get_caller_identity()[\"Account\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.2 Initialize variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Lets pull some of the variables from cache\n",
    "%store -r MODEL_NAME\n",
    "%store -r DETECTOR_NAME\n",
    "%store -r S3_FILE_LOC\n",
    "%store -r DATA_ACCESS_ROLE_ARN\n",
    "\n",
    "%store -r signups_fg_name \n",
    "%store -r outcomes_fg_name\n",
    "%store -r signup_attempts_table\n",
    "%store -r signup_outcomes_table\n",
    "%store -r afd_database_name\n",
    "%store -r afd_bucket\n",
    "%store -r afd_prefix\n",
    "\n",
    "processing_dir = \"/opt/ml/processing\"\n",
    "create_dataset_script_uri = f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/code/create_dataset.py'\n",
    "train_model_script_uri = f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/code/train_afd.py'\n",
    "activate_model_script_uri = f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/code/activate_afd.py'\n",
    "setup_detector_script_uri = f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/code/setup_detector.py'\n",
    "\n",
    "#======> variables used for parameterizing the notebook run\n",
    "flow_instance_count = 1\n",
    "flow_instance_type = \"ml.m5.4xlarge\"\n",
    "\n",
    "train_instance_count = 1\n",
    "train_instance_type = \"ml.t3.medium\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Pipeline Parameters <a id=\"#pipeline\"></a>\n",
    "---\n",
    "\n",
    "An important feature of SageMaker Pipelines is the ability to define the steps ahead of time, but be able to change the parameters to those steps at execution time without having to re-define the pipeline. This can be achieved by using ParameterInteger, ParameterFloat or ParameterString to define a value upfront which can be modified when you call `pipeline.start(parameters=parameters)` later. Only certain parameters can be defined this way. Check out the [documentation](https://docs.aws.amazon.com/sagemaker/latest/dg/build-and-manage-parameters.html) to learn more about Pipeline Parameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "model_name_param = ParameterString(\n",
    "    name=\"AFDModelName\",\n",
    "    default_value=MODEL_NAME,\n",
    ")\n",
    "\n",
    "detector_name_param = ParameterString(\n",
    "    name=\"AFDDetectorName\",\n",
    "    default_value=DETECTOR_NAME,\n",
    ")\n",
    "\n",
    "data_role_param = ParameterString(\n",
    "    name=\"DataAccessRoleARN\",\n",
    "    default_value=DATA_ACCESS_ROLE_ARN\n",
    ")\n",
    "\n",
    "data_path_param = ParameterString(\n",
    "    name=\"TrainDataS3Path\",\n",
    "    default_value=f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/train-data'\n",
    ")\n",
    "\n",
    "auc_threshold_param = ParameterFloat(\n",
    "    name=\"AFDAUCThreshold\",\n",
    "    default_value=0.75\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 1: Signup attempts Data Wrangler Preprocessing Step <a id='step1'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "Recall that in the first notebook we processed our raw `signup_attempts.csv` file using the `signup_attempts.flow` file. The flow file is a SageMaker data wrangler construct using which we defined all the necessary transformations. Once the flow file was ready we executed it (By dynamically generating a Jupyter notebook using the \"Export to S3\" option in the flow) to generate our `signup_attempts_preprocessed.csv` file. In this step, we are going to use the same flow file, but instead of manually running it, we will setup so that SageMaker pipeline does the job for us."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Upload flow to S3\n",
    "This will become an input to the first step and, as such, the flow file needs to be in S3. You may use the same S3 location for the flow file from the first Notebook, however, it is recommended that you store Pipeline specific artifacts under a separate bucket or prefix."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "signup_attempts flow file uploaded to S3\n"
     ]
    }
   ],
   "source": [
    "s3_client.upload_file(Filename='signup_attempts.flow', Bucket=afd_bucket, Key=f'{afd_prefix}/afd-pipeline/dataprep-notebooks/signup_attempts.flow')\n",
    "signups_flow_uri = f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/dataprep-notebooks/signup_attempts.flow'\n",
    "print(f\"signup_attempts flow file uploaded to S3\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define the first Data Wrangler step's inputs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('signup_attempts.flow', 'r') as f:\n",
    "    signups_flow = json.load(f)\n",
    "\n",
    "flow_step_inputs_1 = []\n",
    "\n",
    "# flow file contains the code for each transformation\n",
    "flow_file_input = sagemaker.processing.ProcessingInput(\n",
    "    source=signups_flow_uri,            \n",
    "    destination=f\"{processing_dir}/flow\", \n",
    "    input_name='flow')\n",
    "\n",
    "flow_step_inputs_1.append(flow_file_input)\n",
    "\n",
    "# parse the flow file for S3 inputs to Data Wranger job\n",
    "for node in signups_flow[\"nodes\"]:\n",
    "    if \"dataset_definition\" in node[\"parameters\"]:\n",
    "        data_def = node[\"parameters\"][\"dataset_definition\"]\n",
    "        name = data_def[\"name\"]\n",
    "        s3_input = sagemaker.processing.ProcessingInput(\n",
    "            source=data_def[\"s3ExecutionContext\"][\"s3Uri\"], \n",
    "            destination=f'{processing_dir}/{name}', \n",
    "            input_name=name)\n",
    "        flow_step_inputs_1.append(s3_input)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define outputs for first Data Wranger step"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "signups_output_name = f\"{signups_flow['nodes'][-1]['node_id']}.{signups_flow['nodes'][-1]['outputs'][0]['name']}\"\n",
    "\n",
    "flow_step_outputs_1 = []\n",
    "\n",
    "flow_output = sagemaker.processing.ProcessingOutput(\n",
    "    output_name=signups_output_name,\n",
    "    feature_store_output=sagemaker.processing.FeatureStoreOutput(\n",
    "        feature_group_name=signups_fg_name), \n",
    "    app_managed=True)\n",
    "\n",
    "flow_step_outputs_1.append(flow_output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 2: Outcomes Data Wrangler Preprocessing Step <a id='step2'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "We will repeat the same process for processing the `signup_outcomes.csv` file using the `signup_outcomes.flow` file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Customers flow file uploaded to S3\n"
     ]
    }
   ],
   "source": [
    "s3_client.upload_file(Filename='signup_outcomes.flow', Bucket=afd_bucket, Key=f'{afd_prefix}/afd-pipeline/dataprep-notebooks/signup_outcomes.flow')\n",
    "outcomes_flow_uri = f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/dataprep-notebooks/signup_outcomes.flow'\n",
    "print(f\"Customers flow file uploaded to S3\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "with open('signup_outcomes.flow', 'r') as f:\n",
    "    outcomes_flow = json.load(f)\n",
    "    \n",
    "flow_step_inputs_2 = []\n",
    "\n",
    "# flow file contains the code for each transformation\n",
    "flow_file_input = sagemaker.processing.ProcessingInput(\n",
    "    source=outcomes_flow_uri,            \n",
    "    destination=f\"{processing_dir}/flow\", \n",
    "    input_name='flow')\n",
    "\n",
    "flow_step_inputs_2.append(flow_file_input)\n",
    "\n",
    "# parse the flow file for S3 inputs to Data Wranger job\n",
    "for node in outcomes_flow[\"nodes\"]:\n",
    "    if \"dataset_definition\" in node[\"parameters\"]:\n",
    "        data_def = node[\"parameters\"][\"dataset_definition\"]\n",
    "        name = data_def[\"name\"]\n",
    "        s3_input = sagemaker.processing.ProcessingInput(\n",
    "            source=data_def[\"s3ExecutionContext\"][\"s3Uri\"], \n",
    "            destination=f'{processing_dir}/{name}', \n",
    "            input_name=name)\n",
    "        flow_step_inputs_2.append(s3_input)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "outcomes_output_name = f\"{outcomes_flow['nodes'][-1]['node_id']}.{outcomes_flow['nodes'][-1]['outputs'][0]['name']}\"\n",
    "\n",
    "flow_step_outputs_2 = []\n",
    "\n",
    "flow_output = sagemaker.processing.ProcessingOutput(\n",
    "    output_name=outcomes_output_name,\n",
    "    feature_store_output=sagemaker.processing.FeatureStoreOutput(\n",
    "        feature_group_name=outcomes_fg_name), \n",
    "    app_managed=True)\n",
    "\n",
    "flow_step_outputs_2.append(flow_output)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Finally, Define processor and processing steps for Signups and Outcomes flows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# You can find the proper image uri by exporting your Data Wrangler flow to a pipeline notebook\n",
    "# =================================\n",
    "image_uri = \"415577184552.dkr.ecr.us-east-2.amazonaws.com/sagemaker-data-wrangler-container:1.0.2\"\n",
    "\n",
    "flow_processor = sagemaker.processing.Processor(\n",
    "    role=sagemaker_role, \n",
    "    image_uri=image_uri, \n",
    "    instance_count=flow_instance_count, \n",
    "    instance_type=flow_instance_type, \n",
    "    max_runtime_in_seconds=86400)\n",
    "\n",
    "# Signups data flow step\n",
    "signups_flow_step = ProcessingStep(\n",
    "    name='Step1SignupsDataWranglerProcessing', \n",
    "    processor=flow_processor, \n",
    "    inputs=flow_step_inputs_1, \n",
    "    outputs=flow_step_outputs_1)\n",
    "\n",
    "# Outcomes data flow step\n",
    "outcomes_flow_step = ProcessingStep(\n",
    "    name='Step2OutcomesDataWranglerProcessing', \n",
    "    processor=flow_processor, \n",
    "    inputs=flow_step_inputs_2, \n",
    "    outputs=flow_step_outputs_2,\n",
    "    depends_on=['Step1SignupsDataWranglerProcessing'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3: Create Training Data Set Step <a id='step3'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "In this step you will query the Feature Store offline datastore to generate the training dataset. Recall that in the first notebook you created two feature groups for the signups and outcomes raw data set using the pre-processed files. We will, lookup the details of those feature groups, such as the Athena database name, table names, column names etc. and will construct an Athena query to generate our final training dataset. This is done using the rovided script [`create_dataset.py`](./scripts/create_dataset.py) under the `scripts` directory in this project. We will upload this script to an S3 location and refer to that location in our `ProcessingStep`.\n",
    "\n",
    "We will define all the subsequent steps using Sagemaker processing `ScriptProcessor` since we will be running external scripts within each of those steps. For more information on `ScriptProcessor` see [documentation](https://docs.aws.amazon.com/sagemaker/latest/dg/processing-container-run-scripts.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "%store -r CONTAINER_IMAGE_URI\n",
    "\n",
    "create_dataset_processor = ScriptProcessor(command=['python3'],\n",
    "                                           image_uri=CONTAINER_IMAGE_URI,\n",
    "                                           role=sagemaker_role,\n",
    "                                           instance_count=flow_instance_count,\n",
    "                                           instance_type=flow_instance_type)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_client.upload_file(Filename='./scripts/create_dataset.py', Bucket=afd_bucket, Key=f'{afd_prefix}/afd-pipeline/code/create_dataset.py')\n",
    "\n",
    "create_dataset_step = ProcessingStep(\n",
    "    name='Step3CreateAFDTrainingDataset',\n",
    "    processor=create_dataset_processor,\n",
    "    outputs=[sagemaker.processing.ProcessingOutput(output_name='train_data', \n",
    "                                                   source='/opt/ml/processing/output/train', \n",
    "                                                   destination=data_path_param),\n",
    "             sagemaker.processing.ProcessingOutput(output_name='train_schema', \n",
    "                                                   source='/opt/ml/processing/output/schema', \n",
    "                                                   destination=f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/train-schema')],\n",
    "    job_arguments=[\"--signups-feature-group-name\", signups_fg_name,\n",
    "                   \"--outcomes-feature-group-name\", outcomes_fg_name,\n",
    "                   \"--region\", region,\n",
    "                   \"--bucket-name\", afd_bucket,\n",
    "                   \"--bucket-prefix\", afd_prefix],\n",
    "    code=create_dataset_script_uri,\n",
    "    depends_on=['Step2OutcomesDataWranglerProcessing'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 4: Train Amazon Fraud Detector Model Step <a id='step4'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "We will create a similar script processor as we did in the previous step. The script in this step will make use of the dataset and the data schema files generated by the previous stem to train our Amazon Fraud Detector model. The model train call is asynchronous so we will wait for the process to complete and finally store the response information of the model training into a file. The response will contain all the information that will be required for avtivating the model in the next step. The Amazon Fraud detector training script is included under the scripts folder [`train_afd.py`](./scripts/train_afd.py). This script, does the following -\n",
    "\n",
    "* Trains the AFD Model with the data generated in the previous step\n",
    "* Generates a training response data to be used by subsequent steps\n",
    "* Generates a property file with the AUC Metric value of the model (more on this later in Step 6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "afd_train_processor = ScriptProcessor(command=['python3'],\n",
    "                                      image_uri=CONTAINER_IMAGE_URI,\n",
    "                                      role=sagemaker_role,\n",
    "                                      instance_count=train_instance_count,\n",
    "                                      instance_type=train_instance_type)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_client.upload_file(Filename='./scripts/train_afd.py', Bucket=afd_bucket, Key=f'{afd_prefix}/afd-pipeline/code/train_afd.py')\n",
    "\n",
    "#define the property file which will store the AUC metric from the outcome of the model training\n",
    "\n",
    "training_response = PropertyFile(\n",
    "    name=\"AUCPropertyFile\",\n",
    "    output_name=\"training_auc\",\n",
    "    path=\"train_auc.json\"           # the property file generated by the train_afd.py that the Pipeline will index and keep track of to evaluate later\n",
    ")\n",
    "\n",
    "afd_train_processingstep = ProcessingStep(name=\"Step4AFDModelTrainProcess\",\n",
    "                                          processor=afd_train_processor,\n",
    "                                          job_arguments=[\"--region\", region,\n",
    "                                                         \"--s3-file-loc\", f'{data_path_param}/afd_training_data.csv',\n",
    "                                                         \"--data-access-role\", data_role_param,\n",
    "                                                         \"--model-name\", model_name_param],\n",
    "                                          inputs=[sagemaker.processing.ProcessingInput(source=create_dataset_step.properties.ProcessingOutputConfig.Outputs[\"train_schema\"].S3Output.S3Uri,\n",
    "                                                                                       destination='/opt/ml/processing/schema')],\n",
    "                                          outputs=[sagemaker.processing.ProcessingOutput(output_name='training_response', \n",
    "                                                                                         source='/opt/ml/processing/output',\n",
    "                                                                                         destination=f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/train-response'),\n",
    "                                                   sagemaker.processing.ProcessingOutput(output_name='training_auc', \n",
    "                                                                                         source='/opt/ml/processing/auc',\n",
    "                                                                                         destination=f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/train-response')],\n",
    "                                          property_files=[training_response],\n",
    "                                          code=train_model_script_uri\n",
    "                                         )\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 5: Check AUC Metric (Area Under the ROC Curve) Condition <a id='step5'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "AFD Training completion generates [training metrics](https://docs.aws.amazon.com/frauddetector/latest/api/API_TrainingMetrics.html) such as the Area Under the ROC Curve. Ideally, an AUC value ranges between `0` and `1`. The closer the AUC value is to `1` the better the model's accuracy. To learn more about ROC/AUC check this [article](https://www.sciencedirect.com/science/article/pii/S1556086415306043). Typically, this would involve a human review of the metric to make a decision on whether the AUC value is acceptable or not in order to activate the model, which is what we did in the 3rd notebook. In this case, we will use Pipeline Conditions to make a decision on whether to progress to the next step and activate the model in an automated manner. Step 4, the training step, generates the AUC metric property file which can be read in this step and evaluated. We are assuming an AUC equal to or above a threshold of `0.75` as acceptable. The high level logic is `if auc >= 0.75 then activate_model`. \n",
    "\n",
    "We defined the AUC threshold as a Pipeline parameter `auc_threshold_param` with a default value of `0.75` at the beginning of this notebook. We will use this parameter in our Pipeline Condition to make a decision whether to proceed to the next step i.e. `afd_activate_processingstep` (defined in the previous section - Step 4) which will activate the trained AFD Model.\n",
    "\n",
    "We created our Activate Model step in Step 5 (`afd_activate_processingstep`) which is ready to execute if the condition evalues to `true`. We will set that up in the following code cell.\n",
    "\n",
    "In order to create a _condition_ within the SageMaker Pipeline we will use a [ConditionStep](https://docs.aws.amazon.com/sagemaker/latest/dg/build-and-manage-steps.html#step-type-condition). You can define, conditions such as `equals to`, `greater than`, `less than`, `less than or equals to`, `greater than or equals to` and so on with a Condition step and then execute other processing step(s) when the conditon is true or false using an `if...else` pattern. In our case we want to use the `ConditionGreaterThanOrEqualTo` condition to check the AUC metric and only activate the model (by running the activate model step) if the metric is greater than or equal to the pre-defined threshold."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo\n",
    "from sagemaker.workflow.condition_step import ( ConditionStep, JsonGet )\n",
    "\n",
    "condition_gte = ConditionGreaterThanOrEqualTo(\n",
    "    left=JsonGet(                           #the left value of the evaluation expression\n",
    "        step=afd_train_processingstep,      #the step from which the property file will be grabbed\n",
    "        property_file=training_response,    #the property file instance that was created earlier in Step 4\n",
    "        json_path=\"auc_metric\"              #the JSON path of the property within the property file train_auc.json (refer train_afd.py line 71)\n",
    "    ),\n",
    "    right=auc_threshold_param               #the right value of the evaluation expression, i.e. the AUC threshold\n",
    ")\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 6: Activate Amazon Fraud Detector Model Step <a id='step6'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "This step will activate the trained AFD Model in case the AUC threshold is met in the condition step (which we will defined in the previous step). This step uses the script `activate_afd.py` included in the `scripts` directory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "afd_activate_processor = ScriptProcessor(command=['python3'],\n",
    "                                         image_uri=CONTAINER_IMAGE_URI,\n",
    "                                         role=sagemaker_role,\n",
    "                                         instance_count=train_instance_count,\n",
    "                                         instance_type=train_instance_type)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_client.upload_file(Filename='./scripts/activate_afd.py', Bucket=afd_bucket, Key=f'{afd_prefix}/afd-pipeline/code/activate_afd.py')\n",
    "\n",
    "afd_activate_processingstep = ProcessingStep(name=\"Step6AFDModelActivateProcess\",\n",
    "                                          processor=afd_activate_processor,\n",
    "                                          job_arguments=[\"--region\", region],\n",
    "                                          inputs=[sagemaker.processing.ProcessingInput(source=afd_train_processingstep.properties.ProcessingOutputConfig.Outputs[\"training_response\"].S3Output.S3Uri,\n",
    "                                                                                       destination='/opt/ml/processing/input')],\n",
    "                                          outputs=[sagemaker.processing.ProcessingOutput(output_name='activation_response', \n",
    "                                                                                         source='/opt/ml/processing/output',\n",
    "                                                                                         destination=f's3://{afd_bucket}/{afd_prefix}/afd-pipeline/train-response')],\n",
    "                                          code=activate_model_script_uri\n",
    "                                         )\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we have the model activation step defined, we can setup the condition which we created in Step 5 to check the AUC metric. If the metric is `>= 0.75` then pipeline will kick-off the model activation step, else the pipeline ends there."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Define the condition step\n",
    "auc_condition_step = ConditionStep(\n",
    "    name=\"Step5CheckAUCThreshold\",\n",
    "    conditions=[condition_gte],             #the greater than equal to condition defined above\n",
    "    if_steps=[afd_activate_processingstep], #if the condition evaluates to true then Step 5 processing step will be executed\n",
    "    else_steps=[]                           #there are no else steps so we will keep it empty\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 7: Setup Amazon Fraud Detector Model detector Step <a id='step7'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "This is the last and final step and it will be executed after the AFD model has been activated based on the condition check. In this step, we will do a few things-\n",
    "* Setup Outcomes (if they don't exist already)\n",
    "* Setup rules based on the metrics generated during the model training stage and map them to the outcomes\n",
    "* Setup a new Detector Version using the new list of rules created above\n",
    "\n",
    "We will not activate this detector, however, activating the new detector version is certainly a step that can be added on to the pipeline. Note, activating the detector is akin to deploying that detector as the latest version in production (you can only have 1 detector in `ACTIVE` status at a time, activating this detector will de-activate your previous detector version and may inadvertently break your application's code). \n",
    "\n",
    "In certain cases, you may not want to activate a new detector version created by this automated pipeline and may want to analyze the results of the training further before activating the new detector version manually, or you may want to wait until your next production release cycle etc. The detector version created in this step will continue to remain in `DRAFT` status until a manual action is taken either via the Amazon Fraud Detector console or using the `update_detector_version_status` API.\n",
    "\n",
    "Similar to Steps 4 & 5, we will be running a script to setup the new detector version, the script `setup_detector.py` is included in the project `scripts` directory. We will create a new version of the existing detector with the updated Rules. The new version of the detector will remain in `DRAFT` status."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "setup_detector_processor = ScriptProcessor(command=['python3'],\n",
    "                                         image_uri=CONTAINER_IMAGE_URI,\n",
    "                                         role=sagemaker_role,\n",
    "                                         instance_count=train_instance_count,\n",
    "                                         instance_type=train_instance_type)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_client.upload_file(Filename='./scripts/setup_detector.py', Bucket=afd_bucket, Key=f'{afd_prefix}/afd-pipeline/code/setup_detector.py')\n",
    "\n",
    "setup_detector_processingstep = ProcessingStep(name=\"Step7AFDSetupDetectorProcess\",\n",
    "                                          processor=setup_detector_processor,\n",
    "                                          job_arguments=[\"--region\", region,\n",
    "                                                         \"--detector-name\", detector_name_param],\n",
    "                                          inputs=[sagemaker.processing.ProcessingInput(source=afd_activate_processingstep.properties.ProcessingOutputConfig.Outputs[\"activation_response\"].S3Output.S3Uri,\n",
    "                                                                                       destination='/opt/ml/processing/input')],\n",
    "                                          code=setup_detector_script_uri\n",
    "                                         )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Combine the Pipeline Steps and Run <a id='define-pipeline'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "Now that we have defined all the processing and condition steps, we will setup the SageMaker Pipeline. Though easier to reason with, the parameters and steps don't need to be in order. The pipeline DAG will parse it out properly."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Stored 'afd_pipeline_name' (str)\n"
     ]
    }
   ],
   "source": [
    "afd_pipeline_name = f'AFDPipeline'\n",
    "%store afd_pipeline_name\n",
    "\n",
    "pipeline = Pipeline(\n",
    "    name=afd_pipeline_name,\n",
    "    parameters=[\n",
    "        model_name_param, \n",
    "        detector_name_param,\n",
    "        data_role_param,\n",
    "        data_path_param,\n",
    "        auc_threshold_param],\n",
    "    steps=[\n",
    "        signups_flow_step,\n",
    "        outcomes_flow_step,\n",
    "        create_dataset_step,        \n",
    "        afd_train_processingstep,\n",
    "        auc_condition_step,\n",
    "        setup_detector_processingstep\n",
    "    ])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4.1 Submit the pipeline definition to the SageMaker Pipeline service\n",
    "We now have the Pipeline and all of it's steps and condition defined. The next step would be to create or update the Pipeline. Note, If an existing pipeline has the same name it will be overwritten with the `upsert` function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "import botocore\n",
    "\n",
    "try:\n",
    "    pipeline.upsert(role_arn=sagemaker_role)\n",
    "except botocore.exceptions.ClientError as error:\n",
    "    print(error.response)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we will execute the pipeline i.e. start execution of the pipeline. This API action can also be done from, let's say, a Lambda function. You can also override the Pipeline parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Special pipeline parameters can be defined or changed here\n",
    "# parameters = {'TrainingInstance': 'ml.m5.xlarge'}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "# start_response = pipeline.start(parameters=parameters)\n",
    "start_response = pipeline.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the Pipleine execution starts, you can view the execution status of the pipeline as shown below\n",
    "\n",
    "<img src=\"images/pipeline_exec.png\" width=\"800\" height=\"800\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Clicking on the \"**Graph**\" tab will show the Pipeline DAG (Directed Acyclic Graph). The Pipeline DAG while it's executing will look like below\n",
    "\n",
    "<img src=\"images/pipeline.png\" width=\"800\" height=\"800\"/>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "start_response.wait()\n",
    "start_response.describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5. Delete Pipeline [Optional] <a id='delete'></a>\n",
    "---\n",
    "<a href=\"#overview\">overview</a>\n",
    "\n",
    "You may, optionally, delete the pipeline as a cleanup step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sagemaker_boto_client.delete_pipeline(PipelineName=afd_pipeline_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:us-east-2:429704687514: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
}