10 - Create a Batch Inferencing Service.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Create a Batch Inferencing Service\n",
    "\n",
    "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. With Azure Machine Learning, you can accomplish this by creating a *batch inferencing pipeline*; and that's what you'll implement in this exercise.\n",
    "\n",
    "## Install the Azure Machine Learning SDK\n",
    "\n",
    "The Azure Machine Learning SDK is updated frequently. Run the following cell to upgrade to the latest release, including the *notebooks* extra package."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install --upgrade azureml-sdk[notebooks]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connect to your workspace\n",
    "\n",
    "With the latest version of the SDK installed, now you're ready to connect to your workspace.\n",
    "\n",
    "> **Note**: If you haven't already established an authenticated session with your Azure subscription, you'll be prompted to authenticate by clicking a link, entering an authentication code, and signing into Azure."
   ]
  },
  {
   "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",
    "Now let's train and register a model to deploy in a batch inferencing pipeline."
   ]
  },
  {
   "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 exercise, you'll generate a random sample from our diabetes CSV file, 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",
    "# Set default data store\n",
    "ws.set_default_datastore('workspaceblobstore')\n",
    "default_ds = ws.get_default_datastore()\n",
    "\n",
    "# Enumerate all datastores, indicating which is the default\n",
    "for ds_name in ws.datastores:\n",
    "    print(ds_name, \"- Default =\", ds_name == default_ds.name)\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 following code to specify an Azure Machine Learning compute cluster (it will be created if it doesn't already exist).\n",
    "\n",
    "> **Important**: Change *your-compute-cluster* to the name of your compute cluster in the code below before running it! Cluster names must be globally unique names between 2 to 16 characters in length. Valid characters are letters, digits, and the - character."
   ]
  },
  {
   "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 = \"your-compute-cluster\"\n",
    "\n",
    "try:\n",
    "    # Check for existing compute target\n",
    "    inference_cluster = ComputeTarget(workspace=ws, name=cluster_name)\n",
    "    print('Found existing cluster, use it.')\n",
    "except ComputeTargetException:\n",
    "    # If it doesn't already exist, create it\n",
    "    try:\n",
    "        compute_config = AmlCompute.provisioning_configuration(vm_size='STANDARD_DS11_V2', max_nodes=2)\n",
    "        inference_cluster = ComputeTarget.create(ws, cluster_name, compute_config)\n",
    "        inference_cluster.wait_for_completion(show_output=True)\n",
    "    except Exception as ex:\n",
    "        print(ex)\n",
    "    "
   ]
  },
  {
   "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",
    "# For parallel pipeline steps, you need azureml-core and azureml-dataprep[fuse]\n",
    "cd = CondaDependencies.create(conda_packages=['scikit-learn','pip'],\n",
    "                              pip_packages=['azureml-defaults','azureml-core','azureml-dataprep[fuse]'])\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*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.steps import ParallelRunConfig, ParallelRunStep\n",
    "from azureml.pipeline.core import PipelineData\n",
    "\n",
    "default_ds = ws.get_default_datastore()\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",
    "    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": true
   },
   "outputs": [],
   "source": [
    "from azureml.core import Experiment\n",
    "from azureml.pipeline.core import Pipeline\n",
    "\n",
    "pipeline = Pipeline(workspace=ws, steps=[parallelrun_step])\n",
    "pipeline_run = Experiment(ws, 'batch_prediction_pipeline').submit(pipeline)\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",
    "# Remove the local results folder if left over from a previous run\n",
    "shutil.rmtree('diabetes-results', ignore_errors=True)\n",
    "\n",
    "# Get the run for the first step and download its output\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",
    "# Traverse the folder hierarchy and find the results file\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",
    "\n",
    "# Block until the run completes\n",
    "published_pipeline_run.wait_for_completion(show_output=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "published_pipeline_run.wait_for_completion(show_output=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Wait for the pipeline run to complete, and then run the following cell to see the results.\n",
    "\n",
    "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",
    "# Remove the local results folder if left over from a previous run\n",
    "shutil.rmtree('diabetes-results', ignore_errors=True)\n",
    "\n",
    "# Get the run for the first step and download its output\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",
    "# Traverse the folder hierarchy and find the results file\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
}