sagemaker-python-sdk/scikit_learn_iris/Scikit-learn Estimator Example With Batch Transform.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Iris Training and Prediction with Sagemaker Scikit-learn\n",
    "This tutorial shows you how to use [Scikit-learn](https://scikit-learn.org/stable/) with Sagemaker by utilizing the pre-built container. Scikit-learn is a popular Python machine learning framework. It includes a number of different algorithms for classification, regression, clustering, dimensionality reduction, and data/feature pre-processing. \n",
    "\n",
    "The [sagemaker-python-sdk](https://github.com/aws/sagemaker-python-sdk) module  makes it easy to take existing scikit-learn code, which we will show by training a model on the IRIS dataset and generating a set of predictions. For more information about the Scikit-learn container, see the [sagemaker-scikit-learn-containers](https://github.com/aws/sagemaker-scikit-learn-container) repository and the [sagemaker-python-sdk](https://github.com/aws/sagemaker-python-sdk) repository.\n",
    "\n",
    "For more on Scikit-learn, please visit the Scikit-learn website: <http://scikit-learn.org/stable/>.\n",
    "\n",
    "### Table of contents\n",
    "* [Upload the data for training](#upload_data)\n",
    "* [Create a Scikit-learn script to train with](#create_sklearn_script)\n",
    "* [Create the SageMaker Scikit Estimator](#create_sklearn_estimator)\n",
    "* [Train the SKLearn Estimator on the Iris data](#train_sklearn)\n",
    "* [Using the trained model to make inference requests](#inferece)\n",
    " * [Deploy the model](#deploy)\n",
    " * [Choose some data and use it for a prediction](#prediction_request)\n",
    " * [Endpoint cleanup](#endpoint_cleanup)\n",
    "* [Batch Transform](#batch_transform)\n",
    " * [Prepare Input Data](#prepare_input_data)\n",
    " * [Run Transform Job](#run_transform_job)\n",
    " * [Check Output Data](#check_output_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, lets create our Sagemaker session and role, and create a S3 prefix to use for the notebook example."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# S3 prefix\n",
    "prefix = 'Scikit-iris'\n",
    "\n",
    "import sagemaker\n",
    "from sagemaker import get_execution_role\n",
    "\n",
    "sagemaker_session = sagemaker.Session()\n",
    "\n",
    "# Get a SageMaker-compatible role used by this Notebook Instance.\n",
    "role = get_execution_role()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Upload the data for training <a class=\"anchor\" id=\"upload_data\"></a>\n",
    "\n",
    "When training large models with huge amounts of data, you'll typically use big data tools, like Amazon Athena, AWS Glue, or Amazon EMR, to create your data in S3. For the purposes of this example, we're using a sample of the classic [Iris dataset](https://en.wikipedia.org/wiki/Iris_flower_data_set), which is included with Scikit-learn. We will load the dataset, write locally, then write the dataset to s3 to use."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import os\n",
    "from sklearn import datasets\n",
    "\n",
    "# Load Iris dataset, then join labels and features\n",
    "iris = datasets.load_iris()\n",
    "joined_iris = np.insert(iris.data, 0, iris.target, axis=1)\n",
    "\n",
    "# Create directory and write csv\n",
    "os.makedirs('./data', exist_ok=True)\n",
    "np.savetxt('./data/iris.csv', joined_iris, delimiter=',', fmt='%1.1f, %1.3f, %1.3f, %1.3f, %1.3f')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once we have the data locally, we can use use the tools provided by the SageMaker Python SDK to upload the data to a default bucket. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "WORK_DIRECTORY = 'data'\n",
    "\n",
    "train_input = sagemaker_session.upload_data(WORK_DIRECTORY, key_prefix=\"{}/{}\".format(prefix, WORK_DIRECTORY) )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a Scikit-learn script to train with <a class=\"anchor\" id=\"create_sklearn_script\"></a>\n",
    "SageMaker can now run a scikit-learn script using the `SKLearn` estimator. When executed on SageMaker a number of helpful environment variables are available to access properties of the training environment, such as:\n",
    "\n",
    "* `SM_MODEL_DIR`: A string representing the path to the directory to write model artifacts to. Any artifacts saved in this folder are uploaded to S3 for model hosting after the training job completes.\n",
    "* `SM_OUTPUT_DIR`: A string representing the filesystem path to write output artifacts to. Output artifacts may include checkpoints, graphs, and other files to save, not including model artifacts. These artifacts are compressed and uploaded to S3 to the same S3 prefix as the model artifacts.\n",
    "\n",
    "Supposing two input channels, 'train' and 'test', were used in the call to the `SKLearn` estimator's `fit()` method, the following environment variables will be set, following the format `SM_CHANNEL_[channel_name]`:\n",
    "\n",
    "* `SM_CHANNEL_TRAIN`: A string representing the path to the directory containing data in the 'train' channel\n",
    "* `SM_CHANNEL_TEST`: Same as above, but for the 'test' channel.\n",
    "\n",
    "A typical training script loads data from the input channels, configures training with hyperparameters, trains a model, and saves a model to model_dir so that it can be hosted later. Hyperparameters are passed to your script as arguments and can be retrieved with an `argparse.ArgumentParser` instance. For example, the script that we will run in this notebook is the below:\n",
    "\n",
    "```python\n",
    "from __future__ import print_function\n",
    "\n",
    "import argparse\n",
    "import joblib\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "from sklearn import tree\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    parser = argparse.ArgumentParser()\n",
    "\n",
    "    # Hyperparameters are described here. In this simple example we are just including one hyperparameter.\n",
    "    parser.add_argument('--max_leaf_nodes', type=int, default=-1)\n",
    "\n",
    "    # Sagemaker specific arguments. Defaults are set in the environment variables.\n",
    "    parser.add_argument('--output-data-dir', type=str, default=os.environ['SM_OUTPUT_DATA_DIR'])\n",
    "    parser.add_argument('--model-dir', type=str, default=os.environ['SM_MODEL_DIR'])\n",
    "    parser.add_argument('--train', type=str, default=os.environ['SM_CHANNEL_TRAIN'])\n",
    "\n",
    "    args = parser.parse_args()\n",
    "\n",
    "    # Take the set of files and read them all into a single pandas dataframe\n",
    "    input_files = [ os.path.join(args.train, file) for file in os.listdir(args.train) ]\n",
    "    if len(input_files) == 0:\n",
    "        raise ValueError(('There are no files in {}.\\n' +\n",
    "                          'This usually indicates that the channel ({}) was incorrectly specified,\\n' +\n",
    "                          'the data specification in S3 was incorrectly specified or the role specified\\n' +\n",
    "                          'does not have permission to access the data.').format(args.train, \"train\"))\n",
    "    raw_data = [ pd.read_csv(file, header=None, engine=\"python\") for file in input_files ]\n",
    "    train_data = pd.concat(raw_data)\n",
    "\n",
    "    # labels are in the first column\n",
    "    train_y = train_data.iloc[:, 0]\n",
    "    train_X = train_data.iloc[:, 1:]\n",
    "\n",
    "    # Here we support a single hyperparameter, 'max_leaf_nodes'. Note that you can add as many\n",
    "    # as your training my require in the ArgumentParser above.\n",
    "    max_leaf_nodes = args.max_leaf_nodes\n",
    "\n",
    "    # Now use scikit-learn's decision tree classifier to train the model.\n",
    "    clf = tree.DecisionTreeClassifier(max_leaf_nodes=max_leaf_nodes)\n",
    "    clf = clf.fit(train_X, train_y)\n",
    "\n",
    "    # Print the coefficients of the trained classifier, and save the coefficients\n",
    "    joblib.dump(clf, os.path.join(args.model_dir, \"model.joblib\"))\n",
    "\n",
    "\n",
    "def model_fn(model_dir):\n",
    "    \"\"\"Deserialized and return fitted model\n",
    "    \n",
    "    Note that this should have the same name as the serialized model in the main method\n",
    "    \"\"\"\n",
    "    clf = joblib.load(os.path.join(model_dir, \"model.joblib\"))\n",
    "    return clf\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Because the Scikit-learn container imports your training script, you should always put your training code in a main guard `(if __name__=='__main__':)` so that the container does not inadvertently run your training code at the wrong point in execution.\n",
    "\n",
    "For more information about training environment variables, please visit https://github.com/aws/sagemaker-containers."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create SageMaker Scikit Estimator <a class=\"anchor\" id=\"create_sklearn_estimator\"></a>\n",
    "\n",
    "To run our Scikit-learn training script on SageMaker, we construct a `sagemaker.sklearn.estimator.sklearn` estimator, which accepts several constructor arguments:\n",
    "\n",
    "* __entry_point__: The path to the Python script SageMaker runs for training and prediction.\n",
    "* __role__: Role ARN\n",
    "* __train_instance_type__ *(optional)*: The type of SageMaker instances for training. __Note__: Because Scikit-learn does not natively support GPU training, Sagemaker Scikit-learn does not currently support training on GPU instance types.\n",
    "* __sagemaker_session__ *(optional)*: The session used to train on Sagemaker.\n",
    "* __hyperparameters__ *(optional)*: A dictionary passed to the train function as hyperparameters.\n",
    "\n",
    "To see the code for the SKLearn Estimator, see here: https://github.com/aws/sagemaker-python-sdk/tree/master/src/sagemaker/sklearn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.sklearn.estimator import SKLearn\n",
    "\n",
    "FRAMEWORK_VERSION = \"0.23-1\"\n",
    "script_path = 'scikit_learn_iris.py'\n",
    "\n",
    "sklearn = SKLearn(\n",
    "    entry_point=script_path,\n",
    "    framework_version=FRAMEWORK_VERSION,\n",
    "    train_instance_type=\"ml.c4.xlarge\",\n",
    "    role=role,\n",
    "    sagemaker_session=sagemaker_session,\n",
    "    hyperparameters={'max_leaf_nodes': 30})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train SKLearn Estimator on Iris data <a class=\"anchor\" id=\"train_sklearn\"></a>\n",
    "Training is very simple, just call `fit` on the Estimator! This will start a SageMaker Training job that will download the data for us, invoke our scikit-learn code (in the provided script file), and save any model artifacts that the script creates."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "sklearn.fit({'train': train_input})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using the trained model to make inference requests <a class=\"anchor\" id=\"inference\"></a>\n",
    "\n",
    "### Deploy the model <a class=\"anchor\" id=\"deploy\"></a>\n",
    "\n",
    "Deploying the model to SageMaker hosting just requires a `deploy` call on the fitted model. This call takes an instance count and instance type."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictor = sklearn.deploy(initial_instance_count=1, instance_type=\"ml.m5.xlarge\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Choose some data and use it for a prediction <a class=\"anchor\" id=\"prediction_request\"></a>\n",
    "\n",
    "In order to do some predictions, we'll extract some of the data we used for training and do predictions against it. This is, of course, bad statistical practice, but a good way to see how the mechanism works."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import itertools\n",
    "import pandas as pd\n",
    "\n",
    "shape = pd.read_csv(\"data/iris.csv\", header=None)\n",
    "\n",
    "a = [50*i for i in range(3)]\n",
    "b = [40+i for i in range(10)]\n",
    "indices = [i+j for i,j in itertools.product(a,b)]\n",
    "\n",
    "test_data = shape.iloc[indices[:-1]]\n",
    "test_X = test_data.iloc[:,1:]\n",
    "test_y = test_data.iloc[:,0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Prediction is as easy as calling predict with the predictor we got back from deploy and the data we want to do predictions with. The output from the endpoint return an numerical representation of the classification prediction; in the original dataset, these are flower names, but in this example the labels are numerical. We can compare against the original label that we parsed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(predictor.predict(test_X.values))\n",
    "print(test_y.values)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Endpoint cleanup <a class=\"anchor\" id=\"endpoint_cleanup\"></a>\n",
    "\n",
    "When you're done with the endpoint, you'll want to clean it up."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sklearn.delete_endpoint()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Batch Transform <a class=\"anchor\" id=\"batch_transform\"></a>\n",
    "We can also use the trained model for asynchronous batch inference on S3 data using SageMaker Batch Transform."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a SKLearn Transformer from the trained SKLearn Estimator\n",
    "transformer = sklearn.transformer(instance_count=1, instance_type='ml.m5.xlarge')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Prepare Input Data <a class=\"anchor\" id=\"prepare_input_data\"></a>\n",
    "We will extract 10 random samples of 100 rows from the training data, then split the features (X) from the labels (Y). Then upload the input data to a given location in S3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "# Randomly sample the iris dataset 10 times, then split X and Y\n",
    "mkdir -p batch_data/XY batch_data/X batch_data/Y\n",
    "for i in {0..9}; do\n",
    "    cat data/iris.csv | shuf -n 100 > batch_data/XY/iris_sample_${i}.csv\n",
    "    cat batch_data/XY/iris_sample_${i}.csv | cut -d',' -f2- > batch_data/X/iris_sample_X_${i}.csv\n",
    "    cat batch_data/XY/iris_sample_${i}.csv | cut -d',' -f1 > batch_data/Y/iris_sample_Y_${i}.csv\n",
    "done"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Upload input data from local filesystem to S3\n",
    "batch_input_s3 = sagemaker_session.upload_data('batch_data/X', key_prefix=prefix + '/batch_input')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Run Transform Job <a class=\"anchor\" id=\"run_transform_job\"></a>\n",
    "Using the Transformer, run a transform job on the S3 input data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start a transform job and wait for it to finish\n",
    "transformer.transform(batch_input_s3, content_type='text/csv')\n",
    "print('Waiting for transform job: ' + transformer.latest_transform_job.job_name)\n",
    "transformer.wait()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check Output Data  <a class=\"anchor\" id=\"check_output_data\"></a>\n",
    "After the transform job has completed, download the output data from S3. For each file \"f\" in the input data, we have a corresponding file \"f.out\" containing the predicted labels from each input row. We can compare the predicted labels to the true labels saved earlier."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Download the output data from S3 to local filesystem\n",
    "batch_output = transformer.output_path\n",
    "!mkdir -p batch_data/output\n",
    "!aws s3 cp --recursive $batch_output/ batch_data/output/\n",
    "# Head to see what the batch output looks like\n",
    "!head batch_data/output/*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "# For each sample file, compare the predicted labels from batch output to the true labels\n",
    "for i in {1..9}; do\n",
    "    diff -s batch_data/Y/iris_sample_Y_${i}.csv \\\n",
    "        <(cat batch_data/output/iris_sample_X_${i}.csv.out | sed 's/[[\"]//g' | sed 's/, \\|]/\\n/g') \\\n",
    "        | sed \"s/\\/dev\\/fd\\/63/batch_data\\/output\\/iris_sample_X_${i}.csv.out/\"\n",
    "done"
   ]
  }
 ],
 "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"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}