recipes/Horovod/Horovod-Tensorflow/Horovod-Tensorflow.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Horovod-Tensorflow"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction\n",
    "\n",
    "This recipe shows how to run [Horovod](https://github.com/uber/horovod) distributed training framework for Tensorflow using Batch AI.\n",
    "\n",
    "\n",
    "## Details\n",
    "\n",
    "- Standard Horovod [tensorflow_mnist.py](https://github.com/uber/horovod/blob/v0.9.10/examples/tensorflow_mnist.py) example will be used;\n",
    "- tensorflow_mnist.py downloads training data on its own during execution;\n",
    "- The job will be run on standard tensorflow container tensorflow/tensorflow:1.8.0-gpu;\n",
    "- Horovod framework will be installed in the container using job preparation command line. Note, you can build your own docker image containing tensorflow and horovod instead.\n",
    "- Standard output of the job will be stored on Azure File Share."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Instructions\n",
    "\n",
    "### Install Dependencies and Create Configuration file.\n",
    "Follow [instructions](/recipes) to install all dependencies and create configuration file."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read Configuration and Create Batch AI client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "nbpresent": {
     "id": "bfa11f00-8866-4051-bbfe-a9646e004910"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "No handlers could be found for logger \"msrestazure.azure_active_directory\"\n"
     ]
    }
   ],
   "source": [
    "from __future__ import print_function\n",
    "\n",
    "from datetime import datetime\n",
    "import sys\n",
    "\n",
    "from azure.storage.file import FileService\n",
    "import azure.mgmt.batchai.models as models\n",
    "\n",
    "# The BatchAI/utilities folder contains helper functions used by different notebooks\n",
    "sys.path.append('../../../')\n",
    "import utilities as utils\n",
    "\n",
    "cfg = utils.config.Configuration('../../configuration.json')\n",
    "client = utils.config.create_batchai_client(cfg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create Resource Group and Batch AI workspace if not exists:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "utils.config.create_resource_group(cfg)\n",
    "_ = client.workspaces.create(cfg.resource_group, cfg.workspace, cfg.location).result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Prepare Training Dataset and Script in Azure Storage"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create File Share\n",
    "\n",
    "For this example we will create a new File Share with name `batchaisample` under your storage account. This share will be populated with sample scripts and will contain job's output.\n",
    "\n",
    "**Note** You don't need to create new file share for every cluster. We are doing this in this sample to simplify resource management for you."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Done\n"
     ]
    }
   ],
   "source": [
    "azure_file_share_name = 'batchaisample'\n",
    "service = FileService(cfg.storage_account_name, cfg.storage_account_key)\n",
    "service.create_share(azure_file_share_name, fail_on_exist=False)\n",
    "print('Done')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Deploy Sample Script and Configure the Input Directories\n",
    "\n",
    "- Download original sample script:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Downloading https://raw.githubusercontent.com/uber/horovod/v0.9.10/examples/tensorflow_mnist.py ...Done\n"
     ]
    }
   ],
   "source": [
    "sample_script_url = 'https://raw.githubusercontent.com/uber/horovod/v0.9.10/examples/tensorflow_mnist.py'\n",
    "utils.dataset.download_file(sample_script_url, 'tensorflow_mnist.py')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Create a folder in the file share and upload the sample script to it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "samples_dir = 'horovod_samples'\n",
    "service = FileService(cfg.storage_account_name, cfg.storage_account_key)\n",
    "service.create_directory(\n",
    "    azure_file_share_name, samples_dir, fail_on_exist=False)\n",
    "service.create_file_from_path(\n",
    "    azure_file_share_name, samples_dir, 'tensorflow_mnist.py', 'tensorflow_mnist.py')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Create Azure Batch AI Compute Cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Configure Compute Cluster\n",
    "- For this example we will use a GPU cluster of 2 `STANDARD_NC6` nodes. You can increase the number of nodes by changing `nodes_count` variable;\n",
    "- We will call the cluster `nc6`;\n",
    "\n",
    "So, the cluster will have the following parameters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "nodes_count = 2\n",
    "cluster_name = 'nc6'\n",
    "\n",
    "parameters = models.ClusterCreateParameters(\n",
    "    location=cfg.location,\n",
    "    vm_size='STANDARD_NC6',\n",
    "    scale_settings=models.ScaleSettings(\n",
    "        manual=models.ManualScaleSettings(target_node_count=nodes_count)\n",
    "    ),\n",
    "    user_account_settings=models.UserAccountSettings(\n",
    "        admin_user_name=cfg.admin,\n",
    "        admin_user_password=cfg.admin_password or None,\n",
    "        admin_user_ssh_public_key=cfg.admin_ssh_key or None,\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Compute Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "_ = client.clusters.create(cfg.resource_group, cfg.workspace, cluster_name, parameters).result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Monitor Cluster Creation\n",
    "\n",
    "The `utilities` module contains a helper function allowing to wait for the cluster to become available - all nodes are allocated and finished preparation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Cluster state: steady Target: 2; Allocated: 2; Idle: 2; Unusable: 0; Running: 0; Preparing: 0; Leaving: 0\n"
     ]
    }
   ],
   "source": [
    "cluster = client.clusters.get(cfg.resource_group,cfg.workspace, cluster_name)\n",
    "utils.cluster.print_cluster_status(cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Run Azure Batch AI Training Job"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Configure Job\n",
    "\n",
    "- The job will use `tensorflow/tensorflow:1.8.0-gpu` container.\n",
    "- Horovod framework will be installed by job preparation command line;\n",
    "- Will mount file share at folder with name `afs`. Full path of this folder on a computer node will be `$AZ_BATCHAI_JOB_MOUNT_ROOT/afs`;\n",
    "- The model output will be stored in File Share. The job will be able to reference this directory as `$AZ_BATCHAI_OUTPUT_MODEL` and we will be able to enumerate files in this directory using `MODEL` id.\n",
    "- Will output standard output and error streams to file share.\n",
    "\n",
    "You can delete ```container_settings``` from the job definition to run the job directly on host DSVM."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "azure_file_share = 'afs'\n",
    "parameters = models.JobCreateParameters(\n",
    "     location=cfg.location,\n",
    "     cluster=models.ResourceId(id=cluster.id),\n",
    "     node_count=2,\n",
    "     input_directories=[\n",
    "        models.InputDirectory(\n",
    "            id='SCRIPTS',\n",
    "            path='$AZ_BATCHAI_JOB_MOUNT_ROOT/{0}/{1}'.format(azure_file_share, samples_dir))],\n",
    "     std_out_err_path_prefix='$AZ_BATCHAI_JOB_MOUNT_ROOT/{0}'.format(azure_file_share),\n",
    "     mount_volumes=models.MountVolumes(\n",
    "            azure_file_shares=[\n",
    "                models.AzureFileShareReference(\n",
    "                    account_name=cfg.storage_account_name,\n",
    "                    credentials=models.AzureStorageCredentialsInfo(\n",
    "                        account_key=cfg.storage_account_key),\n",
    "                    azure_file_url='https://{0}.file.core.windows.net/{1}'.format(\n",
    "                        cfg.storage_account_name, azure_file_share_name),\n",
    "                    relative_mount_path=azure_file_share)\n",
    "            ]\n",
    "        ),  \n",
    "     container_settings=models.ContainerSettings(\n",
    "         image_source_registry=models.ImageSourceRegistry(image='tensorflow/tensorflow:1.8.0-gpu')),\n",
    "     job_preparation=models.JobPreparation(\n",
    "         command_line='apt update; apt install mpi-default-dev mpi-default-bin -y; pip install horovod'),\n",
    "     horovod_settings = models.HorovodSettings(\n",
    "         python_script_file_path='$AZ_BATCHAI_INPUT_SCRIPTS/tensorflow_mnist.py')\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a training Job and wait for Job completion\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Created Job horovod_07_05_2018_225502 in Experiment horovod_experiment\n"
     ]
    }
   ],
   "source": [
    "experiment_name = 'horovod_experiment'\n",
    "experiment = client.experiments.create(cfg.resource_group, cfg.workspace, experiment_name).result()\n",
    "job_name = datetime.utcnow().strftime('horovod_%m_%d_%Y_%H%M%S')\n",
    "job = client.jobs.create(cfg.resource_group, cfg.workspace, experiment_name, job_name, parameters).result()\n",
    "print('Created Job {0} in Experiment {1}'.format(job.name, experiment.name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Wait for Job to Finish\n",
    "The job will start running when the cluster will have enough idle nodes. The following code waits for job to start running printing the cluster state. During job run, the code prints current content of stderr.txt.\n",
    "\n",
    "**Note** Execution may take several minutes to complete."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Cluster state: steady Target: 2; Allocated: 2; Idle: 2; Unusable: 0; Running: 0; Preparing: 0; Leaving: 0\n",
      "Job state: running ExitCode: None\n",
      "Waiting for job output to become available...\n"
     ]
    }
   ],
   "source": [
    "utils.job.wait_for_job_completion(client, cfg.resource_group, cfg.workspace, \n",
    "                                  experiment_name, job_name, cluster_name, 'stdouterr', 'stderr.txt')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### List stdout.txt and stderr.txt files for the Job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "files = client.jobs.list_output_files(cfg.resource_group, cfg.workspace, experiment_name, job_name,\n",
    "                                      models.JobsListOutputFilesOptions(outputdirectoryid='stdouterr')) \n",
    "for f in list(files):\n",
    "    print(f.name, f.download_url or 'directory')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Clean Up (Optional)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Delete the Job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "_ = client.jobs.delete(cfg.resource_group, cfg.workspace, experiment_name, job_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Delete the Cluster\n",
    "When you are finished with the sample and don't want to submit any more jobs you can delete the cluster using the following code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "_= client.clusters.delete(cfg.resource_group, cfg.workspace, cluster_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Delete File Share\n",
    "When you are finished with the sample and don't want to submit any more jobs you can delete the file share completely with all files using the following code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "service = FileService(cfg.storage_account_name, cfg.storage_account_key)\n",
    "service.delete_share(azure_file_share_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.15"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}