EMRSparkNotebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# EMR Notebook SageMaker Custom Abalone Ring Estimator\n",
    "\n",
    "1. [Setup](#Setup)\n",
    "2. [Load the Data](#Load-the-Data)\n",
    "3. [Train the Model](#Train-the-Model)\n",
    "4. [Inference Results](#Inference-Results)\n",
    "5. [Wrap-Up](#Wrap-Up)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup - You MUST specify the user specific parameters below.\n",
    "\n",
    "**Enter the SageMaker Execution Role ARN that you created earlier in the IAM console.**\n",
    "\n",
    "**Enter the region code corresponding to the region your EMR cluster is in.**\n",
    "You can look up the region code (us-east-1 for North Virginia, for example) on [this page](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.RegionsAndAvailabilityZones.html). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# *****DEFINE USER SPECIFIC PARAMETERS******\n",
    "region = ''\n",
    "sagemaker_execution_role = ''\n",
    "\n",
    "#T he number of EMR nodes to process the data.\n",
    "num_workers = 12\n",
    "\n",
    "# The location of the dataset we will be using. \n",
    "source_bucket = 'ee-assets-prod-us-east-1'\n",
    "source_key = 'modules/8560b4bd6942403b9fe3291928df2453/v1/data/abalone.csv'\n",
    "\n",
    "if (region and source_bucket and sagemaker_execution_role and num_workers):\n",
    "    print('All necessary user parameters are entered.')\n",
    "else:\n",
    "    print('Please check to make sure you entered all default parameters!')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Each EMR notebook is launched with its own Spark context (variable sc). A Spark Context is the entry point for communication with Spark. First you need to install the Python packages that you'll use throughout the notebook. EMR notebooks come with a default set of libraries for data processing. You can see which libraries are installed on the notebook by calling the Spark Context's list_packages() function. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.list_packages()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To comunicate with SageMaker you need to install notebook scoped libraries. These libraries are available only during the notebook session. After the session ends, the libraries are deleted. \n",
    "\n",
    "We install [boto3 (the AWS Python 3 SDK)](https://aws.amazon.com/sdk-for-python/) and the [high level SageMaker SDK](https://sagemaker.readthedocs.io/en/stable/). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.install_pypi_package(\"boto3==1.16.9\");\n",
    "sc.install_pypi_package('sagemaker==2.16.1');"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import sagemaker\n",
    "\n",
    "#We initiate a session for the boto3 and sagemaker APIs. The session includes information necessary to call the\n",
    "#AWS APIs, such as AWS credentials and default AWS region. For this lab we will leverage the IAM role attached to\n",
    "#the EMR notebook, so we only need to provide a region.\n",
    "boto_sess = boto3.Session(region_name=region)\n",
    "sage_sdk_session = sagemaker.Session(boto_session=boto_sess)\n",
    "bucket = sage_sdk_session.default_bucket()\n",
    "\n",
    "print('A SageMaker session was initiated! You are using {} as your S3 bucket for intermediate files.'.format(bucket))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load the Data\n",
    "\n",
    "We will use the public abalone data set from the [UC Irvine Machine Learning Repository](https://archive.ics.uci.edu/ml/datasets/Abalone)\n",
    "to train and test a regression model.\n",
    "\n",
    "   Given in the dataset is the attribute name, attribute type, the measurement unit and a\n",
    "   brief description.  The number of rings is the value to predict: either\n",
    "   as a continuous value or as a classification problem.\n",
    "   \n",
    "   The age of an abalone is the number of rings in the shell + 1.5 years. Without a model researchers must cut through the abalone shell\n",
    "   and use a microscope to count the rings. Using a model to predict rings eliminates this time consuming process.\n",
    "\n",
    "\tName\t\t\tData Type\t\tMeas.\tDescription\n",
    "\t----\t\t\t---------\t\t-----\t-----------\n",
    "\tRings\t\t\tinteger\t\t\t\t\t+1.5 gives the age in years\n",
    "\tLength\t\t\tcontinuous\t\tmm\t\tLongest shell measurement\n",
    "\tDiameter\t\tcontinuous\t\tmm\t\tperpendicular to length\n",
    "\tHeight\t\t\tcontinuous\t\tmm\t\twith meat in shell\n",
    "\tWhole weight\tcontinuous\t\tgrams\twhole abalone\n",
    "\tShucked weight\tcontinuous\t\tgrams\tweight of meat\n",
    "\tViscera weight\tcontinuous\t\tgrams\tgut weight (after bleeding)\n",
    "\tShell weight\tcontinuous\t\tgrams\tafter being dried\n",
    "\tMale\t\t\tinteger\t\t\t1/0 \t1 encodes true, 0 false\n",
    "\tFemale\t\t\tinteger\t\t\t1/0 \t1 encodes true, 0 false\n",
    "\tInfant\t\t\tinteger\t\t\t1/0 \t1 encodes true, 0 false"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, we need to copy the public files to the S3 bucket in our account."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3 = boto3.client('s3', region_name=region)\n",
    "\n",
    "# Local bucket S3 prefix to store the data under.\n",
    "local_key = 'data/abalone.csv'\n",
    "\n",
    "s3.copy(CopySource={'Bucket' : source_bucket,\n",
    "                    'Key' : source_key}, \n",
    "        Bucket=bucket, \n",
    "        Key=local_key)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read the dataset from S3 in to a Spark dataframe.\n",
    "abalone_data = spark.read.load(f's3://{bucket}/{local_key}', format='csv', inferSchema=True, header=True).repartition(num_workers)\n",
    "abalone_data.show(n=5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that the data is in Spark we can modify and enhance our data. As an example, including all four abalone weights may be unnecessary. What really matters may be the difference between the whole weight and the shell weight. Making such changes on large datasets can be done easily in Spark.\n",
    "\n",
    "Let's try adding a column that is the difference between whole weight and shell weight. Then remove the whole, shucked, weight, and shell weight columns. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "abalone_data = abalone_data.withColumn('Difference_weight', abalone_data.Whole_weight - abalone_data.Shell_weight)\n",
    "abalone_data = abalone_data.drop('Whole_weight', 'Shucked_weight', 'Viscera_weight', 'Shell_weight')\n",
    "abalone_data.show(n=5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Split the dataframe in to training and validation data.\n",
    "# The training will be used to refine our model.\n",
    "# The test data will be used to measure the model's accuracy.\n",
    "train_data, test_data = abalone_data.randomSplit([.75,.25])\n",
    "\n",
    "s3_train = f's3://{bucket}/train/'\n",
    "s3_test = f's3://{bucket}/test/'\n",
    "data_format = 'csv'\n",
    "\n",
    "# Save the data in to S3 for training by SageMaker\n",
    "train_data.write.save(s3_train, format=data_format, mode='overwrite')\n",
    "test_data.write.save(s3_test, format=data_format, mode='overwrite')\n",
    "\n",
    "print(f'Training dataset saved in {data_format} format to {s3_train}!')\n",
    "print(f'Testing dataset saved in {data_format} format to {s3_test}!')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train the Model\n",
    "SageMaker contains several common built-in algorithms. For this lab you have the choice of using either the [LinearLearner](https://docs.aws.amazon.com/sagemaker/latest/dg/linear-learner.html) or [XGBoost](https://docs.aws.amazon.com/sagemaker/latest/dg/xgboost.html) built-in algorithms. Both are regression models that estimate the number of rings on the abalone."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Uncomment the LinearLearner line to use the LinearLearner algorithm. \n",
    "model = 'xgboost'\n",
    "#model = 'linear-learner'\n",
    "\n",
    "print('The SageMaker {} model will be used.'.format(model))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following cell defines the hyperparameters for each algorithn. You may leave them as the defaults, but if you are interested you could try changing a few to see if it improves model performance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set the regularization weights. Increasing these will reduce how closely the model fits to the training data.\n",
    "l1 = .25\n",
    "l2 = .25\n",
    "\n",
    "# Hyperparameters for XGBoost algorithm\n",
    "xgboost_params = {\n",
    "    'num_round':100,\n",
    "    'objective': 'reg:linear',\n",
    "    'alpha': l1,\n",
    "    'lambda': l2\n",
    "}\n",
    "\n",
    "# Hyperparameters for LinearLearner algorithm\n",
    "linear_params = {\n",
    "    'feature_dim':len(abalone_data.columns)-1,\n",
    "    'predictor_type': 'regressor',\n",
    "    'loss': 'squared_loss',\n",
    "    'l1': l1,\n",
    "    'wd': l2\n",
    "}\n",
    "\n",
    "hyperparams = {\n",
    "    'linear-learner': linear_params,\n",
    "    'xgboost': xgboost_params\n",
    "}\n",
    "\n",
    "print('All model parameters have been set!')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.image_uris import retrieve\n",
    "\n",
    "estimator = sagemaker.estimator.Estimator(\n",
    "    image_uri=retrieve(framework=model, region=region, version='latest', py_version='py3'), \n",
    "    role=sagemaker_execution_role, \n",
    "    train_instance_count=1, \n",
    "    train_instance_type='ml.m5.large',\n",
    "    sagemaker_session=sage_sdk_session, \n",
    "    hyperparameters=hyperparams[model]\n",
    ")\n",
    "\n",
    "print('The SageMaker model was constructed with parameters: {}.'.format(estimator.hyperparameters()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we initialized the model, we can train the model by calling the fit() function. After calling fit(), SageMaker will create a training instance, train a model on the instance, save the model artifacts to S3, then take down the training instance.\n",
    "\n",
    "This usually takes about 3 minutes. \n",
    "\n",
    "(**Optional**) While you wait, you may check the model training progress through the SageMaker console by following these instructions:  \n",
    "a.\tOpen SageMaker console in AWS.  \n",
    "b.\tOn the left panel, scroll until you see ‘training jobs’ beneath the ‘Training’ section.  \n",
    "c.\tClick into the job to examine further details; wait until you see the status change to ‘Completed’.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_channel = sagemaker.session.s3_input(s3_train + 'part', content_type='text/csv')\n",
    "estimator.fit({'train': train_channel})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Inference Results"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "How well did our model perform? Let's see how it does on the test data set we saved to S3 earlier. We'll use [SageMaker batch transform](https://docs.aws.amazon.com/sagemaker/latest/dg/batch-transform.html) to run our test data set through our model. Batch transform creates a SageMaker instance, deploys the model, runs the dataset through the model, then takes down the instance. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_inference = s3_train.replace('train', 'inference')\n",
    "\n",
    "transformer = estimator.transformer(\n",
    "    instance_count = 1,\n",
    "    instance_type = 'ml.m5.large',\n",
    "    strategy = 'MultiRecord',\n",
    "    output_path = s3_inference,\n",
    "    assemble_with= 'Line',\n",
    "    accept=('text/'+data_format)\n",
    ")\n",
    "\n",
    "print('SageMaker batch transform initialized with the following parameters:')\n",
    "for key in transformer.__dict__:\n",
    "    print('{}:{}'.format(key, transformer.__dict__[key]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The transform() function initiates the SageMaker batch transform job. SageMaker will create an inference instance, run the specified test set through the model, save the results to S3, and take down the inference instance. Batch transform is a great option if you require inference for large datasets and don't need sub-second response time.\n",
    "\n",
    "This usually takes 3 minutes. \n",
    "\n",
    "(**Optional**) While you wait, you may check the batch transform progress through the SageMaker console by following these instructions:  \n",
    "a.\tOpen SageMaker console in AWS.  \n",
    "b.\tOn the left panel, scroll until you see ‘Batch transform jobs’ beneath the ‘Inference’ section.  \n",
    "c.\tClick into the job to examine further details; wait until you see the status change to ‘Completed’."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The test data set still contains the \"Rings\" column the model tries to predict. \n",
    "# We do not want to send this column to the model, though. We use the SageMaker\n",
    "# input_filter to filter out that column before sending to the model. We then\n",
    "# join the model output with the input so we can compare the actual Rings count\n",
    "# to the predicted count.\n",
    "transformer.transform(\n",
    "    data=s3_test,\n",
    "    content_type='text/csv',\n",
    "    split_type='Line',\n",
    "    input_filter='$[1:]',\n",
    "    join_source='Input',\n",
    "    wait=True\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "SageMaker batch transform completed and saved the model inference results to S3. Now let's pull the results in to Spark for analysis."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from copy import deepcopy\n",
    "from pyspark.sql.types import FloatType\n",
    "\n",
    "# Read the schema from the initial dataset so you can apply it to the inference data.\n",
    "schema = deepcopy(abalone_data.schema)\n",
    "schema.add(\"Estimated_rings\", FloatType())\n",
    "\n",
    "# Pull down the inference data from S3\n",
    "inference_data = spark.read.load(s3_inference, format=data_format, schema=schema).repartition(num_workers)\n",
    "inference_data.show(n=5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we have our results, we need to quantify our model's performance. We will use root mean square error (RMSE) to measure how close Estimated_rings is to the actual Rings value.\n",
    "\n",
    "RMSE is a popular way to measure how closely a regression model predicts a response. A lower RMSE indicates a closer prediction.\n",
    "\n",
    "Here is the equation for RMSE:\n",
    "\n",
    "\\begin{equation*}\n",
    "RMSE = \\sqrt{\\frac{\\sum_{i=1}^n (\\hat{y_i}-y_i)^2}{N}}\n",
    "\\end{equation*}\n",
    "\n",
    "where $\\hat{y_i}$ is the number of predicted rings, $y_i$ is the observed number of rings, and N is the number of rows in the test data set.\n",
    "\n",
    "We'll use Spark SQL to run a SQL query on our data to calculate the RMSE."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rings = inference_data.schema.names[0]\n",
    "predicted_rings = inference_data.schema.names[-1]\n",
    "table_name = 'inference'\n",
    "\n",
    "inference_data.registerTempTable(table_name)\n",
    "sql_rmse = 'SELECT SQRT(AVG(POWER({}-{}, 2))) AS RMSE FROM {}'.format(rings, predicted_rings, table_name)\n",
    "\n",
    "rmse_results = spark.sql(sql_rmse)\n",
    "rmse_results.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Wrap-Up\n",
    "Congratulations! You processed data in Apache Spark on EMR and trained and deployed a machine learning model in Amazon SageMaker! Feel free to try different combinations of models and hyperparameters to see if you can reduce your model's RMSE."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 3
   },
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python3"
  },
  "notice": "Copyright 2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.  Licensed under the Apache License, Version 2.0 (the \"License\"). You may not use this file except in compliance with the License. A copy of the License is located at http://aws.amazon.com/apache2.0/ or in the \"license\" file accompanying this file. This file is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License."
 },
 "nbformat": 4,
 "nbformat_minor": 4
}