Text classification model using BlazingText with Amazon SageMaker, Spark Pipeline, and AWS Glue .ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Text classification model using BlazingText with Amazon SageMaker, Spark Pipeline, and AWS Glue "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this example, we will train the text classification model using SageMaker BlazingText algorithm on the DBPedia Ontology Dataset as done by Zhang et al. Many of the training step resembles the 'Predict age of Alabone' notebook, so explanations are limited in this example, and we dive into code directly\n",
    "\n",
    "The DBpedia ontology dataset is constructed by picking 14 nonoverlapping classes from DBpedia 2014. It has 560,000 training samples and 70,000 testing samples. The fields we used for this dataset contain title and abstract of each Wikipedia article."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In many cases, when the trained model is used for processing real time or batch prediction requests, the model receives data in a format which needs to pre-processed (e.g. featurized) before it can be passed to the algorithm. In the following notebook, we will demonstrate how you can build your ML Pipeline leveraging Spark Feature Transformers and SageMaker BlazingText algorithm & after the model is trained, deploy the Pipeline (Feature Transformer and BlazingText) as an Inference Pipeline behind a single Endpoint for real-time inference and for batch inferences using Amazon SageMaker Batch Transform.\n",
    "\n",
    "In this notebook, we use Amazon Glue to run serverless Spark. Though the notebook demonstrates the end-to-end flow on a small dataset, the setup can be seamlessly used to scale to larger datasets."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Methods**\n",
    "The Notebook consists of a few high-level steps:\n",
    "\n",
    "- Using AWS Glue for executing the SparkML feature processing job.\n",
    "- Using SageMaker BlazingText to train on the processed dataset produced by SparkML job.\n",
    "- Building an Inference Pipeline consisting of SparkML & BlazingText models for a realtime inference endpoint.\n",
    "- Building an Inference Pipeline consisting of SparkML & BlazingText models for a single Batch Transform job."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to enable this Notebook to run AWS Glue jobs, we need to add one additional permission to the default execution role of this notebook. Please follow the steps listed in [this notebook example](https://github.com/hyunjoonbok/amazon-sagemaker/blob/master/Predict%20the%20age%20of%20Abalone%20(regression%20problem)%20with%20Amazon%20SageMaker%2C%20Spark%20Pipeline%2C%20and%20AWS%20Glue%20.ipynb).  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np                                # For matrix operations and numerical processing\n",
    "import pandas as pd                               # For munging tabular data\n",
    "import matplotlib.pyplot as plt                   # For charts and visualizations\n",
    "import matplotlib.ticker as ticker\n",
    "%matplotlib inline\n",
    "\n",
    "from IPython.display import Image                 # For displaying images in the notebook\n",
    "from IPython.display import display               # For displaying outputs in the notebook\n",
    "from time import gmtime, strftime                 # For labeling SageMaker models, endpoints, etc.\n",
    "import time\n",
    "import sys                                        # For writing outputs to notebook\n",
    "import math                                       # For ceiling function\n",
    "import json                                       # For parsing hosting outputs\n",
    "import os                                         # For manipulating filepath names\n",
    "import boto3\n",
    "import botocore\n",
    "from botocore.exceptions import ClientError\n",
    "\n",
    "import csv\n",
    "import io\n",
    "import re\n",
    "import s3fs\n",
    "import mxnet as mx\n",
    "import seaborn as sns\n",
    "import pickle\n",
    "import gzip\n",
    "import urllib\n",
    "import csv\n",
    "\n",
    "import cv2\n",
    "\n",
    "import sagemaker                                 \n",
    "from sagemaker.predictor import csv_serializer \n",
    "from sagemaker.predictor import json_deserializer\n",
    "import sagemaker.amazon.common as smac\n",
    "from sagemaker import get_execution_role"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Preparation (Specifying Sagemaker roles)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sagemaker session : <sagemaker.session.Session object at 0x000001A4364E5748>\n",
      "BoTo3 session : Session(region_name='us-west-2')\n",
      "Account: 570447867175\n",
      "S3 bucket : aws-glue-570447867175-us-west-2\n",
      "Region selected : us-west-2\n",
      "IAM role : arn:aws:iam::570447867175:role/SageMakerNotebookRole\n"
     ]
    }
   ],
   "source": [
    "sess = sagemaker.Session()\n",
    "boto_session = sess.boto_session\n",
    "s3 = boto_session.resource('s3')\n",
    "account = boto_session.client('sts').get_caller_identity()['Account']\n",
    "region = boto3.Session().region_name\n",
    "default_bucket = 'aws-glue-{}-{}'.format(account, region)                     \n",
    "role = 'arn:aws:iam::570447867175:role/SageMakerNotebookRole' # pass your IAM role name\n",
    "\n",
    "print('Sagemaker session :', sess)\n",
    "print('BoTo3 session :', boto_session)\n",
    "print('Account:', account)\n",
    "print('S3 bucket :', default_bucket)\n",
    "print('Region selected :', region)\n",
    "print('IAM role :', role)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "'get_image_uri' method will be deprecated in favor of 'ImageURIProvider' class in SageMaker Python SDK v2.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "433757028032.dkr.ecr.us-west-2.amazonaws.com/blazingtext:latest\n"
     ]
    }
   ],
   "source": [
    "from sagemaker.amazon.amazon_estimator import get_image_uri\n",
    "\n",
    "training_image = get_image_uri(region, 'blazingtext', repo_version=\"latest\")\n",
    "print(training_image)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Load Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dataset can be downloaded from this website [Link](https://wiki.dbpedia.org/services-resources/dbpedia-data-set-2014#2) as done by Zhang et al.\n",
    "Or you can try wget method for easy download from Official AWS Sagemaker S3. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !wget https://s3-us-west-2.amazonaws.com/sparkml-mleap/data/dbpedia/train.csv\n",
    "# !wget https://s3-us-west-2.amazonaws.com/sparkml-mleap/data/dbpedia/test.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "A bucket with the same name already exists in your account - using the same bucket.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'s3://aws-glue-570447867175-us-west-2/input/dbpedia/test.csv'"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "try:\n",
    "    if region == 'us-east-1':\n",
    "        s3.create_bucket(Bucket=default_bucket)\n",
    "    else:\n",
    "        s3.create_bucket(Bucket=default_bucket, CreateBucketConfiguration={'LocationConstraint': region})\n",
    "except ClientError as e:\n",
    "    error_code = e.response['Error']['Code']\n",
    "    message = e.response['Error']['Message']\n",
    "    if error_code == 'BucketAlreadyOwnedByYou':\n",
    "        print ('A bucket with the same name already exists in your account - using the same bucket.')\n",
    "        pass        \n",
    "\n",
    "# Uploading the training data to S3\n",
    "# (I have changed the filename to dbpedia-train.csv and dbpedia-test.csv)\n",
    "sess.upload_data(path='train.csv', bucket=default_bucket, key_prefix='input/dbpedia')    \n",
    "sess.upload_data(path='test.csv', bucket=default_bucket, key_prefix='input/dbpedia')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Upload the featurizer script to S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "s3://aws-glue-570447867175-us-west-2/codes/dbpedia_processing.py\n"
     ]
    }
   ],
   "source": [
    "script_location = sess.upload_data(path='./tools/dbpedia_processing.py', bucket=default_bucket, key_prefix='codes')\n",
    "print(script_location)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Upload MLeap dependencies to S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !wget https://s3-us-west-2.amazonaws.com/sparkml-mleap/0.9.6/python/python.zip\n",
    "# !wget https://s3-us-west-2.amazonaws.com/sparkml-mleap/0.9.6/jar/mleap_spark_assembly.jar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "python_dep_location = sess.upload_data(path='python.zip', bucket=default_bucket, key_prefix='dependencies/python')\n",
    "jar_dep_location = sess.upload_data(path='mleap_spark_assembly.jar', bucket=default_bucket, key_prefix='dependencies/jar')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "python Dependency Location : s3://aws-glue-570447867175-us-west-2/dependencies/python/python.zip\n",
      "Jar file Location s3://aws-glue-570447867175-us-west-2/dependencies/jar/mleap_spark_assembly.jar\n"
     ]
    }
   ],
   "source": [
    "print('python Dependency Location :', python_dep_location)\n",
    "print('Jar file Location', jar_dep_location)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Defining output locations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "timestamp_prefix = strftime(\"%Y-%m-%d-%H-%M-%S\", gmtime())\n",
    "\n",
    "# Input location of the data, We uploaded our train.csv file to input key previously\n",
    "s3_input_bucket = default_bucket\n",
    "s3_input_key_prefix = 'input/dbpedia'\n",
    "\n",
    "# Output location of the data. The input data will be split, transformed, and \n",
    "# uploaded to output/train and output/validation\n",
    "s3_output_bucket = default_bucket\n",
    "s3_output_key_prefix = timestamp_prefix + '/dbpedia'\n",
    "\n",
    "# the MLeap serialized SparkML model will be uploaded to output/mleap\n",
    "s3_model_bucket = default_bucket\n",
    "s3_model_key_prefix = s3_output_key_prefix + '/mleap'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Calling Glue APIs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sparkml-dbpedia-2020-07-13-05-08-47\n"
     ]
    }
   ],
   "source": [
    "glue_client = boto_session.client('glue')\n",
    "job_name = 'sparkml-dbpedia-' + timestamp_prefix\n",
    "response = glue_client.create_job(\n",
    "    Name=job_name,\n",
    "    Description='PySpark job to featurize the DBPedia dataset',\n",
    "    Role=role, # you can pass your existing AWS Glue role here if you have used Glue before\n",
    "    ExecutionProperty={\n",
    "        'MaxConcurrentRuns': 1\n",
    "    },\n",
    "    Command={\n",
    "        'Name': 'glueetl',\n",
    "        'ScriptLocation': script_location\n",
    "    },\n",
    "    DefaultArguments={\n",
    "        '--job-language': 'python',\n",
    "        '--extra-jars' : jar_dep_location,\n",
    "        '--extra-py-files': python_dep_location\n",
    "    },\n",
    "    AllocatedCapacity=10,\n",
    "    Timeout=60,\n",
    ")\n",
    "glue_job_name = response['Name']\n",
    "print(glue_job_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "jr_199c05a2bda8a761946a552a41b1ebffe74b8be40899d3c2fbf89b24b02d8b47\n"
     ]
    }
   ],
   "source": [
    "job_run_id = glue_client.start_job_run(JobName=job_name,\n",
    "                                       Arguments = {\n",
    "                                        '--S3_INPUT_BUCKET': s3_input_bucket,\n",
    "                                        '--S3_INPUT_KEY_PREFIX': s3_input_key_prefix,\n",
    "                                        '--S3_OUTPUT_BUCKET': s3_output_bucket,\n",
    "                                        '--S3_OUTPUT_KEY_PREFIX': s3_output_key_prefix,\n",
    "                                        '--S3_MODEL_BUCKET': s3_model_bucket,\n",
    "                                        '--S3_MODEL_KEY_PREFIX': s3_model_key_prefix\n",
    "                                       })['JobRunId']\n",
    "print(job_run_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###### Checking Glue job status"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "RUNNING\n",
      "RUNNING\n",
      "RUNNING\n",
      "SUCCEEDED\n"
     ]
    }
   ],
   "source": [
    "job_run_status = glue_client.get_job_run(JobName=job_name,RunId=job_run_id)['JobRun']['JobRunState']\n",
    "while job_run_status not in ('FAILED', 'SUCCEEDED', 'STOPPED'):\n",
    "    job_run_status = glue_client.get_job_run(JobName=job_name,RunId=job_run_id)['JobRun']['JobRunState']\n",
    "    print(job_run_status)\n",
    "    time.sleep(30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Create a Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:root:Parameter image_name will be renamed to image_uri in SageMaker Python SDK v2.\n",
      "WARNING:sagemaker:'s3_input' class will be renamed to 'TrainingInput' in SageMaker Python SDK v2.\n",
      "WARNING:sagemaker:'s3_input' class will be renamed to 'TrainingInput' in SageMaker Python SDK v2.\n"
     ]
    }
   ],
   "source": [
    "s3_train_data = 's3://{}/{}/{}'.format(s3_output_bucket, s3_output_key_prefix, 'train')\n",
    "s3_validation_data = 's3://{}/{}/{}'.format(s3_output_bucket, s3_output_key_prefix, 'validation')\n",
    "s3_output_location = 's3://{}/{}/{}'.format(s3_output_bucket, s3_output_key_prefix, 'bt_model')\n",
    "\n",
    "bt_model = sagemaker.estimator.Estimator(training_image,\n",
    "                                         role, \n",
    "                                         train_instance_count=1, \n",
    "                                         train_instance_type='ml.p2.xlarge',\n",
    "                                         train_volume_size = 20,\n",
    "                                         train_max_run = 3600,\n",
    "                                         input_mode= 'File',\n",
    "                                         output_path=s3_output_location,\n",
    "                                         sagemaker_session=sess)\n",
    "\n",
    "bt_model.set_hyperparameters(mode=\"supervised\",\n",
    "                            epochs=10,\n",
    "                            min_count=2,\n",
    "                            learning_rate=0.05,\n",
    "                            vector_dim=10,\n",
    "                            early_stopping=True,\n",
    "                            patience=4,\n",
    "                            min_epochs=5,\n",
    "                            word_ngrams=2)\n",
    "\n",
    "train_data = sagemaker.session.s3_input(s3_train_data, distribution='FullyReplicated', \n",
    "                        content_type='text/plain', s3_data_type='S3Prefix')\n",
    "validation_data = sagemaker.session.s3_input(s3_validation_data, distribution='FullyReplicated', \n",
    "                             content_type='text/plain', s3_data_type='S3Prefix')\n",
    "\n",
    "data_channels = {'train': train_data, 'validation': validation_data}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Start Training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2020-07-13 05:11:44 Starting - Starting the training job...\n",
      "2020-07-13 05:11:46 Starting - Launching requested ML instances......\n",
      "2020-07-13 05:12:55 Starting - Preparing the instances for training......\n",
      "2020-07-13 05:13:59 Downloading - Downloading input data...\n",
      "2020-07-13 05:14:42 Training - Training image download completed. Training in progress.\u001b[34mArguments: train\u001b[0m\n",
      "\u001b[34m[07/13/2020 05:14:43 WARNING 139844829321024] Loggers have already been setup.\u001b[0m\n",
      "\u001b[34m[07/13/2020 05:14:43 WARNING 139844829321024] Loggers have already been setup.\u001b[0m\n",
      "\u001b[34m[07/13/2020 05:14:43 INFO 139844829321024] nvidia-smi took: 0.100651025772 secs to identify 1 gpus\u001b[0m\n",
      "\u001b[34m[07/13/2020 05:14:43 INFO 139844829321024] Running BlazingText on singe GPU using supervised\u001b[0m\n",
      "\u001b[34m[07/13/2020 05:14:43 INFO 139844829321024] Processing /opt/ml/input/data/train/part-00000 . File size: 158 MB\u001b[0m\n",
      "\u001b[34m[07/13/2020 05:14:43 INFO 139844829321024] Processing /opt/ml/input/data/validation/part-00000 . File size: 19 MB\u001b[0m\n",
      "\u001b[34mRead 10M words\u001b[0m\n",
      "\u001b[34mRead 20M words\u001b[0m\n",
      "\u001b[34mRead 26M words\u001b[0m\n",
      "\u001b[34mNumber of words:  381326\u001b[0m\n",
      "\u001b[34mInitialized GPU 0 successfully! Now starting training....\u001b[0m\n",
      "\u001b[34mLoading validation data from /opt/ml/input/data/validation/part-00000\u001b[0m\n",
      "\u001b[34mLoaded validation data on GPU.\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0482  Progress: 3.52%  Million Words/sec: 8.27 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0457  Progress: 8.65%  Million Words/sec: 9.52 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 1\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0431  Progress: 13.89%  Million Words/sec: 9.98 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0405  Progress: 18.92%  Million Words/sec: 10.09 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 2\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0379  Progress: 24.19%  Million Words/sec: 10.10 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0353  Progress: 29.30%  Million Words/sec: 10.18 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 3\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0327  Progress: 34.52%  Million Words/sec: 10.27 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0301  Progress: 39.88%  Million Words/sec: 10.27 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 4\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0275  Progress: 45.05%  Million Words/sec: 10.32 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0248  Progress: 50.39%  Million Words/sec: 10.31 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 5\u001b[0m\n",
      "\u001b[34mValidation accuracy: 0.543\u001b[0m\n",
      "\u001b[34mValidation accuracy improved! Storing best weights...\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0223  Progress: 55.39%  Million Words/sec: 10.30 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 6\u001b[0m\n",
      "\u001b[34mValidation accuracy: 0.530786\u001b[0m\n",
      "\u001b[34mValidation accuracy has not improved for last 1 epochs.\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0198  Progress: 60.44%  Million Words/sec: 10.31 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0172  Progress: 65.51%  Million Words/sec: 10.32 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 7\u001b[0m\n",
      "\u001b[34mValidation accuracy: 0.6343\u001b[0m\n",
      "\u001b[34mValidation accuracy improved! Storing best weights...\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0147  Progress: 70.61%  Million Words/sec: 10.33 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0122  Progress: 75.63%  Million Words/sec: 10.33 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 8\u001b[0m\n",
      "\u001b[34mValidation accuracy: 0.653714\u001b[0m\n",
      "\u001b[34mValidation accuracy improved! Storing best weights...\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0096  Progress: 80.70%  Million Words/sec: 10.33 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0071  Progress: 85.84%  Million Words/sec: 10.35 #####\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 9\u001b[0m\n",
      "\u001b[34mValidation accuracy: 0.816114\u001b[0m\n",
      "\u001b[34mValidation accuracy improved! Storing best weights...\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0044  Progress: 91.24%  Million Words/sec: 10.35 #####\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0017  Progress: 96.61%  Million Words/sec: 10.34 #####\u001b[0m\n",
      "\u001b[34mExiting thread 3\u001b[0m\n",
      "\u001b[34mExiting thread 1\u001b[0m\n",
      "\u001b[34mExiting thread 2\u001b[0m\n",
      "\u001b[34mExiting thread 0\u001b[0m\n",
      "\u001b[34m-------------- End of epoch: 10\u001b[0m\n",
      "\u001b[34mValidation accuracy: 0.944029\u001b[0m\n",
      "\u001b[34mValidation accuracy improved! Storing best weights...\u001b[0m\n",
      "\u001b[34m##### Alpha: 0.0000  Progress: 100.00%  Million Words/sec: 10.33 #####\u001b[0m\n",
      "\u001b[34mTraining finished.\u001b[0m\n",
      "\u001b[34mAverage throughput in Million words/sec: 10.33\u001b[0m\n",
      "\u001b[34mTotal training time in seconds: 26.11\n",
      "\u001b[0m\n",
      "\u001b[34m#train_accuracy: 0.9489\u001b[0m\n",
      "\u001b[34mNumber of train examples: 560000\n",
      "\u001b[0m\n",
      "\u001b[34m#validation_accuracy: 0.944\u001b[0m\n",
      "\u001b[34mNumber of validation examples: 70000\u001b[0m\n",
      "\n",
      "2020-07-13 05:15:50 Uploading - Uploading generated training model\n",
      "2020-07-13 05:15:50 Completed - Training job completed\n",
      "Training seconds: 111\n",
      "Billable seconds: 111\n"
     ]
    }
   ],
   "source": [
    "bt_model.fit(inputs=data_channels, logs=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5. Inference Pipeline in Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\"input\": [{\"name\": \"abstract\", \"type\": \"string\"}], \"output\": {\"name\": \"tokenized_abstract\", \"type\": \"string\", \"struct\": \"array\"}}\n"
     ]
    }
   ],
   "source": [
    "# Passing the schema of the payload via environment variable\n",
    "\n",
    "schema = {\n",
    "    \"input\": [\n",
    "        {\n",
    "            \"name\": \"abstract\",\n",
    "            \"type\": \"string\"\n",
    "        }\n",
    "    ],\n",
    "    \"output\": \n",
    "        {\n",
    "            \"name\": \"tokenized_abstract\",\n",
    "            \"type\": \"string\",\n",
    "            \"struct\": \"array\"\n",
    "        }\n",
    "}\n",
    "schema_json = json.dumps(schema)\n",
    "print(schema_json)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:sagemaker:Parameter image will be renamed to image_uri in SageMaker Python SDK v2.\n",
      "WARNING:sagemaker:Parameter image will be renamed to image_uri in SageMaker Python SDK v2.\n"
     ]
    }
   ],
   "source": [
    "# In order for the sagemaker-sparkml-serving to emit the output with the right format,\n",
    "# we need to pass a second environment variable SAGEMAKER_DEFAULT_INVOCATIONS_ACCEPT \n",
    "# with the value application/jsonlines;data=text to ensure that sagemaker-sparkml-serving container emits response in the proper format which BlazingText can parse.\n",
    "\n",
    "from sagemaker.model import Model\n",
    "from sagemaker.pipeline import PipelineModel\n",
    "from sagemaker.sparkml.model import SparkMLModel\n",
    "\n",
    "sparkml_data = 's3://{}/{}/{}'.format(s3_model_bucket, s3_model_key_prefix, 'model.tar.gz')\n",
    "# passing the schema defined above by using an environment variable that sagemaker-sparkml-serving understands\n",
    "sparkml_model = SparkMLModel(model_data=sparkml_data,\n",
    "                             env={'SAGEMAKER_SPARKML_SCHEMA' : schema_json, \n",
    "                                  'SAGEMAKER_DEFAULT_INVOCATIONS_ACCEPT': \"application/jsonlines;data=text\"})\n",
    "bt_model = Model(model_data=bt_model.model_data, image=training_image)\n",
    "\n",
    "model_name = 'inference-pipeline-' + timestamp_prefix\n",
    "sm_model = PipelineModel(name=model_name, role=role, models=[sparkml_model, bt_model])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------!"
     ]
    }
   ],
   "source": [
    "# Deploying the PipelineModel to SageMaker Endpoint for realtime inference\n",
    "endpoint_name = 'inference-pipeline-ep-' + timestamp_prefix\n",
    "sm_model.deploy(initial_instance_count=1, \n",
    "                instance_type='ml.p2.xlarge', \n",
    "                endpoint_name=endpoint_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "b'{\"label\": [\"__label__1\"], \"prob\": [0.9886016249656677]}\\n'\n"
     ]
    }
   ],
   "source": [
    "# Passing a CSV string one\n",
    "from sagemaker.predictor import json_serializer, csv_serializer, json_deserializer, RealTimePredictor\n",
    "from sagemaker.content_types import CONTENT_TYPE_CSV, CONTENT_TYPE_JSON\n",
    "payload = \"Convair was an american aircraft manufacturing company which later expanded into rockets and spacecraft.\"\n",
    "predictor = RealTimePredictor(endpoint=endpoint_name, sagemaker_session=sess, serializer=csv_serializer,\n",
    "                                content_type=CONTENT_TYPE_CSV, accept='application/jsonlines')\n",
    "print(predictor.predict(payload))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "b'{\"label\": [\"__label__2\"], \"prob\": [0.9784008860588074]}\\n'\n"
     ]
    }
   ],
   "source": [
    "# Passing a JSON string one\n",
    "payload = {\"data\": [\"Berwick secondary college is situated in the outer melbourne metropolitan suburb of berwick .\"]}\n",
    "predictor = RealTimePredictor(endpoint=endpoint_name, sagemaker_session=sess, serializer=json_serializer,\n",
    "                                content_type=CONTENT_TYPE_JSON)\n",
    "\n",
    "print(predictor.predict(payload))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6. Close the SageMaker Instance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'ResponseMetadata': {'RequestId': 'e0244f7d-01f7-48a7-a677-0f00496c6a1d',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amzn-requestid': 'e0244f7d-01f7-48a7-a677-0f00496c6a1d',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '0',\n",
       "   'date': 'Mon, 13 Jul 2020 05:26:52 GMT'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sm_client = boto_session.client('sagemaker')\n",
    "sm_client.delete_endpoint(EndpointName=endpoint_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 7. (optional) Building an Inference Pipeline for a single Batch Transform job"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "SageMaker Batch Transform also supports chaining multiple containers together when deploying an Inference Pipeline and performing a single Batch Transform job to transform your data for a batch use-case similar to the real-time use-case we have seen above.\n",
    "\n",
    "**Preparing data for Batch Transform**\n",
    "Batch Transform requires data in the same format described above, with one CSV or JSON being per line. For this notebook, SageMaker team has created a sample input in CSV format which Batch Transform can process. The input is a simple CSV file with one input string per line.\n",
    "\n",
    "Next we will download a sample of this data from one of the SageMaker buckets (named batch_input_dbpedia.csv) and upload to your S3 bucket. We will also inspect first five rows of the data post downloading."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !wget https://s3-us-west-2.amazonaws.com/sparkml-mleap/data/batch_input_dbpedia.csv\n",
    "# !printf \"\\n\\nShowing first two lines\\n\\n\"    \n",
    "# !head -n 3 batch_input_dbpedia.csv\n",
    "# !printf \"\\n\\nAs we can see, it is just one input string per line.\\n\\n\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "batch_input_loc = sess.upload_data(path='batch_input_dbpedia.csv', bucket=default_bucket, key_prefix='batch')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Invoking the Transform API to create a Batch Transform job\n",
    "\n",
    "input_data_path = 's3://{}/{}/{}'.format(default_bucket, 'batch', 'batch_input_dbpedia.csv')\n",
    "output_data_path = 's3://{}/{}/{}'.format(default_bucket, 'batch_output/dbpedia', timestamp_prefix)\n",
    "transformer = sagemaker.transformer.Transformer(\n",
    "    model_name = model_name,\n",
    "    instance_count = 1,\n",
    "    instance_type = 'ml.m4.xlarge',\n",
    "    strategy = 'SingleRecord',\n",
    "    assemble_with = 'Line',\n",
    "    output_path = output_data_path,\n",
    "    base_transform_job_name='serial-inference-batch',\n",
    "    sagemaker_session=sess,\n",
    "    accept = CONTENT_TYPE_CSV\n",
    ")\n",
    "transformer.transform(data = input_data_path, \n",
    "                        content_type = CONTENT_TYPE_CSV, \n",
    "                        split_type = 'Line')\n",
    "transformer.wait()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "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.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}