2020-11-10 Sagemaker JSON to XGBoost Model Example.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Deploying an XGBoost Model to a SageMaker Endpoint that Accepts JSON Payloads"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook shows how to deploy an XGBoost model in an Amazon SageMaker Endpoint that accepts a JSON payload. \n",
    "\n",
    "For that purpose two [SKlearn model](https://sagemaker.readthedocs.io/en/stable/frameworks/sklearn/sagemaker.sklearn.html) models are built:\n",
    "\n",
    "* A transformer model that converts JSON payloads into numeric arrays\n",
    "* A wrapper for the XGBoost model\n",
    "\n",
    "\n",
    "These two models are combined into a [PipelineModel](https://sagemaker.readthedocs.io/en/stable/api/inference/pipeline.html) and deployed to a SageMaker endpoint.\n",
    "\n",
    "The approach is inspired by the pre-processing example found [here](https://github.com/aws/amazon-sagemaker-examples/blob/master/sagemaker-python-sdk/scikit_learn_inference_pipeline/Inference%20Pipeline%20with%20Scikit-learn%20and%20Linear%20Learner.ipynb).\n",
    "\n",
    "To run this notebook please set up a conda environment with the specifications found in the *environment.yaml* in your Jupyter Notebook instance in Amazon SageMaker"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Imports & Accessing SageMaker"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import sagemaker\n",
    "import pickle\n",
    "\n",
    "import scipy\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import xgboost as xgb\n",
    "\n",
    "from time import gmtime, strftime\n",
    "\n",
    "from sklearn.metrics import roc_auc_score, log_loss\n",
    "from sklearn.datasets import fetch_openml\n",
    "\n",
    "from sagemaker.sklearn.estimator import SKLearn\n",
    "from sagemaker.pipeline import PipelineModel\n",
    "from sagemaker.predictor import Predictor\n",
    "from sagemaker.serializers import JSONSerializer\n",
    "from sagemaker.deserializers import CSVDeserializer\n",
    "\n",
    "sagemaker_session = sagemaker.Session()\n",
    "role = sagemaker.get_execution_role()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Downloading & Pre-Processing Example Training Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We use the [*adult*](https://www.openml.org/d/1590) dataset from OpenML which contains a mix of categorical and continuous predictors."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "adult_data = fetch_openml(name = 'adult',as_frame = True,version = 1)\n",
    "X = adult_data['data'].drop(columns = ['education'])\n",
    "y = adult_data['target'].apply(lambda x: 0 if x == \"<=50K\" else 1).astype(float)\n",
    "\n",
    "continuous_vars = X.dtypes[(X.dtypes == \"float\")].index\n",
    "categorical_vars = X.dtypes[(X.dtypes == \"category\")].index\n",
    "\n",
    "# introduce new category for missing values in categorical variables\n",
    "X[categorical_vars] = X[categorical_vars].apply(lambda x: x.cat.add_categories(\"missing\")).fillna(\"missing\").astype(str)\n",
    "\n",
    "pd.concat([X,y],axis =1).head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# dummify categorical variables\n",
    "categorical_dummified = pd.get_dummies(X[categorical_vars])\n",
    "data_train = pd.concat([X.drop(columns = categorical_vars),categorical_dummified,y],axis = 1)\n",
    "feature_names = data_train.drop(columns = ['class']).columns.tolist()\n",
    "data_train.head(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Training the XGBoost model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this step we train the XGBoost model and write it out to a binary file that can be read-in later when deploying the model (no cross-validation step here or hyperparameter tuning to keep the example simple)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "xgboost_params =  {\n",
    "    \"objective\": \"binary:logistic\",\n",
    "    \"eval_metric\": \"logloss\",\n",
    "    \"max_depth\": 11,\n",
    "    \"eta\": 0.052,\n",
    "    \"nthread\": 4,\n",
    "    \"booster\": \"gbtree\",\n",
    "    \"subsample\": 0.87,\n",
    "    \"min_child_weight\": 2,\n",
    "    \"colsample_bytree\": 0.71,\n",
    "    \"colsample_bylevel\": 0.64\n",
    "  }\n",
    "    \n",
    "dtrain = xgb.DMatrix(scipy.sparse.csr_matrix(data_train[feature_names]),label = data_train['class'].to_numpy(),feature_names = feature_names) \n",
    "xgb_model = xgb.train(dtrain = dtrain, params = xgboost_params,num_boost_round = 500)\n",
    "\n",
    "print(\"log loss: {0}\".format(log_loss(data_train['class'],xgb_model.predict(dtrain))))\n",
    "print(\"AUC: {0}\".format(roc_auc_score(data_train['class'],xgb_model.predict(dtrain))))\n",
    "\n",
    "xgb_model.save_model(\"sklearn_xgboost_wrapper/xgboost_model.bin\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating the JSON Transformer Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to make predictions with the XGBoost we need to transform the JSON payload into a numeric array. The entries of the array must be corresponding to the columns in the training dataset.\n",
    "\n",
    "For this purpose we wrote a helper class (see file *sklearn_json_transformer/json_parser.py*) which needs to be initialised with specific information about the training data (categorical/continuous variables, observed values of categorical variables, order of features in the training data).\n",
    "\n",
    "This information is persisted as a pickled dictionary that will be passed on to the model creation process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "categorical_features_values = {k: X[k].unique().tolist() for k in categorical_vars}\n",
    "feature_definitions = {'continuous_variables': continuous_vars.tolist(),\n",
    "                      'categorical_variables': categorical_vars.tolist(),\n",
    "                      'target_columns': feature_names,\n",
    "                      'categorical_variables_values': categorical_features_values}\n",
    "pickle.dump(feature_definitions, open(\"sklearn_json_transformer/features.pkl\", \"wb\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sklearn_json_transformer = SKLearn(\n",
    "            entry_point = \"json_transformer.py\",\n",
    "            role = role,\n",
    "            instance_type = 'ml.c4.xlarge',\n",
    "            sagemaker_session = sagemaker_session,\n",
    "            source_dir = 'sklearn_json_transformer',\n",
    "            framework_version = \"0.23-1\")\n",
    "sklearn_json_transformer.fit()\n",
    "sklearn_json_transformer = sklearn_json_transformer.create_model(env={\"SAGEMAKER_DEFAULT_INVOCATIONS_ACCEPT\": \"text/csv\"})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating the XGBoost Wrapper Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the same way as the JSON transformer the XGBoost model will be provided as an SKLearn model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sklearn_xgboost_wrapper_model = SKLearn(entry_point = \"xgboost_wrapper.py\",\n",
    "                                        role = role,\n",
    "                                        instance_type = \"ml.c5.xlarge\",\n",
    "                                        sagemaker_session = sagemaker_session,\n",
    "                                        source_dir = \"sklearn_xgboost_wrapper\",\n",
    "                                        py_version = \"py3\",\n",
    "                                        framework_version = \"0.23-1\")\n",
    "sklearn_xgboost_wrapper_model.fit()\n",
    "xgboost_wrapper_model = sklearn_xgboost_wrapper_model.create_model(env={\"SAGEMAKER_DEFAULT_INVOCATIONS_ACCEPT\": \"text/csv\"})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Deploying the Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "timestamp_prefix = strftime(\"%Y-%m-%d-%H-%M-%S\", gmtime())\n",
    "\n",
    "model_name = 'json-to-xgboost-pipeline-' + timestamp_prefix\n",
    "endpoint_name = 'json-to-xgboost-pipeline-ep-' + timestamp_prefix\n",
    "sm_model = PipelineModel(\n",
    "    name=model_name, \n",
    "    role=role,\n",
    "    models=[\n",
    "        sklearn_json_transformer, \n",
    "        xgboost_wrapper_model\n",
    "    ])\n",
    "\n",
    "sm_model.deploy(initial_instance_count=1, instance_type='ml.c4.xlarge', endpoint_name=endpoint_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Testing the Endpoint\n",
    "\n",
    "To make sure the deployment has succeeded and the JSON transformation is working correctly we generate some JSON payloads from the training data and compare the results of the local version of the model and the endpoint."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_observations = X.sample(n=100)\n",
    "test_predictions = pd.DataFrame({\"prediction_offline\":xgb_model.predict(xgb.DMatrix(scipy.sparse.csr_matrix(data_train.iloc[sample_observations.index].drop(columns = ['class'])),feature_names = feature_names))})\n",
    "predictor_test = Predictor(\n",
    "            endpoint_name=endpoint_name,\n",
    "            sagemaker_session=sagemaker_session,\n",
    "            serializer=JSONSerializer(),\n",
    "            deserializer=CSVDeserializer())\n",
    "test_predictions = pd.concat([test_predictions,pd.DataFrame({\"prediction_endpoint\":[float(p[0]) for p in  predictor_test.predict(sample_observations.T.apply(lambda x: dict(x)).tolist())]})],axis = 1)\n",
    "assert np.allclose(test_predictions['prediction_offline'],test_predictions['prediction_endpoint'],atol = 1e-10), \"Mismatch between offline and endpoint predictions\"\n",
    "test_predictions.head(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Tidying up\n",
    "\n",
    "Delete the endpoint to avoid unnecessary costs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sm_client = sagemaker_session.boto_session.client('sagemaker')\n",
    "sm_client.delete_endpoint(EndpointName=endpoint_name)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "aws_blogpost",
   "language": "python",
   "name": "aws_blogpost"
  },
  "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
}