05-Creating_a_Pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating an Azure Machine Learning Pipeline\n",
    "\n",
    "You can perform the various steps required to ingest data, train a model, and register the model individually by using the Azure ML SDK to run script-based experiments. However, in an enterprise environment it is common to encapsulate the sequence of discrete steps required to build a machine learning solution into a *pipeline* that can be run on one or more compute targets, either on-demand by a user, from an automated build process, or on a schedule.\n",
    "\n",
    "In this lab, you'll bring together all of these elements to create a simple pipeline that trains and registers a model.\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 the Training Data\n",
    "\n",
    "You can use local data files to train a model, but when running training workloads automatically on cloud-based compute, it makes more sense to store the data centrally in the cloud and ingest it into the training script wherever it happens to be running.\n",
    "\n",
    "In this lab, you'll upload the training data to a *datastore* and define a *dataset* that can be used to access the data from a training script. For simplicity, you'll upload the data to the default datastore for your Azure Machine Learning workspace - this is an Azure Storage blob container that was created when you provisioned the workspace. In a real solution, you'd likely register a datastore that references the cloud location where you typically store your data. You'll then create a *tabular* dataset that references the CSV files you uploaded."
   ]
  },
  {
   "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",
    "Now you're ready to start work on your pipeline. 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). Start by creating a folder to contain the scripts for each step."
   ]
  },
  {
   "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": [
    "The first step will use an estimator to run a training script. The code in the following cell creates this script for you. Note that 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",
    "\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",
    "# 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 to the folder where the model was saved by the previous step."
   ]
  },
  {
   "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",
    "The pipeline will eventually be published and run on-demand, so it needs a compute environment in which to run. 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, you need a compute target. In this case, you create an Azure Machine Learning compute cluster in your workspace (or use an existing one if you have created it previously).\n",
    "\n",
    "> **Important**: Change *your-compute-cluster* to a unique name for 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 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-pipeline-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','pandas'],\n",
    "                                             pip_packages=['azureml-defaults','azureml-dataprep[pandas]'])\n",
    "\n",
    "# Add the dependencies to the environment\n",
    "diabetes_env.python.conda_dependencies = diabetes_packages\n",
    "\n",
    "# Register the environment (just in case you want to use it again)\n",
    "diabetes_env.register(workspace=ws)\n",
    "registered_env = Environment.get(ws, 'diabetes-pipeline-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 define and run the 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 to pass data from the output of one pipeline step to the input of another, creating a dependency between them. You'll create one and use it 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 your 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",
    "# Step 1, run the estimator to train the model\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, now you're ready to build the pipeline from the steps you've defined and run it as an experiment.\n",
    "\n",
    "> **Note**: This may take a while. The training cluster must be started and configured with the Python environment before the scripts can be run. Now might be a good time to take a coffee break!"
   ]
  },
  {
   "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": [
    "## Publish the Pipeline\n",
    "\n",
    "Now that you've created a pipeline and verified it works, you can publish it as a REST service."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "published_pipeline = pipeline.publish(name=\"Diabetes_Training_Pipeline\",\n",
    "                                      description=\"Trains diabetes model\",\n",
    "                                      version=\"1.0\")\n",
    "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. A real application would require a service principal with which to be authenticated, but 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:"
   ]
  },
  {
   "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()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now you're ready to call the REST interface. The pipeline runs asynchronously, so you'll get an identifier back, which you can use to track the pipeline experiment as it runs:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "experiment_name = 'Run-diabetes-pipeline'\n",
    "\n",
    "response = requests.post(rest_endpoint, \n",
    "                         headers=auth_header, \n",
    "                         json={\"ExperimentName\": experiment_name})\n",
    "run_id = response.json()[\"Id\"]\n",
    "run_id"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since you have the run ID, you can use it to wait for the run to complete.\n",
    "\n",
    "> **Note**: The pipeline should complete quickly, because each step was configured to allow output reuse. This was done primarily for convenience and to save time in this example. In reality, you'd likely want the first step to run every time in case the data has changed, and trigger the subsequent steps only if the output from step one changes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.core.run import PipelineRun\n",
    "\n",
    "published_pipeline_run = PipelineRun(ws.experiments[experiment_name], run_id)\n",
    "published_pipeline_run.wait_for_completion(show_output=True)"
   ]
  },
  {
   "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
}