xgboost-workflow-1-data-prep.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## XGBoost Complete Project Workflow in Amazon SageMaker\n",
    "### Data Preprocessing\n",
    "    \n",
    "1. [Introduction](#Introduction)\n",
    "2. [SageMaker Processing for dataset transformation](#SageMakerProcessing)\n",
    "    \n",
    "## Introduction <a class=\"anchor\" id=\"Introduction\">\n",
    "\n",
    "If you are using XGBoost, you can use the Amazon SageMaker prebuilt XGBoost container. Using SageMaker's XGBoost and other SageMaker features, you can build a complete workflow for an XGBoost project. These notebooks present such a workflow, including all key steps such as preprocessing data with SageMaker Processing, code prototyping with SageMaker Local Mode training and inference, and production-ready model training and deployment with SageMaker hosted training and inference. Automatic Model Tuning in SageMaker is used to tune the model's hyperparameters.  Additionally, the [AWS Step Functions Data Science SDK](https://aws-step-functions-data-science-sdk.readthedocs.io/en/latest/readmelink.html) is used to automate the main training and deployment steps for use in a production workflow outside notebooks.    \n",
    "\n",
    "To enable you to run these notebooks within a reasonable time (typically less than an hour), the use case is a straightforward regression task:  predicting house prices based on the well-known Boston Housing dataset. This public dataset contains 13 features regarding housing stock of towns in the Boston area.  Features include average number of rooms, accessibility to radial highways, adjacency to the Charles River, etc.  \n",
    "\n",
    "To begin, we'll import some necessary packages and set up directories for local training and test data.  We'll also set up a SageMaker Session to perform various operations, and specify an Amazon S3 bucket to hold input data and output.  The default bucket used here is created by SageMaker if it doesn't already exist, and named in accordance with the AWS account ID and AWS Region.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import sagemaker\n",
    "import random\n",
    "import boto3\n",
    "\n",
    "# seed used by Tensorflow's boston housing data set\n",
    "random.seed(113)\n",
    "\n",
    "sess = sagemaker.Session()\n",
    "bucket = sess.default_bucket()\n",
    "\n",
    "try:\n",
    "    role = sagemaker.get_execution_role()\n",
    "except ValueError:\n",
    "    iam = boto3.client('iam')\n",
    "    role = iam.get_role(RoleName='AmazonSageMaker-ExecutionRole-20200630T141851')['Role']['Arn']\n",
    "\n",
    "data_dir = os.path.join(os.getcwd(), 'data')\n",
    "os.makedirs(data_dir, exist_ok=True)\n",
    "\n",
    "train_dir = os.path.join(os.getcwd(), 'data/train')\n",
    "os.makedirs(train_dir, exist_ok=True)\n",
    "\n",
    "test_dir = os.path.join(os.getcwd(), 'data/test')\n",
    "os.makedirs(test_dir, exist_ok=True)\n",
    "\n",
    "train_dir_csv = os.path.join(os.getcwd(), 'data/train_csv')\n",
    "os.makedirs(train_dir_csv, exist_ok=True)\n",
    "\n",
    "test_dir_csv = os.path.join(os.getcwd(), 'data/test_csv')\n",
    "os.makedirs(test_dir_csv, exist_ok=True)\n",
    "\n",
    "raw_dir = os.path.join(os.getcwd(), 'data/raw')\n",
    "os.makedirs(raw_dir, exist_ok=True)\n",
    "\n",
    "model_dir = os.path.join(os.getcwd(), 'model')\n",
    "os.makedirs(model_dir, exist_ok=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll also import and instantiate a helper class to help us store variables as we go through these notebooks so we can pick up where we left off, even if the notebook kernels get restarted."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from parameter_store import ParameterStore\n",
    "ps = ParameterStore()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SageMaker Processing for dataset transformation <a class=\"anchor\" id=\"SageMakerProcessing\">\n",
    "\n",
    "Next, we'll import the dataset and transform it with SageMaker Processing, which can be used to process terabytes of data in a SageMaker-managed cluster separate from the instance running your notebook server. In a typical SageMaker workflow, notebooks are only used for prototyping and can be run on relatively inexpensive and less powerful instances, while processing, training and model hosting tasks are run on separate, more powerful SageMaker-managed instances.  SageMaker Processing includes off-the-shelf support for Scikit-learn, as well as a Bring Your Own Container option, so it can be used with many different data transformation technologies and tasks.    \n",
    "\n",
    "First we'll load the Boston Housing dataset, save the raw feature data and upload it to Amazon S3 for transformation by SageMaker Processing.  We'll also save the labels for training and testing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import math\n",
    "from sklearn.datasets import load_boston\n",
    "from sklearn.preprocessing import StandardScaler\n",
    "\n",
    "\n",
    "boston = load_boston()\n",
    "x, y = boston['data'], boston['target']\n",
    "training_index = math.floor(.8 * boston['data'].shape[0])\n",
    "x_train, y_train = x[:training_index], y[:training_index]\n",
    "x_test, y_test = x[training_index:], y[training_index:]\n",
    "\n",
    "np.save(os.path.join(raw_dir, 'x_train.npy'), x_train)\n",
    "np.save(os.path.join(raw_dir, 'x_test.npy'), x_test)\n",
    "np.save(os.path.join(train_dir, 'y_train.npy'), y_train)\n",
    "np.save(os.path.join(test_dir, 'y_test.npy'), y_test)\n",
    "s3_prefix = 'xgboost-workflow'\n",
    "rawdata_s3_prefix = '{}/data/raw'.format(s3_prefix)\n",
    "raw_s3 = sess.upload_data(path='./data/raw/', key_prefix=rawdata_s3_prefix)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To use SageMaker Processing, simply supply a Python data preprocessing script as shown below.  For this example, we're using a SageMaker prebuilt Scikit-learn container, which includes many common functions for processing data.  There are few limitations on what kinds of code and operations you can run, and only a minimal contract:  input and output data must be placed in specified directories.  If this is done, SageMaker Processing automatically loads the input data from S3 and uploads transformed data back to S3 when the job is complete."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting preprocessing.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile preprocessing.py\n",
    "\n",
    "import glob\n",
    "import numpy as np\n",
    "import os\n",
    "from sklearn.preprocessing import StandardScaler\n",
    "\n",
    "if __name__=='__main__':\n",
    "    \n",
    "    input_files = glob.glob('{}/*.npy'.format('/opt/ml/processing/input'))\n",
    "    print('\\nINPUT FILE LIST: \\n{}\\n'.format(input_files))\n",
    "    scaler = StandardScaler()\n",
    "    for file in input_files:\n",
    "        raw = np.load(file)\n",
    "        transformed = scaler.fit_transform(raw)\n",
    "        if 'train' in file:\n",
    "            output_path = os.path.join('/opt/ml/processing/train', 'x_train.npy')\n",
    "            np.save(output_path, transformed)\n",
    "            print('SAVED TRANSFORMED TRAINING DATA FILE\\n')\n",
    "        else:\n",
    "            output_path = os.path.join('/opt/ml/processing/test', 'x_test.npy')\n",
    "            np.save(output_path, transformed)\n",
    "            print('SAVED TRANSFORMED TEST DATA FILE\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Before starting the SageMaker Processing job, we instantiate a `SKLearnProcessor` object.  This object allows you to specify the instance type to use in the job, as well as how many instances.  Although the Boston Housing dataset is quite small, we'll use two instances to showcase how easy it is to spin up a cluster for SageMaker Processing.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.sklearn.processing import SKLearnProcessor\n",
    "\n",
    "sklearn_processor = SKLearnProcessor(framework_version='0.20.0',\n",
    "                                     role=role,\n",
    "                                     instance_type='ml.m5.xlarge',\n",
    "                                     instance_count=2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We're now ready to run the Processing job.  To enable distributing the data files equally among the instances, we specify the `ShardedByS3Key` distribution type in the `ProcessingInput` object.  This ensures that if we have `n` instances, each instance will receive `1/n` files from the specified S3 bucket.  It may take around 3 minutes for the following code cell to run, mainly to set up the cluster.  At the end of the job, the cluster automatically will be torn down by SageMaker.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Parameter 'session' will be renamed to 'sagemaker_session' in SageMaker Python SDK v2.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Job Name:  xgboost-workflow-preprocessing-23-15-57-58\n",
      "Inputs:  [{'InputName': 'input-1', 'S3Input': {'S3Uri': 's3://sagemaker-us-west-2-781297242028/xgboost-workflow/data/raw', 'LocalPath': '/opt/ml/processing/input', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'ShardedByS3Key', 'S3CompressionType': 'None'}}, {'InputName': 'code', 'S3Input': {'S3Uri': 's3://sagemaker-us-west-2-781297242028/xgboost-workflow-preprocessing-23-15-57-58/input/code/preprocessing.py', 'LocalPath': '/opt/ml/processing/input/code', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'FullyReplicated', 'S3CompressionType': 'None'}}]\n",
      "Outputs:  [{'OutputName': 'train', 'S3Output': {'S3Uri': 's3://sagemaker-us-west-2-781297242028/xgboost-workflow/data/train', 'LocalPath': '/opt/ml/processing/train', 'S3UploadMode': 'EndOfJob'}}, {'OutputName': 'test', 'S3Output': {'S3Uri': 's3://sagemaker-us-west-2-781297242028/xgboost-workflow/data/test', 'LocalPath': '/opt/ml/processing/test', 'S3UploadMode': 'EndOfJob'}}]\n",
      "........................\n",
      "\u001b[34m/miniconda3/lib/python3.7/site-packages/sklearn/externals/joblib/externals/cloudpickle/cloudpickle.py:47: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses\n",
      "  import imp\n",
      "\u001b[0m\n",
      "\u001b[34mINPUT FILE LIST: \u001b[0m\n",
      "\u001b[34m['/opt/ml/processing/input/x_test.npy']\n",
      "\u001b[0m\n",
      "\u001b[34mSAVED TRANSFORMED TEST DATA FILE\n",
      "\u001b[0m\n",
      "\u001b[35m/miniconda3/lib/python3.7/site-packages/sklearn/externals/joblib/externals/cloudpickle/cloudpickle.py:47: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses\n",
      "  import imp\n",
      "\u001b[0m\n",
      "\u001b[35mINPUT FILE LIST: \u001b[0m\n",
      "\u001b[35m['/opt/ml/processing/input/x_train.npy']\n",
      "\u001b[0m\n",
      "\u001b[35mSAVED TRANSFORMED TRAINING DATA FILE\n",
      "\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "from sagemaker.processing import ProcessingInput, ProcessingOutput\n",
    "from time import gmtime, strftime \n",
    "\n",
    "processing_job_name = \"xgboost-workflow-preprocessing-{}\".format(strftime(\"%d-%H-%M-%S\", gmtime()))\n",
    "output_destination = 's3://{}/{}/data'.format(bucket, s3_prefix)\n",
    "\n",
    "# code=can be a s3 uri for the input script\n",
    "sklearn_processor.run(code='preprocessing.py',\n",
    "                      job_name=processing_job_name,\n",
    "                      inputs=[ProcessingInput(\n",
    "                        source=raw_s3,\n",
    "                        destination='/opt/ml/processing/input',\n",
    "                        s3_data_distribution_type='ShardedByS3Key')],\n",
    "                      outputs=[ProcessingOutput(output_name='train',\n",
    "                                                destination='{}/train'.format(output_destination),\n",
    "                                                source='/opt/ml/processing/train'),\n",
    "                               ProcessingOutput(output_name='test',\n",
    "                                                destination='{}/test'.format(output_destination),\n",
    "                                                source='/opt/ml/processing/test')])\n",
    "\n",
    "preprocessing_job_description = sklearn_processor.jobs[-1].describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the log output of the SageMaker Processing job above, you should be able to see logs in two different colors for the two different instances, and that each instance received different files.  Without the `ShardedByS3Key` distribution type, each instance would have received a copy of **all** files.  By spreading the data equally among `n` instances, you should receive a speedup by approximately a factor of `n` for most stateless data transformations.  After saving the job results locally, we'll move on to prototyping training and inference code with Local Mode."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "download: s3://sagemaker-us-west-2-781297242028/xgboost-workflow/data/train/x_train.npy to data/train/x_train.npy\n",
      "download: s3://sagemaker-us-west-2-781297242028/xgboost-workflow/data/test/x_test.npy to data/test/x_test.npy\n"
     ]
    }
   ],
   "source": [
    "train_in_s3 = '{}/train/x_train.npy'.format(output_destination)\n",
    "test_in_s3 = '{}/test/x_test.npy'.format(output_destination)\n",
    "!aws s3 cp {train_in_s3} ./data/train/x_train.npy\n",
    "!aws s3 cp {test_in_s3} ./data/test/x_test.npy"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll use the data and artifacts created in this notebook in downstream notebooks for model training and deployment. Store them here for later retrieval."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ps.update({'bucket': bucket, 's3_prefix': s3_prefix, 'raw_s3': raw_s3, 'train_dir': train_dir,\n",
    "          'test_dir': test_dir, 'role': role, 'train_dir_csv': train_dir_csv,\n",
    "          'test_dir_csv': test_dir_csv})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ps.store()"
   ]
  }
 ],
 "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.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}