Custom_Script/02_CustomScript_Training_Pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Copyright (c) Microsoft Corporation. All rights reserved.\n",
    "\n",
    "Licensed under the MIT License."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Training Pipeline - Custom Script\n",
    "_**Training many models using a custom script**_\n",
    "\n",
    "----\n",
    "\n",
    "This notebook demonstrates how to create a pipeline that trains and registers many models using a custom script. We utilize the [ParallelRunStep](https://docs.microsoft.com/en-us/azure/machine-learning/how-to-use-parallel-run-step) to parallelize the process of training the models to make the process more efficient. For this solution accelerator we are using the [OJ Sales Dataset](https://azure.microsoft.com/en-us/services/open-datasets/catalog/sample-oj-sales-simulated/) to train individual models that predict sales for each store and brand of orange juice.\n",
    "\n",
    "The model we use here is a simple, regression-based forecaster built on scikit-learn and pandas utilities. See the [training script](scripts/train.py) to see how the forecaster is constructed. This forecaster is intended for demonstration purposes, so it does not handle the large variety of special cases that one encounters in time-series modeling. For instance, the model here assumes that all time-series are comprised of regularly sampled observations on a contiguous interval with no missing values. The model does not include any handling of categorical variables. For a more general-use forecaster that handles missing data, advanced featurization, and automatic model selection, see the [AutoML Forecasting task](https://docs.microsoft.com/en-us/azure/machine-learning/how-to-auto-train-forecast). Also, see the notebooks demonstrating [AutoML forecasting in a many models scenario](../Automated_ML).\n",
    "\n",
    "### Prerequisites\n",
    "At this point, you should have already:\n",
    "\n",
    "1. Created your AML Workspace using the [00_Setup_AML_Workspace notebook](../00_Setup_AML_Workspace.ipynb)\n",
    "2. Run [01_Data_Preparation.ipynb](../01_Data_Preparation.ipynb) to setup your compute and create the dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Please ensure you have the latest version of the Azure ML SDK and also install Pipeline Steps Package"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!pip install --upgrade azureml-sdk"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !pip install azureml-pipeline-steps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1.0 Connect to workspace and datastore"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from azureml.core import Workspace\n",
    "\n",
    "# set up workspace\n",
    "ws = Workspace.from_config()\n",
    "\n",
    "# set up datastores\n",
    "dstore = ws.get_default_datastore()\n",
    "\n",
    "print('Workspace Name: ' + ws.name, \n",
    "      'Azure Region: ' + ws.location, \n",
    "      'Subscription Id: ' + ws.subscription_id, \n",
    "      'Resource Group: ' + ws.resource_group, \n",
    "      sep = '\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.0 Create an experiment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from azureml.core import Experiment\n",
    "\n",
    "experiment = Experiment(ws, 'oj_training_pipeline')\n",
    "\n",
    "print('Experiment name: ' + experiment.name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3.0 Get the training Dataset\n",
    "\n",
    "Next, we get the training Dataset using the [Dataset.get_by_name()](https://docs.microsoft.com/python/api/azureml-core/azureml.core.dataset.dataset#get-by-name-workspace--name--version--latest--) method.\n",
    "\n",
    "This is the training dataset we created and registered in the [data preparation notebook](../01_Data_Preparation.ipynb). If you chose to use only a subset of the files, the training dataset name will be `oj_data_small_train`. Otherwise, the name you'll have to use is `oj_data_train`. \n",
    "\n",
    "We recommend to start with the small dataset and make sure everything runs successfully, then scale up to the full dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataset_name = 'oj_data_small_train'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from azureml.core.dataset import Dataset\n",
    "\n",
    "dataset = Dataset.get_by_name(ws, name=dataset_name)\n",
    "dataset_input = dataset.as_named_input(dataset_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4.0 Create the training pipeline\n",
    "Now that the workspace, experiment, and dataset are set up, we can put together a pipeline for training.\n",
    "\n",
    "### 4.1 Configure environment for ParallelRunStep\n",
    "An [environment](https://docs.microsoft.com/en-us/azure/machine-learning/concept-environments) defines a collection of resources that we will need to run our pipelines. We configure a reproducible Python environment for our training script including the [scikit-learn](https://scikit-learn.org/stable/index.html) python library."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from azureml.core import Environment\n",
    "from azureml.core.conda_dependencies import CondaDependencies\n",
    "\n",
    "train_env = Environment(name=\"many_models_environment\")\n",
    "train_conda_deps = CondaDependencies.create(pip_packages=['sklearn', 'pandas', 'joblib', 'azureml-defaults', 'azureml-core', 'azureml-dataprep[fuse]'])\n",
    "train_env.python.conda_dependencies = train_conda_deps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.2 Choose a compute target "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Currently ParallelRunConfig only supports AMLCompute. This is the compute cluster you created in the [setup notebook](../00_Setup_AML_Workspace.ipynb#3.0-Create-compute-cluster)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cpu_cluster_name = \"cpucluster\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from azureml.core.compute import AmlCompute\n",
    "\n",
    "compute = AmlCompute(ws, cpu_cluster_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.3 Set up ParallelRunConfig\n",
    "\n",
    "[ParallelRunConfig](https://docs.microsoft.com/en-us/python/api/azureml-pipeline-steps/azureml.pipeline.steps.parallel_run_config.parallelrunconfig?view=azure-ml-py) provides the configuration for the ParallelRunStep we'll be creating next. Here we specify the environment and compute target we created above along with the entry script that will be for each batch.\n",
    "\n",
    "There's a number of important parameters to configure including:\n",
    "- **mini_batch_size**: The number of files per batch. If you have 500 files and mini_batch_size is 10, 50 batches would be created containing 10 files each. Batches are split across the various nodes. \n",
    "\n",
    "- **node_count**: The number of compute nodes to be used for running the user script. For the small sample of OJ datasets, we only need a single node, but you will likely need to increase this number for larger datasets composed of more files. If you increase the node count beyond five here, you may need to increase the max_nodes for the compute cluster as well.\n",
    "\n",
    "- **process_count_per_node**: The number of processes per node. The compute cluster we are using has 8 cores so we set this parameter to 8.\n",
    "\n",
    "- **run_invocation_timeout**: The run() method invocation timeout in seconds. The timeout should be set to be higher than the maximum training time of one model (in seconds), by default it's 60. Since the batches that takes the longest to train are about 120 seconds, we set it to be 180 to ensure the method has adequate time to run.\n",
    "\n",
    "\n",
    "We also added tags to preserve the information about our training cluster's node count, process count per node, and dataset name. You can find the 'Tags' column in Azure Machine Learning Studio."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.steps import ParallelRunConfig\n",
    "\n",
    "processes_per_node = 8\n",
    "node_count = 1\n",
    "timeout = 180\n",
    "\n",
    "parallel_run_config = ParallelRunConfig(\n",
    "    source_directory='./scripts',\n",
    "    entry_script='train.py',\n",
    "    mini_batch_size=\"1\",\n",
    "    run_invocation_timeout=timeout,\n",
    "    error_threshold=-1,\n",
    "    output_action=\"append_row\",\n",
    "    environment=train_env,\n",
    "    process_count_per_node=processes_per_node,\n",
    "    compute_target=compute,\n",
    "    node_count=node_count)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.4 Set up ParallelRunStep\n",
    "\n",
    "This [ParallelRunStep](https://docs.microsoft.com/en-us/python/api/azureml-pipeline-steps/azureml.pipeline.steps.parallel_run_step.parallelrunstep?view=azure-ml-py) is the main step in our training pipeline. \n",
    "\n",
    "First, we set up the output directory and define the pipeline's output name. The datastore that stores the pipeline's output data is Workspace's default datastore."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from azureml.pipeline.core import PipelineData\n",
    "\n",
    "output_dir = PipelineData(name=\"training_output\", datastore=dstore)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We provide our ParallelRunStep with a name, the ParallelRunConfig created above and several other parameters:\n",
    "\n",
    "- **inputs**: A list of input datasets. Here we'll use the dataset created in the previous notebook. The number of files in that path determines the number of models will be trained in the ParallelRunStep.\n",
    "\n",
    "- **output**: A PipelineData object that corresponds to the output directory. We'll use the output directory we just defined. \n",
    "\n",
    "- **arguments**: A list of arguments required for the train.py entry script. Here, we provide the schema for the timeseries data - i.e. the names of target, timestamp, and id columns - as well as columns that should be dropped prior to modeling, a string identifying the model type, and the number of observations we want to leave aside for testing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from azureml.pipeline.steps import ParallelRunStep\n",
    "\n",
    "parallel_run_step = ParallelRunStep(\n",
    "    name=\"many-models-training\",\n",
    "    parallel_run_config=parallel_run_config,\n",
    "    inputs=[dataset_input],\n",
    "    output=output_dir,\n",
    "    allow_reuse=False,\n",
    "    arguments=['--target_column', 'Quantity', \n",
    "               '--timestamp_column', 'WeekStarting', \n",
    "               '--timeseries_id_columns', 'Store', 'Brand',\n",
    "               '--drop_columns', 'Revenue', 'Store', 'Brand',\n",
    "               '--model_type', 'lr',\n",
    "               '--test_size', 20]\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5.0 Run the pipeline\n",
    "Next, we submit our pipeline to run. The run will train models for each dataset using a train set, compute accuracy metrics for the fits using a test set, and finally re-train models with all the data available. With 10 files, this should only take a few minutes but with the full dataset this can take over an hour."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.core import Pipeline\n",
    "\n",
    "pipeline = Pipeline(workspace=ws, steps=[parallel_run_step])\n",
    "run = experiment.submit(pipeline)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Wait for the run to complete\n",
    "run.wait_for_completion(show_output=False, raise_on_error=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6.0 View results of training pipeline\n",
    "The dataframe we return in the run method of train.py is outputted to *parallel_run_step.txt*. To see the results of our training pipeline, we'll download that file, read in the data to a DataFrame, and then visualize the results, including the in-sample metrics.\n",
    "The run submitted to the Azure Machine Learning Training Compute Cluster may take a while. The output is not generated until the run is complete. You can monitor the status of the run in Azure Portal https://ml.azure.com\n",
    "\n",
    "### 6.1 Download parallel_run_step.txt locally"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "def download_results(run, target_dir=None, step_name='many-models-training', output_name='training_output'):\n",
    "    stitch_run = run.find_step_run(step_name)[0]\n",
    "    port_data = stitch_run.get_output_data(output_name)\n",
    "    port_data.download(target_dir, show_progress=True)\n",
    "    return os.path.join(target_dir, 'azureml', stitch_run.id, output_name)\n",
    "\n",
    "file_path = download_results(run, 'output')\n",
    "file_path"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6.2 Convert the file to a dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "df = pd.read_csv(file_path + '/parallel_run_step.txt', sep=\" \", header=None)\n",
    "df.columns = ['Store', 'Brand', 'Model', 'File Name', 'ModelName', 'StartTime', 'EndTime', 'Duration',\n",
    "              'MSE', 'RMSE', 'MAE', 'MAPE', 'Index', 'Number of Models', 'Status','RunID']\n",
    "\n",
    "df['StartTime'] = pd.to_datetime(df['StartTime'])\n",
    "df['EndTime'] = pd.to_datetime(df['EndTime'])\n",
    "df['Duration'] = df['EndTime'] - df['StartTime']\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6.3 Review Results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total = df['EndTime'].max()  - df['StartTime'].min()\n",
    "\n",
    "print('Number of Models: ' + str(len(df)))\n",
    "print('Total Duration: ' + str(total)[6:])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('Average MAPE: ' + str(round(df['MAPE'].mean(), 5)))\n",
    "print('Average MSE: ' + str(round(df['MSE'].mean(), 5)))\n",
    "print('Average RMSE: ' + str(round(df['RMSE'].mean(), 5)))\n",
    "print('Average MAE: '+ str(round(df['MAE'].mean(), 5)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('Maximum Duration: '+ str(df['Duration'].max())[7:])\n",
    "print('Minimum Duration: ' + str(df['Duration'].min())[7:])\n",
    "print('Average Duration: ' + str(df['Duration'].mean())[7:])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6.4 Visualize Performance across models\n",
    "\n",
    "Here, we produce some charts from the errors metrics calculated during the run using a subset put aside for testing.\n",
    "\n",
    "First, we examine the distribution of mean absolute percentage error (MAPE) over all the models:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import seaborn as sns \n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "fig = sns.boxplot(y='MAPE', data=df)\n",
    "fig.set_title('MAPE across all models')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we can break that down by Brand or Store to see variations in error across our models"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "fig = sns.boxplot(x='Brand', y='MAPE', data=df)\n",
    "fig.set_title('MAPE by Brand')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also look at how long models for different brands took to train"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "brand = df.groupby('Brand')\n",
    "brand = brand['Duration'].sum()\n",
    "brand = pd.DataFrame(brand)\n",
    "brand['time_in_seconds'] = [time.total_seconds()  for time in brand['Duration']]\n",
    "\n",
    "brand.drop(columns=['Duration']).plot(kind='bar')\n",
    "plt.xlabel('Brand')\n",
    "plt.ylabel('Seconds')\n",
    "plt.title('Total Training Time by Brand')\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7.0 Publish and schedule the pipeline (Optional)\n",
    "\n",
    "\n",
    "### 7.1 Publish the pipeline\n",
    "Once you have a pipeline you're happy with, you can publish a pipeline so you can call it programatically later on. See this [tutorial](https://docs.microsoft.com/en-us/azure/machine-learning/how-to-create-your-first-pipeline#publish-a-pipeline) for additional information on publishing and calling pipelines."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# published_pipeline = pipeline.publish(name = 'train_many_models',\n",
    "#                                      description = 'train many models',\n",
    "#                                      version = '1',\n",
    "#                                      continue_on_step_failure = False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 7.2 Schedule the pipeline\n",
    "You can also [schedule the pipeline](https://docs.microsoft.com/en-us/azure/machine-learning/how-to-schedule-pipelines) to run on a time-based or change-based schedule. This could be used to automatically retrain models every month or based on another trigger such as data drift."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# from azureml.pipeline.core import Schedule, ScheduleRecurrence\n",
    "    \n",
    "# training_pipeline_id = published_pipeline.id\n",
    "\n",
    "# recurrence = ScheduleRecurrence(frequency=\"Month\", interval=1, start_time=\"2020-01-01T09:00:00\")\n",
    "# recurring_schedule = Schedule.create(ws, name=\"training_pipeline_recurring_schedule\", \n",
    "#                             description=\"Schedule Training Pipeline to run on the first day of every month\",\n",
    "#                             pipeline_id=training_pipeline_id, \n",
    "#                             experiment_name=experiment.name, \n",
    "#                             recurrence=recurrence)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Next Steps\n",
    "\n",
    "Now that you've trained and scored the models, move on to [03_CustomScript_Forecasting_Pipeline.ipynb](03_CustomScript_Forecasting_Pipeline.ipynb) to make forecasts with your models."
   ]
  }
 ],
 "metadata": {
  "authors": [
   {
    "name": "roastala"
   }
  ],
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}