Feature-Extraction-Orchestration .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": [
    "## Introduction \n",
    "\n",
    "This note book provide a generic tool to create distrebuted jobs on a cluster of machines or clusters. The types of jobs could vary from feature extractions to other training jobs. The type of machines should be selected per type of job.\n",
    "\n",
    "**The following will focus on private image from acr**\n",
    "\n",
    "\n",
    "### Each job has the following attributes:\n",
    "All jobs are identical. they receive the input queues and other storage related parameters.\n",
    "\n",
    "### Image \n",
    "Image should be created and validated prior to start of experiments\n",
    "The service provide the capability to create image based on base image with specific conda or pip packages.\n",
    "The best practice is to use conda packages where available, since they handle dependencies.\n",
    "### Main Script file\n",
    "The script created by the relavant algo team. This is where the main logic reside. It can read from the input file/folders perform its activities and write an output per the internal logic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check core SDK version number\n",
    "import azureml.core\n",
    "import hdutils as utils\n",
    "\n",
    "print(\"SDK version:\", azureml.core.VERSION)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## subscrition connection etc - load configuration\n",
    "Only applicable when you have more than one active subscription"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!az login"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#!az account set --subscription \"XXXXXXX\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Initialize Workspace\n",
    "\n",
    "Initialize a workspace object from persisted configuration.\n",
    "The config file holds information on the subcsription resource group"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core.workspace import Workspace\n",
    "\n",
    "ws = Workspace.from_config()\n",
    "print('Workspace name: ' + ws.name, \n",
    "      'Azure region: ' + ws.location, \n",
    "      'Subscription id: ' + ws.subscription_id, \n",
    "      'Resource group: ' + ws.resource_group, sep = '\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create / Update project folder\n",
    "Assuming the entry script, with the supporting scripts are pushed to the jupyter vm.\n",
    "Addtional settings may be in place, to correspond to specific file location of the scripts."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "project_folder = './mobilenet'\n",
    "common = './common'\n",
    "supporting_scripts = '<any supporting scripts>'\n",
    "entry_script = '<main entry script>.py'\n",
    "# sub folder of supporting scripts - no need to create the folder\n",
    "ss_folder = '{0}/{1}'.format(common,supporting_scripts)\n",
    "\n",
    "os.makedirs(project_folder, exist_ok=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import shutil\n",
    "# assume the supporting scripts are located at:\n",
    "local_ss_folder = './{0}'.format(supporting_scripts)\n",
    "\n",
    "#shutil.copy('{0}/{1}'.format(project_folder,entry_script), project_folder)\n",
    "# recursive copy of the supporting scripts folder - if the folder exist, it will fail.\n",
    "#shutil.copytree(local_ss_folder, ss_folder)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create Experiment\n",
    "\n",
    "**Experiment** is a logical container in an Azure ML Workspace. It hosts run records which can include run metrics and output artifacts from your experiments."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "experiment_name = '<meaning full name>'\n",
    "\n",
    "from azureml.core import Experiment\n",
    "exp = Experiment(workspace=ws, name=experiment_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get the default datastore\n",
    "ds = ws.get_default_datastore()\n",
    "print(ds.name, ds.datastore_type, ds.account_name, ds.container_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create or Attach existing AmlCompute\n",
    "You will need to create a [compute target](https://docs.microsoft.com/azure/machine-learning/service/concept-azure-machine-learning-architecture#compute-target) for training your model. In this tutorial, you create `AmlCompute` as your training compute resource.\n",
    "\n",
    "**Creation of AmlCompute takes approximately 5 minutes.** If the AmlCompute with that name is already in your workspace this code will skip the creation process.\n",
    "\n",
    "As with other Azure services, there are limits on certain resources (e.g. AmlCompute) associated with the Azure Machine Learning service. Please read [this article](https://docs.microsoft.com/en-us/azure/machine-learning/service/how-to-manage-quotas) on the default limits and how to request more quota."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Prepare input & output\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Specific Storage Account\n",
    "Used to host outcome of the jobs (pny files)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Datastore\n",
    "prvt_ds_name = '<storage name>'\n",
    "prvt_ds_container_name = '<container name>'\n",
    "prvt_ds_key = '<sa key>'\n",
    "prvt_ds = Datastore.register_azure_blob_container(workspace=ws, \n",
    "                                             datastore_name=prvt_ds_name, \n",
    "                                             container_name=prvt_ds_container_name,\n",
    "                                             account_name=prvt_ds_name, \n",
    "                                             account_key=prvt_ds_key,\n",
    "                                             create_if_not_exists=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# ensure there is a output & work directories \n",
    "prvt_ds.upload(src_dir='output', target_path=\"./<output>\")\n",
    "prvt_ds.upload(src_dir='output', target_path=\"./<in case transient area is required by your script>\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### create links to the output and work directory\n",
    "To be used as links within the script parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Using specific storage account\n",
    "work_path = prvt_ds.path('<match from above box>')\n",
    "out_path = prvt_ds.path('<match from above box>')\n",
    "\n",
    "print(work_path)\n",
    "print(out_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Private Image\n",
    "The image is taken from specific acr, the registry credentials are currently open - will need to find a better way to handle them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "acr_username='<private acr user name>'\n",
    "acr_pass = '<private acr password>'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Run Configuration\n",
    "The run configration specify, the image name, its credentials, use of gpu, python interperter etc. it is required to be passed to an estimator. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import RunConfiguration\n",
    "from azureml.train.estimator import *\n",
    "\n",
    "rc = RunConfiguration()\n",
    "rc.environment.docker.enabled = True\n",
    "\n",
    "# this is an image available in Docker Hub\n",
    "#rc.environment.docker.base_image = '/feature_extraction:0.1'\n",
    "\n",
    "# point to an image in a private ACR\n",
    "rc.environment.docker.base_image = \"<image name>:<image tag>\"\n",
    "rc.environment.docker.base_image_registry.address = \"<private acr>\"\n",
    "rc.environment.docker.base_image_registry.username = acr_username\n",
    "rc.environment.docker.base_image_registry.password = acr_pass\n",
    "rc.environment.docker.gpu_support = True\n",
    "\n",
    "# don't let the system build a new conda environment\n",
    "rc.environment.python.user_managed_dependencies = True\n",
    "# point to an existing python environment \n",
    "rc.environment.python.interpreter_path = '/usr/bin/python3'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Script Parms & Queue\n",
    "The experiment would receive parameters via the estimator configuration. it is provided as a dictionary. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## specific to queues - need to provide the storage account and key\n",
    "account_name = '<sa for queues>'\n",
    "account_key = 'sa key'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azure.storage.queue import QueueService\n",
    "\n",
    "input_queue = '<input-queue-name>'\n",
    "# the main script is using same naming convention\n",
    "success_queue = '{0}-success'.format(input_queue)\n",
    "fail_queue = '{0}-fail'.format(input_queue)\n",
    "queue_service = QueueService(account_name=account_name, account_key=account_key)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Load data into queues"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_file = '<file to be broken to queue messages>.csv'\n",
    "utils.csv_to_queue(input_file,input_queue,queue_service,2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### define the script params"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "script_params_q = {    \n",
    "    '-inqueue': input_queue, # input q\n",
    "    '-sa' : account_name, # storage account of queue\n",
    "    '-sakey':account_key, # key of the storage account\n",
    "    '-npy': out_path.as_mount(), #output blob\n",
    "    '-workdir': work_path.as_mount() #temporary folder (blob)    \n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### RandomParameterSampling - run once"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.train.hyperdrive import RandomParameterSampling, BanditPolicy, HyperDriveRunConfig, PrimaryMetricGoal\n",
    "ps = RandomParameterSampling( {  } )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Multi Experiment Run \n",
    "Will loop over available compute cluster and submit experiment runs to them\n",
    "\n",
    "The naming convention for a cluster names\n",
    "* Prefix is the VM type, e.g NC6\n",
    "* L - low priority VM\n",
    "* D - Dedicated VM\n",
    "* Number of maximum nodes\n",
    "* Random suffix 2 letters & 2 digits\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# shows the current avilable clusters\n",
    "compute_targets = ws.compute_targets\n",
    "for name, ct in compute_targets.items():\n",
    "    print(name, ct.type, ct.provisioning_state)   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Print experiment details\n",
    "print('Script directory {0}'.format(project_folder))\n",
    "print('Script {0}'.format(entry_script))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create compute clusters - configuration\n",
    "node_count = 72\n",
    "max_node_count = 100\n",
    "num_of_low_clusters = 5\n",
    "num_of_dedicated_clusters = 9\n",
    "idle_timeout = 30 # seconds"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# if the py file changes, this is the way to reload the module\n",
    "#import importlib\n",
    "#importlib.reload(utils)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clust = utils.create_low_clusters(ws,num_of_low_clusters,max_node_count,idle_timeout)\n",
    "clust = utils.create_dedicated_clusters(ws,num_of_dedicated_clusters,max_node_count,idle_timeout)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# loop over avilable clusters\n",
    "from azureml.train.dnn import TensorFlow\n",
    "from azureml.train.hyperdrive import RandomParameterSampling, BanditPolicy, HyperDriveRunConfig, PrimaryMetricGoal\n",
    "from time import gmtime, strftime, sleep\n",
    "\n",
    "print(\"Submitting experiments::\")\n",
    "print(strftime(\"%Y-%m-%d %H:%M:%S\", gmtime()))\n",
    "\n",
    "cluster_size = node_count\n",
    "compute_targets = ws.compute_targets\n",
    "htrDic = {}\n",
    "\n",
    "for name, ct in compute_targets.items():\n",
    "    \n",
    "    est = TensorFlow(source_directory=project_folder,\n",
    "                 script_params=script_params_q,\n",
    "                 compute_target=ct,\n",
    "                 entry_script=entry_script,\n",
    "                 environment_definition=rc.environment,\n",
    "                 source_directory_data_store=ds\n",
    "                )\n",
    "    htc = HyperDriveRunConfig(estimator=est, \n",
    "                          hyperparameter_sampling=ps, \n",
    "                          primary_metric_name='somemetric', \n",
    "                          primary_metric_goal=PrimaryMetricGoal.MAXIMIZE, \n",
    "                          max_total_runs=cluster_size,\n",
    "                          max_concurrent_runs=cluster_size)    \n",
    "    htrDic[name] = exp.submit(config=htc)\n",
    "    # add 46 seconds delay (time it takes to submit 70 jobs)    \n",
    "    print(\"Submited on cluster: {0} at {1}\".format(name,strftime(\"%Y-%m-%d %H:%M:%S\", gmtime())))\n",
    "    sleep(46)\n",
    "    \n",
    "print(\"Finished submitting experiments::\")\n",
    "print(strftime(\"%Y-%m-%d %H:%M:%S\", gmtime()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Delete all compute - once experiment is completed\n",
    "utils.remove_compute(ws.compute_targets)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# export results to a file\n",
    "output_file = '<output file to be created>'\n",
    "auxq = '<if required>'\n",
    "#utils.push_results_to_file(output_file,success_queue,auxq,queue_service)\n",
    "utils.queue_to_csv(output_file,success_queue,queue_service)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## enable experiment run cancelation\n",
    "for item in htrDic:\n",
    "    print('canceling {}'.format(item))\n",
    "    htrDic[item].cancel()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# single cancel - test\n",
    "htrDic['NC6low100-532Vgb'].cancel()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Single Experiment run\n",
    "\n",
    "#### Submitting an experiment\n",
    "For each Experiment:\n",
    "\n",
    "1. establish compute\n",
    "2. create an estimator\n",
    "3. create Hyper Drive config\n",
    "4. submit"
   ]
  },
  {
   "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",
    "# set a cluster to test on\n",
    "cluster_name = 'NC6-L100-21fj'\n",
    "\n",
    "\n",
    "try:\n",
    "    compute_target = ComputeTarget(workspace=ws, name=cluster_name)\n",
    "    print('Found existing compute target.')\n",
    "except ComputeTargetException:\n",
    "    print('Please run configuration notebook, or contact devops')\n",
    "\n",
    "print(compute_target.status.serialize())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.  Estimator Configuration\n",
    "An estimator can be generic, or specific sch as TensorFlow. In the sample below there are two specific parameters that are passed to build a cluster of execution nodes. each process is on dedicated node from the cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.train.dnn import TensorFlow\n",
    "est = TensorFlow(source_directory=project_folder,\n",
    "                 script_params=script_params_q,\n",
    "                 compute_target=compute_target,\n",
    "                 entry_script=entry_script,\n",
    "                 environment_definition=rc.environment,\n",
    "                 source_directory_data_store=ds\n",
    "                )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3. HyperDriveRunConfig\n",
    "max_total_runs and max_concurrent_runs should be equal to the number of nodes up to **100**\n",
    "since this is a test run - best to test with single digits number of nodes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.train.hyperdrive import RandomParameterSampling, BanditPolicy, HyperDriveRunConfig, PrimaryMetricGoal\n",
    "htc = HyperDriveRunConfig(estimator=est, \n",
    "                          hyperparameter_sampling=ps, \n",
    "                          primary_metric_name='somemetric', \n",
    "                          primary_metric_goal=PrimaryMetricGoal.MAXIMIZE, \n",
    "                          max_total_runs=1,\n",
    "                          max_concurrent_runs=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4. Submit\n",
    "submitting the estimator, would create anoter run under the experiment folder. it is best to view the progress of large scale experiments from the portal. (or by examining the queues)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Runing on cluster: \" + cluster_name)\n",
    "htr = exp.submit(config=htc)\n",
    "print(htr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Cancel run when required"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# htr2 htr3\n",
    "\n",
    "htr.cancel()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Monitor progress (via queues)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from time import gmtime, strftime\n",
    "metadata = queue_service.get_queue_metadata(input_queue)\n",
    "count = metadata.approximate_message_count\n",
    "print(strftime(\"%Y-%m-%d %H:%M:%S\", gmtime()))\n",
    "print('todo:'+str(count))\n",
    "metadata = queue_service.get_queue_metadata(success_queue)\n",
    "count = metadata.approximate_message_count\n",
    "print('good:'+str(count))\n",
    "metadata = queue_service.get_queue_metadata(fail_queue)\n",
    "count = metadata.approximate_message_count\n",
    "print('bad:'+str(count))\n",
    "auxq = 'soccer-feature-input'\n",
    "metadata = queue_service.get_queue_metadata(auxq)\n",
    "count = metadata.approximate_message_count\n",
    "print('aux:'+str(count))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "messages = queue_service.get_messages(fail_queue,8)\n",
    "for message in messages:\n",
    "    print(message.content)\n",
    "    #queue_service.delete_message(input_queue, message.id, message.pop_receipt)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Canceling/Failing an experiment - by ID"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import get_run \n",
    "\n",
    "r = get_run(experiment=exp, run_id=\"fe-Hokey_1551955099682\", rehydrate=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "r.fail()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "compute_targets = ws.compute_targets\n",
    "for name, ct in compute_targets.items():\n",
    "    print(name, ct.type, ct.provisioning_state)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python (my-kernel)",
   "language": "python",
   "name": "my-kernel"
  },
  "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}