07B - Creating a Batch Inferencing Service.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Creating a Batch Inferencing Service\n",
    "\n",
    "In previous labs, you used an Azure ML *pipeline* to automate the training and registration of a model, and you published a model as a web service for real-time *inferencing* (getting predictions from a model). Now you'll combine these two concepts to create a pipeline for *batch inferencing*. What does that mean? Well, imagine a health clinic takes patient measurements all day, saving the details for each patient in a separate file. Then overnight, the diabetes prediction model can be used to process all of the day's patient data as a batch, generating predictions that will be waiting the following morning so that the clinic can follow up with patients who are predicted to be at risk of diabetes. That's what you'll implement in this exercise.\n",
    "\n",
    "Let's start by ensuring that you have the latest version of the Azure ML SDK installed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install --upgrade azureml-sdk[notebooks,automl,explain]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connect to Your Workspace\n",
    "\n",
    "Now connect to your workspace using the Azure ML SDK.\n",
    "\n",
    "> **Note**: If the authenticated session with your Azure subscription has expired since you completed the previous exercise, you'll be prompted to reauthenticate."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import azureml.core\n",
    "from azureml.core import Workspace\n",
    "\n",
    "# Load the workspace from the saved config file\n",
    "ws = Workspace.from_config()\n",
    "print('Ready to use Azure ML {} to work with {}'.format(azureml.core.VERSION, ws.name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train and Register a Model\n",
    "\n",
    "If you have completed the previous lab in this course, you will have registered many versions of the **diabetes_model** model, and you're ready to go.\n",
    "\n",
    "If you haven't, you can run the cell below to train and register the model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Experiment\n",
    "from azureml.core import Model\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import joblib\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.tree import DecisionTreeClassifier\n",
    "from sklearn.metrics import roc_auc_score\n",
    "from sklearn.metrics import roc_curve\n",
    "\n",
    "# Create an Azure ML experiment in your workspace\n",
    "experiment = Experiment(workspace = ws, name = \"diabetes-training\")\n",
    "run = experiment.start_logging()\n",
    "print(\"Starting experiment:\", experiment.name)\n",
    "\n",
    "# load the diabetes dataset\n",
    "print(\"Loading Data...\")\n",
    "diabetes = pd.read_csv('data/diabetes.csv')\n",
    "\n",
    "# Separate features and labels\n",
    "X, y = diabetes[['Pregnancies','PlasmaGlucose','DiastolicBloodPressure','TricepsThickness','SerumInsulin','BMI','DiabetesPedigree','Age']].values, diabetes['Diabetic'].values\n",
    "\n",
    "# Split data into training set and test set\n",
    "X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.30, random_state=0)\n",
    "\n",
    "# Train a decision tree model\n",
    "print('Training a decision tree model')\n",
    "model = DecisionTreeClassifier().fit(X_train, y_train)\n",
    "\n",
    "# calculate accuracy\n",
    "y_hat = model.predict(X_test)\n",
    "acc = np.average(y_hat == y_test)\n",
    "print('Accuracy:', acc)\n",
    "run.log('Accuracy', np.float(acc))\n",
    "\n",
    "# calculate AUC\n",
    "y_scores = model.predict_proba(X_test)\n",
    "auc = roc_auc_score(y_test,y_scores[:,1])\n",
    "print('AUC: ' + str(auc))\n",
    "run.log('AUC', np.float(auc))\n",
    "\n",
    "# Save the trained model\n",
    "model_file = 'diabetes_model.pkl'\n",
    "joblib.dump(value=model, filename=model_file)\n",
    "run.upload_file(name = 'outputs/' + model_file, path_or_stream = './' + model_file)\n",
    "\n",
    "# Complete the run\n",
    "run.complete()\n",
    "\n",
    "# Register the model\n",
    "run.register_model(model_path='outputs/diabetes_model.pkl', model_name='diabetes_model',\n",
    "                   tags={'Training context':'Inline Training'},\n",
    "                   properties={'AUC': run.get_metrics()['AUC'], 'Accuracy': run.get_metrics()['Accuracy']})\n",
    "\n",
    "print('Model trained and registered.')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Generate and Upload Batch Data\n",
    "\n",
    "Since we don't actually have a fully staffed clinic with patients from whom to get new data for this course, you'll generate a random sample from our diabetes CSV file and use those to test the pipeline. Then you'll upload that data to a datastore in the Azure Machine Learning workspace and register a dataset for it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Datastore, Dataset\n",
    "import pandas as pd\n",
    "import os\n",
    "\n",
    "# Load the diabetes data\n",
    "diabetes = pd.read_csv('data/diabetes2.csv')\n",
    "# Get a 100-item sample of the feature columns (not the diabetic label)\n",
    "sample = diabetes[['Pregnancies','PlasmaGlucose','DiastolicBloodPressure','TricepsThickness','SerumInsulin','BMI','DiabetesPedigree','Age']].sample(n=100).values\n",
    "\n",
    "# Create a folder\n",
    "batch_folder = './batch-data'\n",
    "os.makedirs(batch_folder, exist_ok=True)\n",
    "print(\"Folder created!\")\n",
    "\n",
    "# Save each sample as a separate file\n",
    "print(\"Saving files...\")\n",
    "for i in range(100):\n",
    "    fname = str(i+1) + '.csv'\n",
    "    sample[i].tofile(os.path.join(batch_folder, fname), sep=\",\")\n",
    "print(\"files saved!\")\n",
    "\n",
    "# Upload the files to the default datastore\n",
    "print(\"Uploading files to datastore...\")\n",
    "default_ds = ws.get_default_datastore()\n",
    "default_ds.upload(src_dir=\"batch-data\", target_path=\"batch-data\", overwrite=True, show_progress=True)\n",
    "\n",
    "# Register a dataset for the input data\n",
    "batch_data_set = Dataset.File.from_files(path=(default_ds, 'batch-data/'), validate=False)\n",
    "try:\n",
    "    batch_data_set = batch_data_set.register(workspace=ws, \n",
    "                                             name='batch-data',\n",
    "                                             description='batch data',\n",
    "                                             create_new_version=True)\n",
    "except Exception as ex:\n",
    "    print(ex)\n",
    "\n",
    "print(\"Done!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Compute\n",
    "\n",
    "We'll need a compute context for the pipeline, so we'll use the Azure ML compute cluster you used in the previous exercises (it will be created if it doesn't already exist)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core.compute import ComputeTarget, AmlCompute\n",
    "from azureml.core.compute_target import ComputeTargetException\n",
    "\n",
    "cluster_name = \"aml-cluster\"\n",
    "\n",
    "# Check for existing cluster\n",
    "try:\n",
    "    inference_cluster = ComputeTarget(workspace=ws, name=cluster_name)\n",
    "    print('Found existing cluster, use it.')\n",
    "except ComputeTargetException:\n",
    "    # Create an AzureMl Compute resource\n",
    "    compute_config = AmlCompute.provisioning_configuration(vm_size='STANDARD_DS2_V2', max_nodes=2)\n",
    "    inference_cluster = ComputeTarget.create(ws, cluster_name, compute_config)\n",
    "\n",
    "inference_cluster.wait_for_completion(show_output=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a Pipeline for Batch Inferencing\n",
    "\n",
    "Now we're ready to define the pipeline we'll use for batch inferencing. Our pipeline will need Python code to perform the batch inferencing, so let's create a folder where we can keep all the files used by the pipeline:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "# Create a folder for the experiment files\n",
    "experiment_folder = 'batch_pipeline'\n",
    "os.makedirs(experiment_folder, exist_ok=True)\n",
    "\n",
    "print(experiment_folder)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we'll create a Python script to do the actual work, and save it in the pipeline folder:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile $experiment_folder/batch_diabetes.py\n",
    "import os\n",
    "import numpy as np\n",
    "from azureml.core import Model\n",
    "import joblib\n",
    "\n",
    "\n",
    "def init():\n",
    "    # Runs when the pipeline step is initialized\n",
    "    global model\n",
    "\n",
    "    # load the model\n",
    "    model_path = Model.get_model_path('diabetes_model')\n",
    "    model = joblib.load(model_path)\n",
    "\n",
    "\n",
    "def run(mini_batch):\n",
    "    # This runs for each batch\n",
    "    resultList = []\n",
    "\n",
    "    # process each file in the batch\n",
    "    for f in mini_batch:\n",
    "        # Read the comma-delimited data into an array\n",
    "        data = np.genfromtxt(f, delimiter=',')\n",
    "        # Reshape into a 2-dimensional array for prediction (model expects multiple items)\n",
    "        prediction = model.predict(data.reshape(1, -1))\n",
    "        # Append prediction to results\n",
    "        resultList.append(\"{}: {}\".format(os.path.basename(f), prediction[0]))\n",
    "    return resultList"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we'll define a run context that includes the dependencies required by the script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Environment\n",
    "from azureml.core.runconfig import DEFAULT_CPU_IMAGE\n",
    "from azureml.core.runconfig import CondaDependencies\n",
    "\n",
    "# Add dependencies required by the model\n",
    "# For scikit-learn models, you need scikit-learn\n",
    "cd = CondaDependencies.create(pip_packages=['scikit-learn'])\n",
    "\n",
    "batch_env = Environment(name='batch_environment')\n",
    "batch_env.python.conda_dependencies = cd\n",
    "batch_env.docker.enabled = True\n",
    "batch_env.docker.base_image = DEFAULT_CPU_IMAGE\n",
    "print('Configuration ready.')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You're going to use a pipeline to run the batch prediction script, generate predictions from the input data, and save the results as a text file in the output folder. To do this, you can use a **ParallelRunStep**, which enables the batch data to be processed in parallel and the results collated in a single output file named *parallel_run_step.txt*.\n",
    "\n",
    "So you'll need to install the library containing the **ParallelRunStep** class."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install --upgrade azureml-contrib-pipeline-steps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "OK, now you're ready to define a **ParallelRunStep** pipeline step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.contrib.pipeline.steps import ParallelRunConfig, ParallelRunStep\n",
    "from azureml.pipeline.core import PipelineData\n",
    "\n",
    "default_ds = ws.get_default_datastore()\n",
    "model = ws.models['diabetes_model']\n",
    "\n",
    "output_dir = PipelineData(name='inferences', \n",
    "                          datastore=default_ds, \n",
    "                          output_path_on_compute='diabetes/results')\n",
    "\n",
    "parallel_run_config = ParallelRunConfig(\n",
    "    source_directory=experiment_folder,\n",
    "    entry_script=\"batch_diabetes.py\",\n",
    "    mini_batch_size=\"5\",\n",
    "    error_threshold=10,\n",
    "    output_action=\"append_row\",\n",
    "    environment=batch_env,\n",
    "    compute_target=inference_cluster,\n",
    "    node_count=2)\n",
    "\n",
    "parallelrun_step = ParallelRunStep(\n",
    "    name='batch-score-diabetes',\n",
    "    models=[model],\n",
    "    parallel_run_config=parallel_run_config,\n",
    "    inputs=[batch_data_set.as_named_input('diabetes_batch')],\n",
    "    output=output_dir,\n",
    "    arguments=[],\n",
    "    allow_reuse=True\n",
    ")\n",
    "\n",
    "print('Steps defined')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now it's time to put the step into a pipeline, and run it.\n",
    "\n",
    "> **Note**: This may take some time!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "from azureml.core import Experiment\n",
    "from azureml.pipeline.core import Pipeline\n",
    "from azureml.widgets import RunDetails\n",
    "\n",
    "pipeline = Pipeline(workspace=ws, steps=[parallelrun_step])\n",
    "pipeline_run = Experiment(ws, 'batch_prediction_pipeline').submit(pipeline)\n",
    "RunDetails(pipeline_run).show()\n",
    "pipeline_run.wait_for_completion(show_output=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When the pipeline has finished running, the resulting predictions will have been saved in the outputs of the experiment associated with the first (and only) step in the pipeline. You can retrieve it as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import shutil\n",
    "\n",
    "shutil.rmtree('diabetes-results', ignore_errors=True)\n",
    "\n",
    "prediction_run = next(pipeline_run.get_children())\n",
    "prediction_output = prediction_run.get_output_data('inferences')\n",
    "prediction_output.download(local_path='diabetes-results')\n",
    "\n",
    "\n",
    "for root, dirs, files in os.walk('diabetes-results'):\n",
    "    for file in files:\n",
    "        if file.endswith('parallel_run_step.txt'):\n",
    "            result_file = os.path.join(root,file)\n",
    "\n",
    "# cleanup output format\n",
    "df = pd.read_csv(result_file, delimiter=\":\", header=None)\n",
    "df.columns = [\"File\", \"Prediction\"]\n",
    "\n",
    "# Display the first 20 results\n",
    "df.head(20)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Publish the Pipeline and use its REST Interface\n",
    "\n",
    "Now that you have a working pipeline for batch inferencing, you can publish it and use a REST endpoint to run it from an application."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "published_pipeline = pipeline_run.publish_pipeline(\n",
    "    name='Diabetes_Parallel_Batch_Pipeline', description='Batch scoring of diabetes data', version='1.0')\n",
    "\n",
    "published_pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that the published pipeline has an endpoint, which you can see in the Azure portal. You can also find it as a property of the published pipeline object:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rest_endpoint = published_pipeline.endpoint\n",
    "print(rest_endpoint)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To use the endpoint, client applications need to make a REST call over HTTP. This request must be authenticated, so an authorization header is required. To test this out, we'll use the authorization header from your current connection to your Azure workspace, which you can get using the following code:\n",
    "\n",
    "> **Note**: A real application would require a service principal with which to be authenticated."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core.authentication import InteractiveLoginAuthentication\n",
    "\n",
    "interactive_auth = InteractiveLoginAuthentication()\n",
    "auth_header = interactive_auth.get_authentication_header()\n",
    "print('Authentication header ready.')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we're ready to call the REST interface. The pipeline runs asynchronously, so we'll get an identifier back, which we can use to track the pipeline experiment as it runs:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "\n",
    "rest_endpoint = published_pipeline.endpoint\n",
    "response = requests.post(rest_endpoint, \n",
    "                         headers=auth_header, \n",
    "                         json={\"ExperimentName\": \"Batch_Pipeline_via_REST\"})\n",
    "run_id = response.json()[\"Id\"]\n",
    "run_id"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since we have the run ID, we can use the **RunDetails** widget to view the experiment as it runs:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.core.run import PipelineRun\n",
    "from azureml.widgets import RunDetails\n",
    "\n",
    "published_pipeline_run = PipelineRun(ws.experiments[\"Batch_Pipeline_via_REST\"], run_id)\n",
    "RunDetails(published_pipeline_run).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As before, the results are in the output of the first pipeline step:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import shutil\n",
    "\n",
    "shutil.rmtree(\"diabetes-results\", ignore_errors=True)\n",
    "\n",
    "prediction_run = next(published_pipeline_run.get_children())\n",
    "prediction_output = prediction_run.get_output_data(\"inferences\")\n",
    "prediction_output.download(local_path=\"diabetes-results\")\n",
    "\n",
    "\n",
    "for root, dirs, files in os.walk(\"diabetes-results\"):\n",
    "    for file in files:\n",
    "        if file.endswith('parallel_run_step.txt'):\n",
    "            result_file = os.path.join(root,file)\n",
    "\n",
    "# cleanup output format\n",
    "df = pd.read_csv(result_file, delimiter=\":\", header=None)\n",
    "df.columns = [\"File\", \"Prediction\"]\n",
    "\n",
    "# Display the first 20 results\n",
    "df.head(20)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now you have a pipeline that can be used to batch process daily patient data.\n",
    "\n",
    "**More Information**: For more details about using pipelines for batch inferencing, see the [How to Run Batch Predictions](https://docs.microsoft.com/azure/machine-learning/how-to-run-batch-predictions) in the Azure Machine Learning documentation."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.6 - AzureML",
   "language": "python",
   "name": "python3-azureml"
  },
  "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.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}