10_pipeline/stepfunctions/01_Create_Pipeline_Train_and_Deploy_Reviews_BERT_TensorFlow.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# NOTE:  THIS NOTEBOOK WILL TAKE A 30 MINUTES TO COMPLETE.\n",
    "\n",
    "# PLEASE BE PATIENT."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create a Training Pipeline with the Step Functions Data Science SDK"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Step Functions SageMaker Pipeline](img/stepfunctions_graph.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from botocore.exceptions import ClientError\n",
    "\n",
    "import os\n",
    "import sagemaker\n",
    "import logging\n",
    "import boto3\n",
    "import sagemaker\n",
    "import pandas as pd\n",
    "\n",
    "sess   = sagemaker.Session()\n",
    "bucket = sess.default_bucket()\n",
    "role = sagemaker.get_execution_role()\n",
    "region = boto3.Session().region_name\n",
    "\n",
    "sm = boto3.Session().client(service_name='sagemaker', region_name=region)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import stepfunctions\n",
    "import logging\n",
    "from stepfunctions.template.pipeline import TrainingPipeline\n",
    "\n",
    "stepfunctions.set_stream_logger(level=logging.INFO)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create an IAM Execution Role for Step Functions\n",
    "We need a StepFunctionsWorkflowExecutionRole so that you can create and execute workflows in Step Functions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "iam = boto3.Session().client(service_name='iam', region_name=region)\n",
    "sts = boto3.Session().client(service_name='sts', region_name=region)\n",
    "sfn = boto3.Session().client(service_name='stepfunctions', region_name=region)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "stepfunction_role_name = 'DSOAWS_StepFunctionsExecutionRole'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create an AssumeRolePolicyDocument"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assume_role_policy_doc = {\n",
    "  \"Version\": \"2012-10-17\",\n",
    "  \"Statement\": [\n",
    "    {\n",
    "      \"Effect\": \"Allow\",\n",
    "      \"Principal\": {\n",
    "        \"Service\": \"states.amazonaws.com\"\n",
    "      },\n",
    "      \"Action\": \"sts:AssumeRole\"\n",
    "    }\n",
    "  ]\n",
    "} "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create `DSOAWS_StepFunctionsExecutionRole`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import time\n",
    "try:\n",
    "    iam.create_role(\n",
    "        RoleName=stepfunction_role_name,\n",
    "        AssumeRolePolicyDocument=json.dumps(assume_role_policy_doc),\n",
    "        Description='DSOAWS Step Function Workflow Execution Role'\n",
    "    )\n",
    "    time.sleep(10)\n",
    "    print(\"Role created.\")\n",
    "except ClientError as e:\n",
    "    if e.response['Error']['Code'] == 'EntityAlreadyExists':\n",
    "        print(\"Role already exists. This is OK.\")\n",
    "    else:\n",
    "        print(\"Unexpected error: %s\" % e)\n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get the Role ARN"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "stepfunction_role = iam.get_role(RoleName=stepfunction_role_name)\n",
    "stepfunction_role_arn = stepfunction_role['Role']['Arn']\n",
    "print(stepfunction_role_arn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Add a Policy to the Role"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define permissions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "stepfunction_permissions = {\n",
    "    \"Version\": \"2012-10-17\",\n",
    "    \"Statement\": [\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"sagemaker:CreateTransformJob\",\n",
    "                \"sagemaker:DescribeTransformJob\",\n",
    "                \"sagemaker:StopTransformJob\",\n",
    "                \"sagemaker:CreateTrainingJob\",\n",
    "                \"sagemaker:DescribeTrainingJob\",\n",
    "                \"sagemaker:StopTrainingJob\",\n",
    "                \"sagemaker:CreateHyperParameterTuningJob\",\n",
    "                \"sagemaker:DescribeHyperParameterTuningJob\",\n",
    "                \"sagemaker:StopHyperParameterTuningJob\",\n",
    "                \"sagemaker:CreateModel\",\n",
    "                \"sagemaker:CreateEndpointConfig\",\n",
    "                \"sagemaker:CreateEndpoint\",\n",
    "                \"sagemaker:DeleteEndpointConfig\",\n",
    "                \"sagemaker:DeleteEndpoint\",\n",
    "                \"sagemaker:UpdateEndpoint\",\n",
    "                \"sagemaker:CreateProcessingJob\",\n",
    "                \"sagemaker:DescribeProcessingJob\",\n",
    "                \"sagemaker:ListProcessingJobs\",\n",
    "                \"sagemaker:StopProcessingJob\",                \n",
    "                \"sagemaker:ListTags\",\n",
    "                \"lambda:InvokeFunction\",\n",
    "                \"sqs:SendMessage\",\n",
    "                \"sns:Publish\",\n",
    "                \"ecs:RunTask\",\n",
    "                \"ecs:StopTask\",\n",
    "                \"ecs:DescribeTasks\",\n",
    "                \"dynamodb:GetItem\",\n",
    "                \"dynamodb:PutItem\",\n",
    "                \"dynamodb:UpdateItem\",\n",
    "                \"dynamodb:DeleteItem\",\n",
    "                \"batch:SubmitJob\",\n",
    "                \"batch:DescribeJobs\",\n",
    "                \"batch:TerminateJob\",\n",
    "                \"glue:StartJobRun\",\n",
    "                \"glue:GetJobRun\",\n",
    "                \"glue:GetJobRuns\",\n",
    "                \"glue:BatchStopJobRun\"\n",
    "            ],\n",
    "            \"Resource\": \"*\"\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"iam:PassRole\"\n",
    "            ],\n",
    "            \"Resource\": \"*\",\n",
    "            \"Condition\": {\n",
    "                \"StringEquals\": {\n",
    "                    \"iam:PassedToService\": \"sagemaker.amazonaws.com\"\n",
    "                }\n",
    "            }\n",
    "        },\n",
    "        {\n",
    "            \"Effect\": \"Allow\",\n",
    "            \"Action\": [\n",
    "                \"events:PutTargets\",\n",
    "                \"events:PutRule\",\n",
    "                \"events:DescribeRule\"\n",
    "            ],\n",
    "            \"Resource\": [\n",
    "                \"arn:aws:events:*:*:rule/StepFunctionsGetEventsForSageMakerTrainingJobsRule\",\n",
    "                \"arn:aws:events:*:*:rule/StepFunctionsGetEventsForSageMakerTransformJobsRule\",\n",
    "                \"arn:aws:events:*:*:rule/StepFunctionsGetEventsForSageMakerTuningJobsRule\",\n",
    "                \"arn:aws:events:*:*:rule/StepFunctionsGetEventsForECSTaskRule\",\n",
    "                \"arn:aws:events:*:*:rule/StepFunctionsGetEventsForBatchJobsRule\",\n",
    "            ]\n",
    "        }\n",
    "    ]\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Turn into Policy Object"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "stepfunction_policy_name = 'DSOAWS_StepFunctionsWorkflowExecutionPolicy'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "account_id = sts.get_caller_identity()['Account']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "try:\n",
    "    stepfunction_policy = iam.create_policy(\n",
    "      PolicyName=stepfunction_policy_name,\n",
    "      PolicyDocument=json.dumps(stepfunction_permissions)\n",
    "    )\n",
    "    stepfunction_policy_arn = f'arn:aws:iam::{account_id}:policy/{stepfunction_policy_name}'\n",
    "    print(\"Policy created.\")\n",
    "except ClientError as e:\n",
    "    if e.response['Error']['Code'] == 'EntityAlreadyExists':\n",
    "        print(\"Policy already exists.  Updating policy...\")\n",
    "        stepfunction_policy_arn = f'arn:aws:iam::{account_id}:policy/{stepfunction_policy_name}'\n",
    "        try:\n",
    "            stepfunction_policy = iam.create_policy_version(\n",
    "                PolicyArn=stepfunction_policy_arn,\n",
    "                PolicyDocument=json.dumps(stepfunction_permissions),\n",
    "                SetAsDefault=True)\n",
    "            print('Policy updated.')\n",
    "        except:\n",
    "            print('** Policy cannot have more than 5 versions.  This is likely OK.')\n",
    "    else:\n",
    "        print(\"Unexpected error: %s\" % e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(stepfunction_policy_arn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Attach Policy To Step Function Workflow Execution Role"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "try:\n",
    "    response = iam.attach_role_policy(\n",
    "        PolicyArn=stepfunction_policy_arn,\n",
    "        RoleName=stepfunction_role_name\n",
    "    )\n",
    "    print(\"Done.\")\n",
    "except ClientError as e:\n",
    "    if e.response['Error']['Code'] == 'EntityAlreadyExists':\n",
    "        print(\"Policy is already attached. This is OK.\")\n",
    "    else:\n",
    "        print(\"Unexpected error: %s\" % e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "try:\n",
    "    response = iam.attach_role_policy(\n",
    "        PolicyArn='arn:aws:iam::aws:policy/service-role/AWSLambdaRole',\n",
    "        RoleName=stepfunction_role_name\n",
    "    )\n",
    "    print(\"Done.\")\n",
    "except ClientError as e:\n",
    "    if e.response['Error']['Code'] == 'EntityAlreadyExists':\n",
    "        print(\"Policy is already attached. This is OK.\")\n",
    "    else:\n",
    "        print(\"Unexpected error: %s\" % e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "try:\n",
    "    response = iam.attach_role_policy(\n",
    "        PolicyArn='arn:aws:iam::aws:policy/CloudWatchEventsFullAccess',\n",
    "        RoleName=stepfunction_role_name\n",
    "    )\n",
    "    print(\"Done.\")\n",
    "except ClientError as e:\n",
    "    if e.response['Error']['Code'] == 'EntityAlreadyExists':\n",
    "        print(\"Policy is already attached. This is OK.\")\n",
    "    else:\n",
    "        print(\"Unexpected error: %s\" % e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Processing Step\n",
    "\n",
    "![](img/prepare_dataset_bert.png)\n",
    "\n",
    "![](img/processing.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Upload the Processing Script to S3 for the Pipeline to Consume"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pygmentize ./preprocess-scikit-text-to-bert.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "processing_code_s3_prefix = 'pipeline_sklearn_processing/{}/code'.format(int(time.time()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_code = sess.upload_data(\n",
    "    './preprocess-scikit-text-to-bert.py',\n",
    "    bucket=bucket,\n",
    "    key_prefix=processing_code_s3_prefix,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store processing_code_s3_prefix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(processing_code_s3_prefix)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Set the Processing Hyper-Parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "max_seq_length=64\n",
    "train_split_percentage=0.90\n",
    "validation_split_percentage=0.05\n",
    "test_split_percentage=0.05\n",
    "balance_dataset=True\n",
    "processing_instance_count=1\n",
    "processing_instance_type='ml.c5.2xlarge'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Specify the Raw Inputs S3 Location"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_input_data_s3_uri = 's3://{}/amazon-reviews-pds/tsv/'.format(bucket)\n",
    "print(raw_input_data_s3_uri)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!aws s3 ls $raw_input_data_s3_uri"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.sklearn.processing import SKLearnProcessor\n",
    "\n",
    "processor = SKLearnProcessor(framework_version='0.20.0',\n",
    "                             role=role,\n",
    "                             instance_type=processing_instance_type,\n",
    "                             instance_count=processing_instance_count,\n",
    "                             max_runtime_in_seconds=7200)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Training Step\n",
    "\n",
    "![](img/bert_training.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Show Training Script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pygmentize src/tf_bert_reviews.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Training Hyper-Parameters\n",
    "Note that `max_seq_length` is re-used from the processing hyper-parameters above"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "epochs=3\n",
    "learning_rate=0.00001\n",
    "epsilon=0.00000001\n",
    "train_batch_size=128\n",
    "validation_batch_size=128\n",
    "test_batch_size=128\n",
    "train_steps_per_epoch=100\n",
    "validation_steps=100\n",
    "test_steps=100\n",
    "train_instance_count=1\n",
    "train_instance_type='ml.c5.9xlarge'\n",
    "train_volume_size=1024\n",
    "use_xla=True\n",
    "use_amp=True\n",
    "freeze_bert_layer=False\n",
    "enable_sagemaker_debugger=False\n",
    "enable_checkpointing=False\n",
    "enable_tensorboard=False\n",
    "input_mode='File'\n",
    "run_validation=True\n",
    "run_test=True\n",
    "run_sample_predictions=True\n",
    "deploy_instance_count=1\n",
    "# deploy_instance_type='ml.m5.4xlarge'\n",
    "deploy_instance_type='ml.m5.large' # bur"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Metrics To Track Model Performance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "metrics_definitions = [\n",
    "     {'Name': 'train:loss', 'Regex': 'loss: ([0-9\\\\.]+)'},\n",
    "     {'Name': 'train:accuracy', 'Regex': 'accuracy: ([0-9\\\\.]+)'},\n",
    "     {'Name': 'validation:loss', 'Regex': 'val_loss: ([0-9\\\\.]+)'},\n",
    "     {'Name': 'validation:accuracy', 'Regex': 'val_accuracy: ([0-9\\\\.]+)'},\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Estimator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.tensorflow import TensorFlow\n",
    "\n",
    "estimator = TensorFlow(entry_point='tf_bert_reviews.py',\n",
    "                       source_dir='src',\n",
    "                       role=role,\n",
    "                       instance_count=train_instance_count, # Make sure you have at least this number of input files or the ShardedByS3Key distibution strategy will fail the job due to no data available\n",
    "                       instance_type=train_instance_type,\n",
    "                       volume_size=train_volume_size,                       \n",
    "                       py_version='py3',\n",
    "                       framework_version='2.1.0',\n",
    "                       hyperparameters={'epochs': epochs,\n",
    "                                        'learning_rate': learning_rate,\n",
    "                                        'epsilon': epsilon,\n",
    "                                        'train_batch_size': train_batch_size,\n",
    "                                        'validation_batch_size': validation_batch_size,\n",
    "                                        'test_batch_size': test_batch_size,                                             \n",
    "                                        'train_steps_per_epoch': train_steps_per_epoch,\n",
    "                                        'validation_steps': validation_steps,\n",
    "                                        'test_steps': test_steps,\n",
    "                                        'use_xla': use_xla,\n",
    "                                        'use_amp': use_amp,                                             \n",
    "                                        'max_seq_length': max_seq_length,\n",
    "                                        'freeze_bert_layer': freeze_bert_layer,\n",
    "                                        'enable_sagemaker_debugger': enable_sagemaker_debugger,\n",
    "                                        'enable_checkpointing': enable_checkpointing,\n",
    "                                        'enable_tensorboard': enable_tensorboard,                                        \n",
    "                                        'run_validation': run_validation,\n",
    "                                        'run_test': run_test,\n",
    "                                        'run_sample_predictions': run_sample_predictions},\n",
    "                       input_mode=input_mode,\n",
    "                       metric_definitions=metrics_definitions,\n",
    "#                       max_run=7200 # max 2 hours * 60 minutes seconds per hour * 60 seconds per minute\n",
    "                      )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup the Deploy\n",
    "![](img/realtime_inference.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup Pipeline with the Step Functions SDK\n",
    "\n",
    "A typical task for a data scientist is to train a model and deploy that model to an endpoint. Without the Step Functions SDK, this is a four step process on SageMaker that includes the following.\n",
    "\n",
    "1. Training the model\n",
    "2. Creating the model on SageMaker\n",
    "3. Creating an endpoint configuration\n",
    "4. Deploying the trained model to the configured endpoint\n",
    "\n",
    "The Step Functions SDK provides the [TrainingPipeline](https://aws-step-functions-data-science-sdk.readthedocs.io/en/latest/pipelines.html#stepfunctions.template.pipeline.train.TrainingPipeline) API to simplify this procedure. The following configures `pipeline` with the necessary parameters to define a training pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "timestamp = int(time.time())\n",
    "\n",
    "pipeline_name = 'bert-pipeline-{}'.format(timestamp)\n",
    "\n",
    "print('Pipeline name {}'.format(pipeline_name))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "\n",
    "from sagemaker.utils import base_name_from_image\n",
    "from sagemaker.sklearn.estimator import SKLearn\n",
    "from sagemaker.model import Model\n",
    "from sagemaker.pipeline import PipelineModel\n",
    "from sagemaker.processing import ProcessingInput, ProcessingOutput\n",
    "from sagemaker.inputs import TrainingInput\n",
    "\n",
    "from stepfunctions.steps import (\n",
    "    TrainingStep, \n",
    "    TransformStep, \n",
    "    ModelStep, \n",
    "    EndpointConfigStep, \n",
    "    EndpointStep, \n",
    "    Chain, \n",
    "    Fail, \n",
    "    Catch,\n",
    "    ProcessingStep\n",
    ")\n",
    "from stepfunctions.workflow import Workflow\n",
    "from stepfunctions.template.pipeline.common import WorkflowTemplate\n",
    "from stepfunctions.template.pipeline.common import StepId\n",
    "\n",
    "class TrainingPipelineWithDifferentDeployInstanceTypeAndProcessingJob(WorkflowTemplate):\n",
    "\n",
    "    \"\"\"\n",
    "    Creates a standard training pipeline with the following steps in order:\n",
    "        1. Train estimator\n",
    "        2. Create estimator model\n",
    "        3. Endpoint configuration\n",
    "        4. Deploy model\n",
    "    \"\"\"\n",
    "\n",
    "    __allowed_kwargs = ('pipeline_name',)\n",
    "    \n",
    "    def __init__(self, \n",
    "                 processor,\n",
    "                 raw_input_data_s3_uri,\n",
    "                 train_split_percentage,\n",
    "                 validation_split_percentage,\n",
    "                 test_split_percentage,\n",
    "                 max_seq_length,\n",
    "                 balance_dataset,\n",
    "                 estimator, \n",
    "                 role, \n",
    "                 bucket,                  \n",
    "                 client, \n",
    "                 deploy_instance_count, \n",
    "                 deploy_instance_type, \n",
    "                 **kwargs):\n",
    "        \"\"\"\n",
    "        Args:\n",
    "            estimator (sagemaker.estimator.EstimatorBase): The estimator to use for training. Can be a BYO estimator, Framework estimator or Amazon algorithm estimator.\n",
    "            role (str): An AWS IAM role (either name or full Amazon Resource Name (ARN)). This role is used to create, manage, and execute the Step Functions workflows.\n",
    "            inputs: Information about the training data. Please refer to the `fit()` method of the associated estimator, as this can take any of the following forms:\n",
    "                * (str) - The S3 location where training data is saved.\n",
    "                * (dict[str, str] or dict[str, `sagemaker.session.s3_input`]) - If using multiple channels for training data, you can specify a dict mapping channel names to strings or `sagemaker.session.s3_input` objects.\n",
    "                * (`sagemaker.session.s3_input`) - Channel configuration for S3 data sources that can provide additional information about the training dataset. See `sagemaker.session.s3_input` for full details.\n",
    "                * (`sagemaker.amazon.amazon_estimator.RecordSet`) - A collection of Amazon `Record` objects serialized and stored in S3. For use with an estimator for an Amazon algorithm.\n",
    "                * (list[`sagemaker.amazon.amazon_estimator.RecordSet`]) - A list of `sagemaker.amazon.amazon_estimator.RecordSet` objects, where each instance is a different channel of training data.\n",
    "            bucket (str): S3 bucket under which the output artifacts from the training job will be stored. The parent path used is built using the format: ``s3://{bucket}/{pipeline_name}/models/{job_name}/``. In this format, `pipeline_name` refers to the keyword argument provided for TrainingPipeline. If a `pipeline_name` argument was not provided, one is auto-generated by the pipeline as `training-pipeline-<timestamp>`. Also, in the format, `job_name` refers to the job name provided when calling the :meth:`TrainingPipeline.run()` method.\n",
    "            client (SFN.Client, optional): boto3 client to use for creating and interacting with the training pipeline in Step Functions. (default: None)\n",
    "        Keyword Args:\n",
    "            pipeline_name (str, optional): Name of the pipeline. This name will be used to name jobs (if not provided when calling execute()), models, endpoints, and S3 objects created by the pipeline. If a `pipeline_name` argument was not provided, one is auto-generated by the pipeline as `training-pipeline-<timestamp>`. (default:None)\n",
    "        \"\"\"\n",
    "        self.processor = processor \n",
    "        self.raw_input_data_s3_uri = raw_input_data_s3_uri\n",
    "        self.train_split_percentage = train_split_percentage\n",
    "        self.validation_split_percentage = validation_split_percentage\n",
    "        self.test_split_percentage = test_split_percentage\n",
    "        self.max_seq_length = max_seq_length\n",
    "        self.balance_dataset = balance_dataset\n",
    "        self.estimator = estimator\n",
    "        self.role = role        \n",
    "        self.bucket = bucket\n",
    "        self.deploy_instance_count = deploy_instance_count\n",
    "        self.deploy_instance_type = deploy_instance_type\n",
    "\n",
    "        for key in self.__class__.__allowed_kwargs:\n",
    "            setattr(self, key, kwargs.pop(key, None))\n",
    "\n",
    "        if not self.pipeline_name:\n",
    "            self.__pipeline_name_unique = True\n",
    "            self.pipeline_name = 'training-pipeline-{date}'.format(date=self._generate_timestamp())\n",
    "\n",
    "        self.definition = self.build_workflow_definition()\n",
    "        self.input_template = self._extract_input_template(self.definition)\n",
    "\n",
    "        workflow = Workflow(name=self.pipeline_name, \n",
    "                            definition=self.definition, \n",
    "                            role=role, \n",
    "                            format_json=True, \n",
    "                            client=client)\n",
    "\n",
    "        super(TrainingPipelineWithDifferentDeployInstanceTypeAndProcessingJob, self).__init__(s3_bucket=bucket, \n",
    "                                                                                              workflow=workflow, \n",
    "                                                                                              role=role, \n",
    "                                                                                              client=client)\n",
    "    \n",
    "    def build_workflow_definition(self):\n",
    "        \"\"\"\n",
    "        Build the workflow definition for the training pipeline with all the states involved.\n",
    "        Returns:\n",
    "            :class:`~stepfunctions.steps.states.Chain`: Workflow definition as a chain of states involved in the the training pipeline.\n",
    "        \"\"\"\n",
    "\n",
    "        processing_inputs=[\n",
    "                ProcessingInput(\n",
    "                    input_name='raw_input',\n",
    "                    source=raw_input_data_s3_uri,\n",
    "                    destination='/opt/ml/processing/input/data/',\n",
    "                    s3_data_distribution_type='ShardedByS3Key'\n",
    "                ),\n",
    "                ProcessingInput(\n",
    "                    input_name='code',            \n",
    "                    source=input_code,\n",
    "                    destination='/opt/ml/processing/input/code',\n",
    "                )\n",
    "        ]\n",
    "\n",
    "        processed_train_data_s3_uri = 's3://{}/{}/processing/output/bert-train'.format(self.bucket, self.pipeline_name)        \n",
    "        processed_validation_data_s3_uri = 's3://{}/{}/processing/output/bert-validation'.format(self.bucket, self.pipeline_name)        \n",
    "        processed_test_data_s3_uri = 's3://{}/{}/processing/output/bert-test'.format(self.bucket, self.pipeline_name)\n",
    "         \n",
    "        processing_outputs=[\n",
    "                ProcessingOutput(s3_upload_mode='EndOfJob',\n",
    "                                 output_name='bert-train',\n",
    "                                 source='/opt/ml/processing/output/bert/train',\n",
    "                                 destination=processed_train_data_s3_uri\n",
    "                                ),\n",
    "                ProcessingOutput(s3_upload_mode='EndOfJob',\n",
    "                                 output_name='bert-validation',\n",
    "                                 source='/opt/ml/processing/output/bert/validation',\n",
    "                                 destination=processed_validation_data_s3_uri\n",
    "                                ),\n",
    "                ProcessingOutput(s3_upload_mode='EndOfJob',\n",
    "                                 output_name='bert-test',\n",
    "                                 source='/opt/ml/processing/output/bert/test',\n",
    "                                 destination=processed_test_data_s3_uri\n",
    "                                ),\n",
    "        ]        \n",
    "\n",
    "        processing_step = ProcessingStep(\n",
    "            'Processing Job', # StepId.ProcessingJob.value?\n",
    "            processor=self.processor,\n",
    "            job_name=self.pipeline_name,\n",
    "            inputs=processing_inputs,\n",
    "            outputs=processing_outputs,\n",
    "            # experiment_config=experiment_config,\n",
    "            container_arguments=['--train-split-percentage', str(self.train_split_percentage),\n",
    "                                 '--validation-split-percentage', str(self.validation_split_percentage),\n",
    "                                 '--test-split-percentage', str(self.test_split_percentage),\n",
    "                                 '--max-seq-length', str(self.max_seq_length),\n",
    "                                 '--balance-dataset', str(self.balance_dataset)],\n",
    "            container_entrypoint=['python3', '/opt/ml/processing/input/code/preprocess-scikit-text-to-bert.py'],\n",
    "        )        \n",
    "\n",
    "        s3_input_train_data = TrainingInput(s3_data=processed_train_data_s3_uri, distribution='ShardedByS3Key')\n",
    "        s3_input_validation_data = TrainingInput(s3_data=processed_validation_data_s3_uri, distribution='ShardedByS3Key')\n",
    "        s3_input_test_data = TrainingInput(s3_data=processed_test_data_s3_uri, distribution='ShardedByS3Key')\n",
    "\n",
    "        training_step = TrainingStep(\n",
    "            StepId.Train.value,\n",
    "            estimator=self.estimator,\n",
    "            job_name=self.pipeline_name + '/estimator-source',\n",
    "            data={\n",
    "                'train': s3_input_train_data,\n",
    "                'validation': s3_input_validation_data,\n",
    "                'test': s3_input_test_data\n",
    "            },\n",
    "        )\n",
    "\n",
    "        model = self.estimator.create_model()\n",
    "        model_step = ModelStep(\n",
    "            StepId.CreateModel.value,\n",
    "            instance_type=deploy_instance_type,\n",
    "            model=model,\n",
    "            model_name=self.pipeline_name\n",
    "        )\n",
    "\n",
    "        endpoint_config_step = EndpointConfigStep(\n",
    "            StepId.ConfigureEndpoint.value,\n",
    "            endpoint_config_name=self.pipeline_name,\n",
    "            model_name=self.pipeline_name,\n",
    "            initial_instance_count=self.deploy_instance_count,\n",
    "            instance_type=self.deploy_instance_type\n",
    "        )\n",
    "        \n",
    "        deploy_step = EndpointStep(\n",
    "            StepId.Deploy.value,\n",
    "            endpoint_name=self.pipeline_name,\n",
    "            endpoint_config_name=self.pipeline_name,\n",
    "        )\n",
    "\n",
    "        return Chain([\n",
    "            processing_step, \n",
    "            training_step, \n",
    "            model_step, \n",
    "            endpoint_config_step, \n",
    "            deploy_step\n",
    "        ])\n",
    "    \n",
    "    def execute(self, job_name=None, hyperparameters=None):\n",
    "        \"\"\"\n",
    "        Run the training pipeline.\n",
    "        \n",
    "        Args:\n",
    "            job_name (str, optional): Name for the training job. If one is not provided, a job name will be auto-generated. (default: None)\n",
    "            hyperparameters (dict, optional): Hyperparameters for the estimator training. (default: None)\n",
    "        \n",
    "        Returns:\n",
    "            :py:class:`~stepfunctions.workflow.Execution`: Running instance of the training pipeline.\n",
    "        \"\"\"\n",
    "        inputs = self.input_template.copy()\n",
    "        \n",
    "        if hyperparameters is not None:\n",
    "            inputs[StepId.Train.value]['HyperParameters'] = {\n",
    "                k: str(v) for k, v in hyperparameters.items()\n",
    "            }\n",
    "        \n",
    "        if job_name is None:\n",
    "            job_name = '{base_name}-{timestamp}'.format(base_name='training-pipeline', timestamp=self._generate_timestamp())\n",
    "            \n",
    "        print(inputs)\n",
    "        \n",
    "        # Configure training and model\n",
    "        inputs[StepId.Train.value]['TrainingJobName'] = 'estimator-' + job_name\n",
    "        inputs[StepId.Train.value]['OutputDataConfig']['S3OutputPath'] = 's3://{s3_bucket}/{pipeline_name}/models'.format(\n",
    "            s3_bucket=self.s3_bucket,\n",
    "            pipeline_name=self.workflow.name\n",
    "        )\n",
    "        inputs[StepId.CreateModel.value]['ModelName'] = job_name\n",
    "\n",
    "        # Configure endpoint\n",
    "        inputs[StepId.ConfigureEndpoint.value]['EndpointConfigName'] = job_name\n",
    "        for variant in inputs[StepId.ConfigureEndpoint.value]['ProductionVariants']:\n",
    "            variant['ModelName'] = job_name\n",
    "        inputs[StepId.Deploy.value]['EndpointConfigName'] = job_name\n",
    "        inputs[StepId.Deploy.value]['EndpointName'] = job_name\n",
    "        \n",
    "        # Configure the path to model artifact\n",
    "        inputs[StepId.CreateModel.value]['PrimaryContainer']['ModelDataUrl'] = '{s3_uri}/{job}/output/model.tar.gz'.format(\n",
    "            s3_uri=inputs[StepId.Train.value]['OutputDataConfig']['S3OutputPath'],\n",
    "            job=inputs[StepId.Train.value]['TrainingJobName']\n",
    "        )\n",
    "        \n",
    "        return self.workflow.execute(inputs=inputs, name=job_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Note:  If you see an error about 'TensorFlowModel' object has no attribute 'image', you are using SageMaker SDK 1.x\n",
    "#        The Data Science SDK only supports 1.x at this time.\n",
    "#        Waiting on this:  https://github.com/aws/aws-step-functions-data-science-sdk-python/issues/69\n",
    "\n",
    "pipeline = TrainingPipelineWithDifferentDeployInstanceTypeAndProcessingJob(\n",
    "    processor=processor,\n",
    "    raw_input_data_s3_uri=raw_input_data_s3_uri,\n",
    "    train_split_percentage=train_split_percentage,\n",
    "    validation_split_percentage=validation_split_percentage,\n",
    "    test_split_percentage=test_split_percentage,\n",
    "    max_seq_length=max_seq_length,\n",
    "    balance_dataset=balance_dataset,\n",
    "    estimator=estimator,\n",
    "    role=stepfunction_role_arn,\n",
    "    bucket=bucket,\n",
    "    client=sfn,\n",
    "    deploy_instance_count=deploy_instance_count,\n",
    "    deploy_instance_type=deploy_instance_type,    \n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Visualize the pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can now view the workflow definition, and also visualize it as a graph. This workflow and graph represent your training pipeline. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## View the workflow definition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(pipeline.workflow.definition.to_json(pretty=True))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Visualize the workflow graph\n",
    "## *Note: This only renders in Jupyter. NOT in JupyterLab.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline.render_graph()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## You should see a graph like this:\n",
    " \n",
    "<img src=\"img/pipeline_created.png\" width=\"70%\" align=\"left\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create and execute the pipeline on AWS Step Functions\n",
    "\n",
    "Create the pipeline in AWS Step Functions with [create](https://aws-step-functions-data-science-sdk.readthedocs.io/en/latest/workflow.html#stepfunctions.workflow.Workflow.create)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sleeping to wait for role and policy creations\n",
    "import time\n",
    "time.sleep(10)\n",
    "\n",
    "pipeline.create()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run the workflow with [execute](https://aws-step-functions-data-science-sdk.readthedocs.io/en/latest/workflow.html#stepfunctions.workflow.Workflow.execute). A link will be provided after the following cell is executed. Following this link, you can monitor your pipeline execution on Step Functions' console."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution = pipeline.execute(job_name=None,\n",
    "                             hyperparameters=None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "stepfunction_arn = 'arn:aws:states:{}:{}:stateMachine:{}'.format(region, account_id, pipeline.pipeline_name)\n",
    "print(stepfunction_arn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store stepfunction_arn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "stepfunction_name = pipeline.pipeline_name\n",
    "print(stepfunction_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store stepfunction_name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Check Pipeline Progress\n",
    "_Note: This only renders in Jupyter at the moment - not in JupyterLab.  This is changing soon._"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "execution.render_progress()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## You should see a graph like this:\n",
    "\n",
    "<img src=\"img/pipeline_executed.png\" width=\"90%\" align=\"left\">"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "import time\n",
    "\n",
    "events = execution.list_events()\n",
    "\n",
    "while len(events) <= 5:\n",
    "    print('Number of events:  {}'.format(len(events)))\n",
    "    time.sleep(30)\n",
    "    events = execution.list_events()\n",
    "\n",
    "print('Number of events:  {}'.format(len(events)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution.render_progress()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# _Wait for ^^ Number of Events ^^ to Reach At Least 6_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "processing_job_name = json.loads(events[5]['taskSucceededEventDetails']['output'])['ProcessingJobName']\n",
    "print('Processing Job Name: {}'.format(processing_job_name))\n",
    "\n",
    "print('')\n",
    "\n",
    "processing_job_outputs = json.loads(events[5]['taskSucceededEventDetails']['output'])['ProcessingOutputConfig']['Outputs']\n",
    "\n",
    "for output in processing_job_outputs:\n",
    "    if output['OutputName'] == 'bert-train':\n",
    "        train_data_s3_uri = output['S3Output']['S3Uri']\n",
    "    if output['OutputName'] == 'bert-validation':\n",
    "        validation_data_s3_uri = output['S3Output']['S3Uri']\n",
    "    if output['OutputName'] == 'bert-test':\n",
    "        test_data_s3_uri = output['S3Output']['S3Uri']\n",
    "\n",
    "print('Processed Data Bert Train S3 URI: {}'.format(train_data_s3_uri))\n",
    "print('Processed Data Bert Validation S3 URI: {}'.format(validation_data_s3_uri))\n",
    "print('Processed Data Bert Test S3 URI: {}'.format(test_data_s3_uri))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.s3 import S3Downloader\n",
    "print(S3Downloader.list(train_data_s3_uri))\n",
    "\n",
    "from sagemaker.s3 import S3Downloader\n",
    "print(S3Downloader.list(validation_data_s3_uri))\n",
    "\n",
    "from sagemaker.s3 import S3Downloader\n",
    "print(S3Downloader.list(test_data_s3_uri))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "import time\n",
    "\n",
    "events = execution.list_events()\n",
    "\n",
    "while len(events) <= 11:\n",
    "    print('Number of events:  {}'.format(len(events)))\n",
    "    time.sleep(30)\n",
    "    events = execution.list_events()\n",
    "\n",
    "print('Number of events:  {}'.format(len(events)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution.render_progress()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "training_job_name = json.loads(events[11]['taskSucceededEventDetails']['output'])['TrainingJobName']\n",
    "print('Training Job Name: {}'.format(training_job_name))\n",
    "\n",
    "print('')\n",
    "\n",
    "trained_model_s3_uri = json.loads(events[11]['taskSucceededEventDetails']['output'])['ModelArtifacts']['S3ModelArtifacts']\n",
    "print('Trained Model S3 URI: {}'.format(trained_model_s3_uri))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.display import display, HTML\n",
    "\n",
    "display(HTML('<b>Review <a target=\"blank\" href=\"https://console.aws.amazon.com/sagemaker/home?region={}#/jobs/{}\">Training Job</a></b>'.format(region, training_job_name)))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Copy the Model from S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!aws s3 cp $trained_model_s3_uri ./model.tar.gz"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!mkdir -p ./model/\n",
    "!tar -xvzf ./model.tar.gz -C ./model/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Show the Model Prediction Signature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!saved_model_cli show --all --dir ./model/tensorflow/saved_model/0/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "import time\n",
    "\n",
    "events = execution.list_events()\n",
    "\n",
    "while len(events) <= 24:\n",
    "    print('Number of events:  {}'.format(len(events)))\n",
    "    time.sleep(30)\n",
    "    events = execution.list_events()\n",
    "\n",
    "print('Number of events:  {}'.format(len(events)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "execution.render_progress()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# _Wait for ^^ Number of Events ^^ to Reach At Least 19_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "step_functions_pipeline_endpoint_name = json.loads(events[24]['taskScheduledEventDetails']['parameters'])['EndpointName']\n",
    "\n",
    "print('Endpoint Name: {}'.format(step_functions_pipeline_endpoint_name))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "import time\n",
    "\n",
    "events = execution.list_events()\n",
    "\n",
    "while len(events) <= 27:\n",
    "    print('Number of events:  {}'.format(len(events)))\n",
    "    time.sleep(30)\n",
    "    events = execution.list_events()    \n",
    "\n",
    "print('Number of events:  {}'.format(len(events)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "execution.render_progress()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# _Wait for ^^ Number of Events ^^ to Reach At Least 22_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "step_functions_pipeline_endpoint_arn = json.loads(events[27]['stateExitedEventDetails']['output'])['EndpointArn']\n",
    "\n",
    "print('Endpoint ARN: {}'.format(step_functions_pipeline_endpoint_arn))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.display import display, HTML\n",
    "\n",
    "display(HTML('<b>Review <a target=\"blank\" href=\"https://console.aws.amazon.com/sagemaker/home?region={}#/endpoints/{}\">SageMaker REST Endpoint</a></b>'.format(region, step_functions_pipeline_endpoint_name)))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Pass Variables to the Next Notebooks(s)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "print(step_functions_pipeline_endpoint_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store step_functions_pipeline_endpoint_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%javascript\n",
    "Jupyter.notebook.save_checkpoint();\n",
    "Jupyter.notebook.session.delete();"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.10"
  },
  "notice": "Copyright 2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.  Licensed under the Apache License, Version 2.0 (the \"License\"). You may not use this file except in compliance with the License. A copy of the License is located at http://aws.amazon.com/apache2.0/ or in the \"license\" file accompanying this file. This file is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License."
 },
 "nbformat": 4,
 "nbformat_minor": 4
}