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 find the existing version)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Dataset\n",
    "\n",
    "default_ds = ws.get_default_datastore()\n",
    "\n",
    "if 'diabetes dataset' not in ws.datasets:\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",
    "    try:\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",
    "        print('Dataset registered.')\n",
    "    except Exception as ex:\n",
    "        print(ex)\n",
    "else:\n",
    "    print('Dataset already registered.')"
   ]
  },
  {
   "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",
    "import os\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 you created in a previous lab (if it doesn't exist, it will be created).\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",
    "    pipeline_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",
    "        pipeline_cluster = ComputeTarget.create(ws, cluster_name, compute_config)\n",
    "        pipeline_cluster.wait_for_completion(show_output=True)\n",
    "    except Exception as ex:\n",
    "        print(ex)\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The compute will require a Python environment with the necessary package dependencies installed, so you'll need to 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-defaults','azureml-dataprep[pandas]','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 you're ready to create and run a pipeline.\n",
    "\n",
    "First you 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 you'll create one and use at as the output for the first step and the input for the second step. Note that you 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, you're ready build the pipeline from the steps you've defined and run it as an experiment."
   ]
  },
  {
   "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",
    "# 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",
    "pipeline_run.wait_for_completion(show_output=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The output from the pipeline experiment will be displayed as it runs. keep an eye on the kernel indicator at the top right of the page, when it turns from **&#9899;** to **&#9711;**, the code has finished running. 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
}