lab/02_TrainYourModel/01_Training our model.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Now, it is time for start an automated ML pipeline using the MLOps environment\n",
    "\n",
    "We'll do that by putting a zip file, called **trainingjob.zip**, in an S3 bucket. CodePipeline will listen to that bucket and start a job. This zip file has the following structure:\n",
    " - trainingjob.json (Sagemaker training job descriptor)\n",
    " - environment.json (Instructions to the environment of how to deploy and prepare the endpoints)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1 Let's start defining the hyperparameters and other attributes\n",
    "\n",
    "If you ran the previous section **01_CreateAlgorithmContainer** and managed to create a custom container, please change the following variable in the next cell, from:  \n",
    "```Python\n",
    "use_xgboost_builtin=True\n",
    "```\n",
    "to:\n",
    "```Python\n",
    "use_xgboost_builtin=False\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sagemaker\n",
    "import boto3\n",
    "\n",
    "use_xgboost_builtin=True\n",
    "\n",
    "sts_client = boto3.client(\"sts\")\n",
    "account_id = sts_client.get_caller_identity()[\"Account\"]\n",
    "region = boto3.session.Session().region_name\n",
    "model_prefix='iris-model'\n",
    "training_image = None\n",
    "hyperparameters = None\n",
    "if use_xgboost_builtin: \n",
    "    training_image = sagemaker.image_uris.retrieve('xgboost', boto3.Session().region_name, version='1.0-1')\n",
    "    hyperparameters = {\n",
    "        \"alpha\": 0.42495142279951414,\n",
    "        \"eta\": 0.4307531922567607,\n",
    "        \"gamma\": 1.8028358018081714,\n",
    "        \"max_depth\": 10,\n",
    "        \"min_child_weight\": 5.925133573560345,\n",
    "        \"num_class\": 3,\n",
    "        \"num_round\": 30,\n",
    "        \"objective\": \"multi:softmax\",\n",
    "        \"reg_lambda\": 10,\n",
    "        \"silent\": 0,\n",
    "    }\n",
    "else:\n",
    "    training_image = '{}.dkr.ecr.{}.amazonaws.com/{}:latest'.format(account_id, region, model_prefix)\n",
    "    hyperparameters = {\n",
    "        \"max_depth\": 11,\n",
    "        \"n_jobs\": 5,\n",
    "        \"n_estimators\": 120\n",
    "    }\n",
    "print(training_image)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.2 Then, let's  create the trainingjob descriptor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import time\n",
    "import sagemaker\n",
    "import boto3\n",
    "\n",
    "roleArn = \"arn:aws:iam::{}:role/MLOps\".format(account_id)\n",
    "timestamp = time.strftime('-%Y-%m-%d-%H-%M-%S', time.gmtime())\n",
    "job_name = model_prefix + timestamp\n",
    "sagemaker_session = sagemaker.Session()\n",
    "\n",
    "training_params = {}\n",
    "\n",
    "# Here we set the reference for the Image Classification Docker image, stored on ECR (https://aws.amazon.com/pt/ecr/)\n",
    "training_params[\"AlgorithmSpecification\"] = {\n",
    "    \"TrainingImage\": training_image,\n",
    "    \"TrainingInputMode\": \"File\"\n",
    "}\n",
    "\n",
    "# The IAM role with all the permissions given to Sagemaker\n",
    "training_params[\"RoleArn\"] = roleArn\n",
    "\n",
    "# Here Sagemaker will store the final trained model\n",
    "training_params[\"OutputDataConfig\"] = {\n",
    "    \"S3OutputPath\": 's3://{}/{}'.format(sagemaker_session.default_bucket(), model_prefix)\n",
    "}\n",
    "\n",
    "# This is the config of the instance that will execute the training\n",
    "training_params[\"ResourceConfig\"] = {\n",
    "    \"InstanceCount\": 1,\n",
    "    \"InstanceType\": \"ml.m4.xlarge\",\n",
    "    \"VolumeSizeInGB\": 30\n",
    "}\n",
    "\n",
    "# The job name. You'll see this name in the Jobs section of the Sagemaker's console\n",
    "training_params[\"TrainingJobName\"] = job_name\n",
    "\n",
    "for i in hyperparameters:\n",
    "    hyperparameters[i] = str(hyperparameters[i])\n",
    "    \n",
    "# Here you will configure the hyperparameters used for training your model.\n",
    "training_params[\"HyperParameters\"] = hyperparameters\n",
    "\n",
    "# Training timeout\n",
    "training_params[\"StoppingCondition\"] = {\n",
    "    \"MaxRuntimeInSeconds\": 360000\n",
    "}\n",
    "\n",
    "# The algorithm currently only supports fullyreplicated model (where data is copied onto each machine)\n",
    "training_params[\"InputDataConfig\"] = []\n",
    "\n",
    "# Please notice that we're using application/x-recordio for both \n",
    "# training and validation datasets, given our dataset is formated in RecordIO\n",
    "\n",
    "# Here we set training dataset\n",
    "training_params[\"InputDataConfig\"].append({\n",
    "    \"ChannelName\": \"train\",\n",
    "    \"DataSource\": {\n",
    "        \"S3DataSource\": {\n",
    "            \"S3DataType\": \"S3Prefix\",\n",
    "            \"S3Uri\": 's3://{}/{}/input/train'.format(sagemaker_session.default_bucket(), model_prefix),\n",
    "            \"S3DataDistributionType\": \"FullyReplicated\"\n",
    "        }\n",
    "    },\n",
    "    \"ContentType\": \"text/csv\",\n",
    "    \"CompressionType\": \"None\"\n",
    "})\n",
    "training_params[\"InputDataConfig\"].append({\n",
    "    \"ChannelName\": \"validation\",\n",
    "    \"DataSource\": {\n",
    "        \"S3DataSource\": {\n",
    "            \"S3DataType\": \"S3Prefix\",\n",
    "            \"S3Uri\": 's3://{}/{}/input/validation'.format(sagemaker_session.default_bucket(), model_prefix),\n",
    "            \"S3DataDistributionType\": \"FullyReplicated\"\n",
    "        }\n",
    "    },\n",
    "    \"ContentType\": \"text/csv\",\n",
    "    \"CompressionType\": \"None\"\n",
    "})\n",
    "training_params[\"Tags\"] = []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "deployment_params = {\n",
    "    \"EndpointPrefix\": model_prefix,\n",
    "    \"DevelopmentEndpoint\": {\n",
    "        # we want to enable the endpoint monitoring\n",
    "        \"InferenceMonitoring\": True,\n",
    "        # we will collect 100% of all the requests/predictions\n",
    "        \"InferenceMonitoringSampling\": 100,\n",
    "        \"InferenceMonitoringOutputBucket\": 's3://{}/{}/monitoring/dev'.format(sagemaker_session.default_bucket(), model_prefix),\n",
    "        # we don't want to enable A/B tests in development\n",
    "        \"ABTests\": False,\n",
    "        # we'll use a basic instance for testing purposes\n",
    "        \"InstanceType\": \"ml.t2.large\",\n",
    "        \"InitialInstanceCount\": 1,\n",
    "        # we don't want high availability/escalability for development\n",
    "        \"AutoScaling\": None\n",
    "    },\n",
    "    \"ProductionEndpoint\": {\n",
    "        # we want to enable the endpoint monitoring\n",
    "        \"InferenceMonitoring\": True,\n",
    "        # we will collect 100% of all the requests/predictions\n",
    "        \"InferenceMonitoringSampling\": 100,\n",
    "        \"InferenceMonitoringOutputBucket\": 's3://{}/{}/monitoring/prd'.format(sagemaker_session.default_bucket(), model_prefix),\n",
    "        # we want to do A/B tests in production\n",
    "        \"ABTests\": True,\n",
    "        # we'll use a better instance for production. CPU optimized\n",
    "        \"InstanceType\": \"ml.c5.large\",\n",
    "        \"InitialInstanceCount\": 2,\n",
    "        \"InitialVariantWeight\": 0.1,\n",
    "        # we want elasticity. at minimum 2 instances to support the endpoint and at maximum 10\n",
    "        # we'll use a threshold of 750 predictions per instance to start adding new instances or remove them\n",
    "        \"AutoScaling\": {\n",
    "            \"MinCapacity\": 2,\n",
    "            \"MaxCapacity\": 10,\n",
    "            \"TargetValue\": 200.0,\n",
    "            \"ScaleInCooldown\": 30,\n",
    "            \"ScaleOutCooldown\": 60,\n",
    "            \"PredefinedMetricType\": \"SageMakerVariantInvocationsPerInstance\"\n",
    "        }\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Preparing and uploading the dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import sagemaker\n",
    "from sklearn import datasets\n",
    "from sklearn.model_selection import train_test_split\n",
    "\n",
    "sagemaker_session = sagemaker.Session()\n",
    "iris = datasets.load_iris()\n",
    "\n",
    "X_train, X_test, y_train, y_test = train_test_split(\n",
    "    iris.data, iris.target, test_size=0.33, random_state=42, stratify=iris.target)\n",
    "np.savetxt(\"iris_train.csv\", np.column_stack((y_train, X_train)), delimiter=\",\", fmt='%0.3f')\n",
    "np.savetxt(\"iris_test.csv\", np.column_stack((y_test, X_test)), delimiter=\",\", fmt='%0.3f')\n",
    "\n",
    "# Upload the dataset to an S3 bucket\n",
    "input_train = sagemaker_session.upload_data(path='iris_train.csv', key_prefix='%s/input/train' % model_prefix)\n",
    "input_test = sagemaker_session.upload_data(path='iris_test.csv', key_prefix='%s/input/validation' % model_prefix)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.3 Alright! Now it's time to start the training process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import io\n",
    "import zipfile\n",
    "import json\n",
    "\n",
    "s3 = boto3.client('s3')\n",
    "sts_client = boto3.client(\"sts\")\n",
    "\n",
    "session = boto3.session.Session()\n",
    "\n",
    "account_id = sts_client.get_caller_identity()[\"Account\"]\n",
    "region = session.region_name\n",
    "\n",
    "bucket_name = \"mlops-%s-%s\" % (region, account_id)\n",
    "key_name = \"training_jobs/%s/trainingjob.zip\" % model_prefix\n",
    "\n",
    "zip_buffer = io.BytesIO()\n",
    "with zipfile.ZipFile(zip_buffer, 'a') as zf:\n",
    "    zf.writestr('trainingjob.json', json.dumps(training_params))\n",
    "    zf.writestr('deployment.json', json.dumps(deployment_params))\n",
    "zip_buffer.seek(0)\n",
    "\n",
    "s3.put_object(Bucket=bucket_name, Key=key_name, Body=bytearray(zip_buffer.read()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Ok, now open the AWS console in another tab and go to the CodePipeline console to see the status of our building pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> Now, click on [THIS NOTEBOOK](02_Check%20Progress%20and%20Test%20the%20endpoint.ipynb) to see the progress and test your endpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A/B TESTS\n",
    "\n",
    "If you take a look on the **deployment** parameters you'll see that we enabled the **Production** endpoint for A/B tests. To try this, just deploy the first model into production, then run the section **1.3** again. Feel free to change some hyperparameter values in the section **1.1** before starting a new training session.\n",
    "\n",
    "When publishing the second model into **Development**, the endpoint will be updated and the model will be replaced without compromising the user experience. This is the natural behavior of an Endpoint in SageMaker when you update it.\n",
    "\n",
    "After you approve the deployment into **Production**, the endpoint will be updated and a second model will be added to it. Now it's time to execute some **A/B tests**. In the **Progress** Jupyter (link above), execute the last cell (test code) to show which model answered your request. You just need to keep sending some requests to see the **Production** endpoint using both models A and B, respecting the proportion defined by the variable **InitialVariantWeight** in the deployment params.\n",
    "\n",
    "In a real life scenario you can monitor the performance of both models and then adjust the **Weight** of each model to do the full transition to the new model (and remove the old one) or to rollback the new deployment.\n",
    "\n",
    "To adjust the weight of each model (Variant Name) in an endpoint, you just need to call the following function: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sagemaker.html#SageMaker.Client.update_endpoint_weights_and_capacities"
   ]
  }
 ],
 "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
}