scoring_on_spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Scoring images on Spark\n",
    "\n",
    "This notebook illustrates how trained Cognitive Toolkit (CNTK) and TensorFlow models can be applied to large image collections using PySpark.\n",
    "\n",
    "This notebook is part of the [Embarrassingly Parallel Image Classification](https://github.com/Azure/Embarrassingly-Parallel-Image-Classification) git repository. It assumes that a dataset and Azure N-series GPU VM have already been created for model training as described in the previous [Image Set Preparation](https://github.com/Azure/Embarrassingly-Parallel-Image-Classification/blob/master/image_set_preparation.ipynb) notebook. Note that an abbreviated instruction set is mentioned in that notebook for users who would like to employ our sample image set rather than generating their own.\n",
    "\n",
    "By default, this notebook uses our provided retrained DNNs. If you have completed the [Model Training](https://github.com/Azure/Embarrassingly-Parallel-Image-Classification/blob/master/model_training.ipynb) notebook in this repository, you can elect to use your own models by modifying the script action used during CNTK and TensorFlow installation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "## Outline\n",
    "- [Set up a Microsoft HDInsight Spark cluster and Azure Data Lake Store](#setup)\n",
    "   - [Provision the resources](#provision)\n",
    "      - [Azure Data Lake Store](#adls)\n",
    "      - [Azure HDInsight Spark cluster](#hdinsight)\n",
    "      - [Check cluster deployment status](#checkstatus)\n",
    "   - [Install Cognitive Toolkit, TensorFlow, and model files](#install)\n",
    "      - [Run the script action](#runsa)\n",
    "      - [Update the Python 3 path](#updatepath)\n",
    "- [Image scoring with PySpark](#pyspark)\n",
    "   - [Define functions/variables/RDDs used by both scoring pipelines](#shared)\n",
    "   - [Cognitive Toolkit](#cntk)\n",
    "      - [Make the trained CNTK model available to all workers](#cntkbroadcast)\n",
    "      - [Define functions to be run by worker nodes](#cntkworker)\n",
    "      - [Score all test set images with the trained model](#cntkscore)\n",
    "      - [Evaluate the model's performance](#cntkevaluate)\n",
    "   - [TensorFlow](#tf)\n",
    "      - [Make the trained TensorFlow model available to all workers](#tfbroadcast)\n",
    "      - [Define functions to be run by worker nodes](#tfworker)\n",
    "      - [Score all test set images with the trained model](#tfscore)\n",
    "      - [Evaluate the model's performance](#tfevaluate)\n",
    "- [Next steps](#next)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"setup\"></a>\n",
    "## Set up a Microsoft HDInsight Spark cluster and associated Azure Data Lake Store\n",
    "\n",
    "In the previous notebooks, we illustrated how to use a Windows Data Science Virtual Machine to create the training/validation image sets and train DNNs for this image classification task. In this section, we illustrate how to transfer the data and models to an Azure Data Lake Store (a cloud-based HDFS). We also show how to provision and set up the HDInsight Spark cluster which will apply the models to the data in subsequent sections."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a name=\"provision\"></a>\n",
    "### Provision the resources\n",
    "\n",
    "<a name=\"adls\"></a>\n",
    "#### Azure Data Lake Store\n",
    "We have provided directions for provisioning and setting up the ADLS through the [Azure CLI 2.0](https://docs.microsoft.com/en-us/cli/azure/install-azure-cli?view=azure-cli-latest), which comes preinstalled on the Data Science Virtual Machine.\n",
    "\n",
    "1. From a command prompt, log into the Azure CLI by running the following command:\n",
    "\n",
    "    ```\n",
    "    az login\n",
    "    ```\n",
    "\n",
    "    You will be asked to visit a website and type in a temporary code. The website may ask you to provide your Azure account credentials.\n",
    "1. When login is complete, return to the CLI and run the following command to determine which subscriptions are available in your account:\n",
    "\n",
    "    ```\n",
    "    az account list\n",
    "    ```\n",
    "\n",
    "    Copy the \"id\" of the subscription you would like to use when creating resources, then execute the command below to set it as the active subscription:\n",
    "\n",
    "    ```\n",
    "    az account set --subscription [subscription id]\n",
    "    ```\n",
    "\n",
    "1. Choose a unique resource group name, then create an Azure resource group using the command below:\n",
    "\n",
    "    ```\n",
    "    set RESOURCE_GROUP_NAME=[your resource group name]\n",
    "    az group create --location eastus2 --name %RESOURCE_GROUP_NAME%\n",
    "    ```\n",
    "    \n",
    "1. Choose a unique Data Lake Store name, then create the resource with the command below:\n",
    "\n",
    "    ```\n",
    "    set DATA_LAKE_STORE_NAME=[your data lake store name]\n",
    "    az dls account create --account %DATA_LAKE_STORE_NAME% --resource-group %RESOURCE_GROUP_NAME% --location eastus2\n",
    "    az dls fs create --account %DATA_LAKE_STORE_NAME% --path /clusters --folder\n",
    "    ```\n",
    "    \n",
    "1. Upload the balanced validation dataset from your DSVM to the appropriate folder of the ADLS (expect this step to take ~10 minutes to complete):\n",
    "    ```\n",
    "    az dls fs upload --account %DATA_LAKE_STORE_NAME% --source-path \"D:\\balanced_validation_set\" --destination-path \"/balancedvalidationset\"\n",
    "    ```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a name=\"hdinsight\"></a>\n",
    "#### Azure HDInsight Spark cluster\n",
    "\n",
    "We provide instructions for creating the HDInsight Spark cluster through the Azure Portal, a graphical interface:\n",
    "\n",
    "1. After logging into [Azure Portal](https://ms.portal.azure.com), click the \"+ New\" button near the upper left to create a new resource.\n",
    "1. In the search field that appears, enter \"HDInsight\" and press Enter.\n",
    "1. In the search results, click on the \"HDInsight\" option published by Microsoft.\n",
    "1. Click the \"Create\" button at the bottom of the new pane that opens to describe the HDInsight resource type.\n",
    "1. In the \"Basics\" section of the \"New HDInsight cluster\" pane:\n",
    "    1. Choose a unique cluster name and the appropriate subscription.\n",
    "    1. Click on \"Cluster configuration\" to load a pane of settings.\n",
    "       1. Set the cluster type to \"Spark\".\n",
    "       1. Set the version to \"Spark 2.1 (HDI 3.6)\".\n",
    "       1. Click the \"Select\" button at the bottom of the pane.\n",
    "    1. Choose a password for the `admin` account. You will use this account to log into Jupyter Notebook later in the walkthrough.\n",
    "    1. Select the resource group and location where your Data Lake Store is located.\n",
    "    1. Click the \"Next\" button at the bottom of the pane.\n",
    "1. In the \"Storage\" section of the \"New HDInsight cluster\" pane:\n",
    "   1. Ensure that \"Data Lake Store\" is selected for the \"Primary storage type\".\n",
    "   1. Click on \"Select Data Lake Storage Account\" to load a pane of settings.\n",
    "       1. Under \"Select a storage account\", select your Azure Data Lake store.\n",
    "       1. Click on \"Configure Data Lake Access\" to load a pane of settings.\n",
    "           1. Create a new service principal with the name and password of your choice. Save the generated certificate file.\n",
    "           1. Click on \"Access\" to load a pane of settings.\n",
    "               1. Under \"Select File Permissions\", click the box to the left of your ADLS name. (The box may not be visible until mouseover.). Click \"Select\".\n",
    "               1. Under \"Assign Selected Permissions\", click \"Run\".\n",
    "               1. When the run completes, click \"Done\".\n",
    "       1. Click the \"Next\" button at the bottom of the pane.\n",
    "1. In the \"Summary\" section of the \"New HDInsight cluster\" pane:\n",
    "   - If desired, you can edit the cluster size settings to choose node counts/sizes based on your budget and time constraints. We recommend completing this tutorial using a cluster with **10** worker nodes and a node size of **D12 v2** (for both worker and head nodes).\n",
    "   - For more information, please see the [cluster](https://docs.microsoft.com/en-us/azure/hdinsight/hdinsight-hadoop-provision-linux-clusters) and [VM](https://docs.microsoft.com/en-us/azure/virtual-machines/virtual-machines-linux-sizes#dv2-series) size guides.\n",
    "1. Click the \"Create\" button at the bottom of the pane."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a name=\"checkstatus\"></a>\n",
    "#### Check cluster deployment status\n",
    "\n",
    "Cluster deployment may take approximately twenty minutes. (We recommend that you continue with the tutorial, creating all other Azure resources and transferring your image set to the Azure Data Lake Store, while you wait for the cluster to deploy.) Cluster deployment status can be checked as follows:\n",
    "1. Click on the \"Search Resources\" magnifying glass icon along the top bar of [Azure Portal](https://ms.portal.azure.com).\n",
    "1. Type in the name of your HDInsight cluster and click on its entry in the resulting drop-down list. The overview pane for your HDInsight cluster will appear.\n",
    "1. During deployment, a blue bar will appear across the top of the overview pane with the title \"Applying changes\". When this bar disappears, deployment is complete."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"install\"></a>\n",
    "### Install Cognitive Toolkit, TensorFlow, and model files\n",
    "\n",
    "Once the HDInsight Spark cluster deployment is complete, we will run a script action to install CNTK, TensorFlow, and the sample trained model files we provide. As of this writing, the script action will install CNTK 2.1 and TensorFlow 1.2.\n",
    "\n",
    "If you completed the previous model training notebook and would prefer to use your own model files, you can modify the script action to download the model files you created from the online location of your choice. Simply download a copy of the sample script action in the `scoring` subdirectory of [the Embarrassingly Parallel Image Classification git repository](https://github.com/Azure/Embarrassingly-Parallel-Image-Classification), edit the `wget` commands to point to your own model files, and upload the modified script action to the online location of your choosing."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a name=\"runsa\"></a>\n",
    "#### Run the script action\n",
    "\n",
    "After HDInsight cluster deployment finishes, run the script action to install CNTK as follows:\n",
    "1. Obtain the URI for the script action.\n",
    "   - If using the unmodified version in the `scoring` subdirectory of this git repo, ensure that your URI points to the \"raw\" file (not a webpage-embedded file), e.g.:\n",
    "   [https://github.com/Azure/Embarrassingly-Parallel-Image-Classification/raw/master/scoring/script_action.sh](https://github.com/Azure/Embarrassingly-Parallel-Image-Classification/raw/master/scoring/script_action.sh)\n",
    "   - If you have modified the script action, upload it to the website or Azure Blob Storage account of your choice and note its URI.\n",
    "1. Click on the \"Search Resources\" magnifying glass icon along the top bar of [Azure Portal](https://ms.portal.azure.com).\n",
    "1. Type in the name of your HDInsight cluster and click on its entry in the resulting drop-down list. The overview pane for your HDInsight cluster will appear.\n",
    "1. In the search field at upper left, type in \"Script actions\". Click the \"Script actions\" option in the results list.\n",
    "1. Click the \"+ Submit new\" button along the top of the Script Actions pane. A new pane of options will appear.\n",
    "   1. Under name, type \"install\" (without the quotes).\n",
    "   1. Under \"Bash script URI\", type in the URI.\n",
    "   1. Ensure that \"Head\" and \"Worker\" boxes are checked.\n",
    "   1. Click the \"Create\" button along the bottom of the pane.\n",
    "   \n",
    "Expect the script action to take roughly fifteen minutes to run. When the script action is complete, the blue bar at the top of the screen will disappear and a green icon will appear next to the submitted script action's name. Do not proceed until the script action has finished."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"updatepath\"></a>\n",
    "#### Update the Python 3 path\n",
    "\n",
    "The script action above installed Cognitive Toolkit and TensorFlow under a new Python environment, `cntk-py35`. Follow the steps below to direct PySpark to use this new environment:\n",
    "\n",
    "1. Navigate back to the HDInsight cluster's overview pane by clicking \"Overview\" near the upper left of the pane.\n",
    "1. Under \"Quick links\" in the main window, click the \"Cluster dashboards\" button. A new pane of dashboard options will appear.\n",
    "1. Click \"HDInsight cluster dashboard\". A new window will load. You may be asked for the username (default: admin) and password you selected during deployment.\n",
    "1. In the menu at left, click \"Spark2\".\n",
    "1. In the main window, click on the \"Configs\" tab.\n",
    "1. Scroll down to the \"Custom spark2-defaults\" option and expand its dropdown by clicking on the label (or triange beside it).\n",
    "1. Find the `spark.yarn.appMasterEnv.PYSPARK3_PYTHON` entry in the dropdown list. Change its path to the following:\n",
    "\n",
    "    `/usr/bin/anaconda/envs/cntk-py35/bin/python`<br/><br/>\n",
    "    \n",
    "1. Click on the green \"Save\" button that appears at upper right.\n",
    "    - You may receive a warning regarding a setting that you did not change. This value was set by default during HDInsight cluster deployment; the warning can be safely disregarded.\n",
    "1. When prompted, click the orange \"Restart\" button and select \"Restart all affected\".\n",
    "1. When the restart concludes, close the window. This will return you to a pane of dashboard options."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "### Upload and start this notebook on HDInsight Spark\n",
    "\n",
    "1. On the Overview pane of your HDInsight Spark cluster, click on \"Cluster dashboards\" and select \"Jupyter Notebooks\".\n",
    "1. If prompted, log in with the username (default: admin) and password you selected during deployment.\n",
    "1. Use the \"Upload\" button to upload a copy of this notebook. You may be prompted to confirm the destination filename during the upload process: the default value will do.\n",
    "1. Once the notebook has been uploaded, double-check on the notebook's name to launch it.\n",
    "1. The PySpark3 kernel should be used to run the notebook. If necessary, change the kernel by clicking \"Kernel -> Change Kernel -> PySpark3\".\n",
    "\n",
    "If you intend to execute the code cells below, please switch to using the notebook copy you've just opened on the Spark cluster. (Note the outline at the top of the notebook includes a hotlink to the \"Image scoring with PySpark\" section."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"pyspark\"></a>\n",
    "## Image scoring with PySpark\n",
    "<a name=\"shared\"></a>\n",
    "### Define functions/variables/RDDs used by both scoring pipelines\n",
    "\n",
    "Edit the variables below to define the name of your Azure Data Lake Store and the folder where the images have been stored. Execute the code cell to create an RDD of the images in the test set. Note that if this is the first code cell executed, there will be an additional delay as the Spark connection initiates."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>0</td><td>application_1504722417824_0006</td><td>pyspark3</td><td>idle</td><td><a target=\"_blank\" href=\"http://hn1-mawahd.w0jjvdca0zye3nndw25ggycjfa.cx.internal.cloudapp.net:8088/proxy/application_1504722417824_0006/\">Link</a></td><td><a target=\"_blank\" href=\"http://10.0.0.13:30060/node/containerlogs/container_1504722417824_0006_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from io import BytesIO\n",
    "from PIL import Image\n",
    "from pyspark import SparkFiles\n",
    "\n",
    "label_to_number_dict = {'Barren': 0,\n",
    "                        'Forest': 1,\n",
    "                        'Shrub': 2,\n",
    "                        'Cultivated': 3,\n",
    "                        'Herbaceous': 4,\n",
    "                        'Developed': 5}\n",
    "\n",
    "def get_nlcd_id(my_filename):\n",
    "    ''' Extracts the true label  '''\n",
    "    folder, _ = os.path.split(my_filename)\n",
    "    return(label_to_number_dict[os.path.basename(folder)])\n",
    "\n",
    "adls_name = 'mawahdemo2'\n",
    "adls_folder = 'balancedvalidationset'\n",
    "\n",
    "n_workers = 10\n",
    "local_tmp_dir = '/tmp/models'\n",
    "\n",
    "dataset_dir = 'adl://{}.azuredatalakestore.net/{}'.format(adls_name, adls_folder)\n",
    "image_rdd = sc.binaryFiles('{}/*/*.png'.format(dataset_dir), minPartitions=n_workers).coalesce(n_workers)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"cntk\"></a>\n",
    "### Score and evaluate with a trained Cognitive Toolkit (CNTK) model\n",
    "<a name=\"cntkbroadcast\"></a>\n",
    "#### Make the trained CNTK model available to all workers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "from cntk import load_model\n",
    "\n",
    "cntk_model_filepath = '{}/retrained.model'.format(local_tmp_dir)\n",
    "cntk_model_filepath_bc = sc.broadcast(cntk_model_filepath)\n",
    "sc.addFile(cntk_model_filepath)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"cntkworker\"></a>\n",
    "#### Define functions to be run by worker nodes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "def cntk_run_worker(files):\n",
    "    ''' Scoring script run by each worker '''\n",
    "    loaded_model = load_model(SparkFiles.get(cntk_model_filepath_bc.value))\n",
    "    \n",
    "    # Iterate through the files. The first value in each tuple is the file name; the second is the image data\n",
    "    for file in files:\n",
    "        # Load the image from its byte array, with proper dimensions and color channel order\n",
    "        image_data = np.array(Image.open(BytesIO(file[1])), dtype=np.float32)\n",
    "        image_data = np.ascontiguousarray(np.transpose(image_data[:, :, ::-1], (2,0,1)))\n",
    "        \n",
    "        # Apply the model to the image and return the true and predicted labels\n",
    "        dnn_output = loaded_model.eval({loaded_model.arguments[0]: [image_data]})\n",
    "        true_label = get_nlcd_id(file[0])\n",
    "        yield (file[0], true_label, np.argmax(np.squeeze(dnn_output)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"cntkscore\"></a>\n",
    "#### Score all test set images with the trained model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Scored 11760 images\n",
      "Time elapsed: 0:07:49.978144"
     ]
    }
   ],
   "source": [
    "labeled_images = image_rdd.mapPartitions(cntk_run_worker)\n",
    "\n",
    "start = pd.datetime.now()\n",
    "cntk_results = labeled_images.collect()\n",
    "print('Scored {} images'.format(len(cntk_results)))\n",
    "stop = pd.datetime.now()\n",
    "print('Time elapsed: {}'.format(stop - start))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Note that this step may take up to minutes to complete.\n",
    "\n",
    "<a name=\"cntkevaluate\"></a>\n",
    "#### Evaluate the model's performance\n",
    "\n",
    "We first report the model's raw overall accuracy. We then calculate the overall accuracy when all undeveloped land types are grouped under the same label. (This is done to illustrate that the majority of errors confuse different types of undeveloped land.)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "When using all six categories, correctly predicted 9528 of 11760 images (81.02%)\n",
      "After regrouping land use categories, correctly predicted 10861 of 11760 images (92.36%)"
     ]
    }
   ],
   "source": [
    "def group_undeveloped_land_types(original_label):\n",
    "    if original_label in [3, 5]:  # developed and cultivated land types\n",
    "        return(original_label)\n",
    "    else:\n",
    "        return(6)  # new grouped label for all undeveloped land types\n",
    "\n",
    "cntk_df = pd.DataFrame(cntk_results, columns=['filename', 'true_label', 'predicted_label'])\n",
    "num_correct = sum(cntk_df['true_label'] == cntk_df['predicted_label'])\n",
    "num_total = len(cntk_results)\n",
    "print('When using all six categories, correctly predicted ' +\n",
    "      '{} of {} images ({:0.2f}%)'.format(num_correct,\n",
    "                                          num_total,\n",
    "                                          100 * num_correct / num_total))\n",
    "\n",
    "cntk_df['true_label_regrouped'] = cntk_df['true_label'].apply(group_undeveloped_land_types)\n",
    "cntk_df['predicted_label_regrouped'] = cntk_df['predicted_label'].apply(group_undeveloped_land_types)\n",
    "num_correct = sum(cntk_df['true_label_regrouped'] == cntk_df['predicted_label_regrouped'])\n",
    "print('After regrouping land use categories, correctly predicted ' +\n",
    "      '{} of {} images ({:0.2f}%)'.format(num_correct,\n",
    "                                          num_total,\n",
    "                                          100 * num_correct / num_total))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"tf\"></a>\n",
    "### Score and evaluate with a trained TensorFlow model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"tfmodel\"></a>\n",
    "#### Make the trained TensorFlow model available to all workers\n",
    "\n",
    "Loads a slightly modified version of the tf-slim ResNet definition from the [TensorFlow models git repository](https://github.com/tensorflow/models)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "sc.addPyFile(os.path.join(local_tmp_dir, 'resnet_utils.py'))\n",
    "sc.addPyFile(os.path.join(local_tmp_dir, 'resnet_v1.py'))\n",
    "model_dir_bc = sc.broadcast(local_tmp_dir)\n",
    "\n",
    "import tensorflow as tf\n",
    "import functools\n",
    "import resnet_v1\n",
    "slim = tf.contrib.slim"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"tfworker\"></a>\n",
    "#### Define functions used by workers for scoring"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "def get_network_fn(num_classes, weight_decay=0.0, is_training=False):\n",
    "    arg_scope = resnet_v1.resnet_arg_scope(weight_decay=weight_decay)\n",
    "    func = resnet_v1.resnet_v1_50\n",
    "    @functools.wraps(func)\n",
    "    def network_fn(images):\n",
    "        with slim.arg_scope(arg_scope):\n",
    "            return func(images, num_classes, is_training=is_training)\n",
    "    if hasattr(func, 'default_image_size'):\n",
    "        network_fn.default_image_size = func.default_image_size\n",
    "    return(network_fn)\n",
    "\n",
    "def mean_image_subtraction(image, means):\n",
    "    num_channels = image.get_shape().as_list()[-1]\n",
    "    channels = tf.split(image, num_channels, 2)\n",
    "    for i in range(num_channels):\n",
    "        channels[i] -= means[i]\n",
    "    return(tf.concat(channels, 2))\n",
    "\n",
    "def get_preprocessing():\n",
    "    def preprocessing_fn(image, output_height=224, output_width=224):\n",
    "        image = tf.expand_dims(image, 0)\n",
    "        resized_image = tf.image.resize_bilinear(image, [output_height, output_width], align_corners=False)\n",
    "        resized_image = tf.squeeze(resized_image)\n",
    "        resized_image.set_shape([output_height, output_width, 3])\n",
    "        image = tf.to_float(resized_image)\n",
    "        return(mean_image_subtraction(image, [123.68, 116.78, 103.94]))\n",
    "    return(preprocessing_fn)\n",
    "\n",
    "def tf_run_worker(files):\n",
    "    model_dir = model_dir_bc.value\n",
    "    results = []\n",
    "    \n",
    "    with tf.Graph().as_default():\n",
    "        network_fn = get_network_fn(num_classes=6, is_training=False)\n",
    "        image_preprocessing_fn = get_preprocessing()\n",
    "        \n",
    "        current_image = tf.placeholder(tf.uint8, shape=(224, 224, 3))\n",
    "        preprocessed_image = image_preprocessing_fn(current_image, 224, 224)\n",
    "        image  = tf.expand_dims(preprocessed_image, 0)\n",
    "        logits, _ = network_fn(image)\n",
    "        predictions = tf.argmax(logits, 1)\n",
    "        \n",
    "        with tf.Session() as sess:\n",
    "            my_saver = tf.train.Saver()\n",
    "            my_saver.restore(sess, tf.train.latest_checkpoint(model_dir))\n",
    "            \n",
    "            coord = tf.train.Coordinator()\n",
    "            threads = tf.train.start_queue_runners(sess=sess, coord=coord)\n",
    "            try:\n",
    "                for file in files:\n",
    "                    imported_image_np = np.asarray(Image.open(BytesIO(file[1])), dtype=np.uint8)\n",
    "                    result = sess.run(predictions, feed_dict={current_image: imported_image_np})\n",
    "                    true_label = get_nlcd_id(file[0])\n",
    "                    results.append([file[0], true_label, result[0]])\n",
    "            finally:\n",
    "                coord.request_stop()\n",
    "            coord.join(threads)\n",
    "    return(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"tfscore\"></a>\n",
    "#### Score all images with trained TensorFlow model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Scored 11760 images\n",
      "0:09:13.936763"
     ]
    }
   ],
   "source": [
    "labeled_images_tf = image_rdd.mapPartitions(tf_run_worker)\n",
    "\n",
    "start = pd.datetime.now()\n",
    "results_tf = labeled_images_tf.collect()\n",
    "print('Scored {} images'.format(len(results_tf)))\n",
    "stop = pd.datetime.now()\n",
    "print(stop - start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Note that this step may take up to 10 minutes to complete.\n",
    "\n",
    "<a name=\"tfevaluate\"></a>\n",
    "#### Evaluate the model's performance\n",
    "\n",
    "We first report the model's raw overall accuracy. We then calculate the overall accuracy when all undeveloped land types are grouped under the same label. (This is done to illustrate that the majority of errors confuse different types of undeveloped land.)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "When using all six categories, correctly predicted 8844 of 11760 images (75.20%)\n",
      "After regrouping land use categories, correctly predicted 10931 of 11760 images (92.95%)"
     ]
    }
   ],
   "source": [
    "def group_undeveloped_land_types(original_label):\n",
    "    if original_label in [3, 5]:  # developed and cultivated land types\n",
    "        return(original_label)\n",
    "    else:\n",
    "        return(6)\n",
    "\n",
    "tf_df = pd.DataFrame(results_tf, columns=['filename', 'true_label', 'predicted_label'])\n",
    "num_correct = sum(tf_df['true_label'] == tf_df['predicted_label'])\n",
    "num_total = len(results_tf)\n",
    "print('When using all six categories, correctly predicted {} of {} images ({:0.2f}%)'.format(num_correct,\n",
    "                                                                                             num_total,\n",
    "                                                                                             100 * num_correct / num_total))\n",
    "\n",
    "tf_df['true_label_regrouped'] = tf_df['true_label'].apply(group_undeveloped_land_types)\n",
    "tf_df['predicted_label_regrouped'] = tf_df['predicted_label'].apply(group_undeveloped_land_types)\n",
    "num_correct = sum(tf_df['true_label_regrouped'] == tf_df['predicted_label_regrouped'])\n",
    "print('After regrouping land use categories, correctly predicted {} of {} images ({:0.2f}%)'.format(num_correct,\n",
    "                                                                                                    num_total,\n",
    "                                                                                                    100 * num_correct / num_total))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<a name=\"next\"></a>\n",
    "## Next Steps\n",
    "\n",
    "For an example of how the trained model can be applied to identify newly developed regions and explore county-level patterns in development, please see the next document in this repository: [Land Use Prediction in Middlesex County, MA](land_use_prediction.md)."
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "PySpark3",
   "language": "",
   "name": "pyspark3kernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 3
   },
   "mimetype": "text/x-python",
   "name": "pyspark3",
   "pygments_lexer": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}