06A - Creating a Pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating a Pipeline\n",
    "\n",
    "In the previous labs, you used the Azure Machine Learning SDK to explore the entire model training process from accessing data through to running training experiments and registering machine learning models. Up until now, you have performed the various steps required to create a machine learning solution interactively. In this lab, you'll explore automation of these steps using *pipelines*.\n",
    "\n",
    "## Connect to Your Workspace\n",
    "\n",
    "The first thing you need to do is to 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": [
    "## Prepare Data for an Experiment\n",
    "\n",
    "In this lab, you'll use a dataset containing details of diabetes patients. Run the cell below to create this dataset (if you created it in the previous lab, the code will create a new version)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Dataset\n",
    "\n",
    "default_ds = ws.get_default_datastore()\n",
    "default_ds.upload_files(files=['./data/diabetes.csv', './data/diabetes2.csv'], # Upload the diabetes csv files in /data\n",
    "                       target_path='diabetes-data/', # Put it in a folder path in the datastore\n",
    "                       overwrite=True, # Replace existing files of the same name\n",
    "                       show_progress=True)\n",
    "\n",
    "#Create a tabular dataset from the path on the datastore (this may take a short while)\n",
    "tab_data_set = Dataset.Tabular.from_delimited_files(path=(default_ds, 'diabetes-data/*.csv'))\n",
    "\n",
    "# Register the tabular dataset\n",
    "tab_data_set = tab_data_set.register(workspace=ws, \n",
    "                           name='diabetes dataset',\n",
    "                           description='diabetes data',\n",
    "                           tags = {'format':'CSV'},\n",
    "                           create_new_version=True)\n",
    "\n",
    "print('Dataset ready.')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Scripts for Pipeline Steps\n",
    "\n",
    "Pipelines consist of one or more *steps*, which can be Python scripts, or specialized steps like an Auto ML training estimator or a data transfer step that copies data from one location to another. Each step can run in its own compute context.\n",
    "\n",
    "In this exercise, you'll build a simple pipeline that contains an estimator step (to train a model) and a Python script step (to register the trained model)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "# Create a folder for the pipeline step files\n",
    "experiment_folder = 'diabetes_pipeline'\n",
    "os.makedirs(experiment_folder, exist_ok=True)\n",
    "\n",
    "print(experiment_folder)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now you can create the script for the first step, which will train a model. The script includes a parameter named **output_folder**, which references the folder where the trained model should be saved."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile $experiment_folder/train_diabetes.py\n",
    "# Import libraries\n",
    "from azureml.core import Run\n",
    "import argparse\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",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# Get parameters\n",
    "parser = argparse.ArgumentParser()\n",
    "parser.add_argument('--output_folder', type=str, dest='output_folder', default=\"diabetes_model\", help='output folder')\n",
    "args = parser.parse_args()\n",
    "output_folder = args.output_folder\n",
    "\n",
    "# Get the experiment run context\n",
    "run = Run.get_context()\n",
    "\n",
    "# load the diabetes data (passed as an input dataset)\n",
    "print(\"Loading Data...\")\n",
    "diabetes = run.input_datasets['diabetes_train'].to_pandas_dataframe()\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 adecision 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",
    "# plot ROC curve\n",
    "fpr, tpr, thresholds = roc_curve(y_test, y_scores[:,1])\n",
    "fig = plt.figure(figsize=(6, 4))\n",
    "# Plot the diagonal 50% line\n",
    "plt.plot([0, 1], [0, 1], 'k--')\n",
    "# Plot the FPR and TPR achieved by our model\n",
    "plt.plot(fpr, tpr)\n",
    "plt.xlabel('False Positive Rate')\n",
    "plt.ylabel('True Positive Rate')\n",
    "plt.title('ROC Curve')\n",
    "run.log_image(name = \"ROC\", plot = fig)\n",
    "plt.show()\n",
    "\n",
    "# Save the trained model\n",
    "os.makedirs(output_folder, exist_ok=True)\n",
    "output_path = output_folder + \"/model.pkl\"\n",
    "joblib.dump(value=model, filename=output_path)\n",
    "\n",
    "run.complete()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The script for the second step of the pipeline will load the model from where it was saved, and then register it in the workspace. It includes a single **model_folder** parameter that contains the path where the model was saved."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile $experiment_folder/register_diabetes.py\n",
    "# Import libraries\n",
    "import argparse\n",
    "import joblib\n",
    "from azureml.core import Workspace, Model, Run\n",
    "\n",
    "# Get parameters\n",
    "parser = argparse.ArgumentParser()\n",
    "parser.add_argument('--model_folder', type=str, dest='model_folder', default=\"diabetes_model\", help='model location')\n",
    "args = parser.parse_args()\n",
    "model_folder = args.model_folder\n",
    "\n",
    "# Get the experiment run context\n",
    "run = Run.get_context()\n",
    "\n",
    "# load the model\n",
    "print(\"Loading model from \" + model_folder)\n",
    "model_file = model_folder + \"/model.pkl\"\n",
    "model = joblib.load(model_file)\n",
    "\n",
    "Model.register(workspace=run.experiment.workspace,\n",
    "               model_path = model_file,\n",
    "               model_name = 'diabetes_model',\n",
    "               tags={'Training context':'Pipeline'})\n",
    "\n",
    "run.complete()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prepare a Compute Environment for the Pipeline Steps\n",
    "\n",
    "In this exercise, you'll use the same compute for both steps, but it's important to realize that each step is run independently; so you could specify different compute contexts for each step if appropriate.\n",
    "\n",
    "First, get the compute target."
   ]
  },
  {
   "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",
    "# Verify that cluster exists\n",
    "try:\n",
    "    pipeline_cluster = ComputeTarget(workspace=ws, name=cluster_name)\n",
    "    print('Found existing cluster, use it.')\n",
    "except ComputeTargetException:\n",
    "    # If not, create it\n",
    "    compute_config = AmlCompute.provisioning_configuration(vm_size='STANDARD_D2_V2', \n",
    "                                                           vm_priority='lowpriority', \n",
    "                                                           max_nodes=4)\n",
    "    pipeline_cluster = ComputeTarget.create(ws, cluster_name, compute_config)\n",
    "\n",
    "pipeline_cluster.wait_for_completion(show_output=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The compute will require a Python environment with the necessary package dependencies installed, so we'll create a run configuration."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Environment\n",
    "from azureml.core.conda_dependencies import CondaDependencies\n",
    "from azureml.core.runconfig import RunConfiguration\n",
    "\n",
    "# Create a Python environment for the experiment\n",
    "diabetes_env = Environment(\"diabetes-experiment-env\")\n",
    "diabetes_env.python.user_managed_dependencies = False # Let Azure ML manage dependencies\n",
    "diabetes_env.docker.enabled = True # Use a docker container\n",
    "\n",
    "# Create a set of package dependencies\n",
    "diabetes_packages = CondaDependencies.create(conda_packages=['scikit-learn','ipykernel','matplotlib', 'pandas'],\n",
    "                                             pip_packages=['azureml-sdk','pyarrow'])\n",
    "\n",
    "# Add the dependencies to the environment\n",
    "diabetes_env.python.conda_dependencies = diabetes_packages\n",
    "\n",
    "# Register the environment (just in case previous lab wasn't completed)\n",
    "diabetes_env.register(workspace=ws)\n",
    "registered_env = Environment.get(ws, 'diabetes-experiment-env')\n",
    "\n",
    "# Create a new runconfig object for the pipeline\n",
    "pipeline_run_config = RunConfiguration()\n",
    "\n",
    "# Use the compute you created above. \n",
    "pipeline_run_config.target = pipeline_cluster\n",
    "\n",
    "# Assign the environment to the run configuration\n",
    "pipeline_run_config.environment = registered_env\n",
    "\n",
    "print (\"Run configuration created.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create and Run a Pipeline\n",
    "\n",
    "Now we're ready to create and run a pipeline.\n",
    "\n",
    "First we need to define the steps for the pipeline, and any data references that need to passed between them. In this case, the first step must write the model to a folder that can be read from by the second step. Since the steps will be run on remote compute (and in fact, could each be run on different compute), the folder path must be passed as a data reference to a location in a datastore within the workspace. The **PipelineData** object is a special kind of data reference that is used for interim storage locations that can be passed between pipeline steps, so we'll create one and use at as the output for the first step and the input for the second step. Note that we also need to pass it as a script argument so our code can access the datastore location referenced by the data reference."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.core import PipelineData\n",
    "from azureml.pipeline.steps import PythonScriptStep, EstimatorStep\n",
    "from azureml.train.estimator import Estimator\n",
    "\n",
    "# Get the training dataset\n",
    "diabetes_ds = ws.datasets.get(\"diabetes dataset\")\n",
    "\n",
    "# Create a PipelineData (temporary Data Reference) for the model folder\n",
    "model_folder = PipelineData(\"model_folder\", datastore=ws.get_default_datastore())\n",
    "\n",
    "estimator = Estimator(source_directory=experiment_folder,\n",
    "                        compute_target = pipeline_cluster,\n",
    "                        environment_definition=pipeline_run_config.environment,\n",
    "                        entry_script='train_diabetes.py')\n",
    "\n",
    "train_step = EstimatorStep(name = \"Train Model\",\n",
    "                           estimator=estimator, \n",
    "                           estimator_entry_script_arguments=['--output_folder', model_folder],\n",
    "                           inputs=[diabetes_ds.as_named_input('diabetes_train')],\n",
    "                           outputs=[model_folder],\n",
    "                           compute_target = pipeline_cluster,\n",
    "                           allow_reuse = True)\n",
    "\n",
    "# Step 2, run the model registration script\n",
    "register_step = PythonScriptStep(name = \"Register Model\",\n",
    "                                source_directory = experiment_folder,\n",
    "                                script_name = \"register_diabetes.py\",\n",
    "                                arguments = ['--model_folder', model_folder],\n",
    "                                inputs=[model_folder],\n",
    "                                compute_target = pipeline_cluster,\n",
    "                                runconfig = pipeline_run_config,\n",
    "                                allow_reuse = True)\n",
    "\n",
    "print(\"Pipeline steps defined\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "OK, we're ready to go. let's build the pipeline from the steps we've defined and run it as an experiment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Experiment\n",
    "from azureml.pipeline.core import Pipeline\n",
    "from azureml.widgets import RunDetails\n",
    "\n",
    "# Construct the pipeline\n",
    "pipeline_steps = [train_step, register_step]\n",
    "pipeline = Pipeline(workspace = ws, steps=pipeline_steps)\n",
    "print(\"Pipeline is built.\")\n",
    "\n",
    "# Create an experiment and run the pipeline\n",
    "experiment = Experiment(workspace = ws, name = 'diabetes-training-pipeline')\n",
    "pipeline_run = experiment.submit(pipeline, regenerate_outputs=True)\n",
    "print(\"Pipeline submitted for execution.\")\n",
    "\n",
    "RunDetails(pipeline_run).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The widget above shows details of the pipeline as it runs. You can also monitor pipeline runs in the **Experiments** page in [Azure Machine Learning studio](https://ml.azure.com).\n",
    "\n",
    "When the pipeline has finished, a new model should be registered with a *Training context* tag indicating it was trained in a pipeline. Run the following code to verify this."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Model\n",
    "\n",
    "for model in Model.list(ws):\n",
    "    print(model.name, 'version:', model.version)\n",
    "    for tag_name in model.tags:\n",
    "        tag = model.tags[tag_name]\n",
    "        print ('\\t',tag_name, ':', tag)\n",
    "    for prop_name in model.properties:\n",
    "        prop = model.properties[prop_name]\n",
    "        print ('\\t',prop_name, ':', prop)\n",
    "    print('\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is a simple example, designed to demonstrate the principle. In reality, you could build more sophisticated logic into the pipeline steps - for example, evaluating the model against some test data to calculate a performance metric like AUC or accuracy, comparing the metric to that of any previously registered versions of the model, and only registering the new model if it performs better.\n",
    "\n",
    "You can use the [Azure Machine Learning extension for Azure DevOps](https://marketplace.visualstudio.com/items?itemName=ms-air-aiagility.vss-services-azureml) to combine Azure ML pipelines with Azure DevOps pipelines (yes, it *is* confusing that they have the same name!) and integrate model retraining into a *continuous integration/continuous deployment (CI/CD)* process. For example you could use an Azure DevOps *build* pipeline to trigger an Azure ML pipeline that trains and registers a model, and when the model is registered it could trigger an Azure Devops *release* pipeline that deploys the model as a web service, along with the application or service that consumes the model."
   ]
  }
 ],
 "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
}