armada-pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Armada Prediction Serving Pipeline\n",
    "\n",
    "This notebook defines an example evaluation pipeline for prediction serving. \n",
    "\n",
    "The DAG is composed of follows. We will be starting with light-weight preprocess models, and then it will broadcast the preprocessed image to $k$ copies of pytorch squeezenet models. Finally, the numpy average models will take the result from all squeezenet and perform an average. \n",
    "\n",
    "This is intended to model use case where there exists $k$ copies of squeezenet models trained on different day; we use ensemble to make sure our final prediction is stable and accurate by combining the output from all $k$ copies. \n",
    "\n",
    "This pipeline will likely illustrate the shared object storage capability of the system as well as parallel execution capability"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/svg+xml": [
       "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?>\n",
       "<!DOCTYPE svg PUBLIC \"-//W3C//DTD SVG 1.1//EN\"\n",
       " \"http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd\">\n",
       "<!-- Generated by graphviz version 2.40.1 (20161225.0304)\n",
       " -->\n",
       "<!-- Title: %3 Pages: 1 -->\n",
       "<svg width=\"566pt\" height=\"188pt\"\n",
       " viewBox=\"0.00 0.00 565.77 188.00\" xmlns=\"http://www.w3.org/2000/svg\" xmlns:xlink=\"http://www.w3.org/1999/xlink\">\n",
       "<g id=\"graph0\" class=\"graph\" transform=\"scale(1 1) rotate(0) translate(4 184)\">\n",
       "<title>%3</title>\n",
       "<polygon fill=\"#ffffff\" stroke=\"transparent\" points=\"-4,4 -4,-184 561.7695,-184 561.7695,4 -4,4\"/>\n",
       "<!-- preprocess -->\n",
       "<g id=\"node1\" class=\"node\">\n",
       "<title>preprocess</title>\n",
       "<ellipse fill=\"none\" stroke=\"#000000\" cx=\"278.8848\" cy=\"-162\" rx=\"130.5353\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"278.8848\" y=\"-157.8\" font-family=\"Times,serif\" font-size=\"14.00\" fill=\"#000000\">Scikit&#45;Image Preprocess (&lt; 15ms)</text>\n",
       "</g>\n",
       "<!-- squeezenet_1 -->\n",
       "<g id=\"node2\" class=\"node\">\n",
       "<title>squeezenet_1</title>\n",
       "<ellipse fill=\"none\" stroke=\"#000000\" cx=\"116.8848\" cy=\"-90\" rx=\"116.7696\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"116.8848\" y=\"-85.8\" font-family=\"Times,serif\" font-size=\"14.00\" fill=\"#000000\">PyTorch Squeezenet (&lt; 50ms)</text>\n",
       "</g>\n",
       "<!-- preprocess&#45;&gt;squeezenet_1 -->\n",
       "<g id=\"edge1\" class=\"edge\">\n",
       "<title>preprocess&#45;&gt;squeezenet_1</title>\n",
       "<path fill=\"none\" stroke=\"#000000\" d=\"M240.0817,-144.7542C217.4393,-134.6909 188.6936,-121.915 164.6653,-111.2358\"/>\n",
       "<polygon fill=\"#000000\" stroke=\"#000000\" points=\"165.9308,-107.9682 155.3712,-107.1051 163.0878,-114.3649 165.9308,-107.9682\"/>\n",
       "</g>\n",
       "<!-- squeezenet... -->\n",
       "<g id=\"node3\" class=\"node\">\n",
       "<title>squeezenet...</title>\n",
       "<ellipse fill=\"none\" stroke=\"#000000\" cx=\"278.8848\" cy=\"-90\" rx=\"27\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"278.8848\" y=\"-85.8\" font-family=\"Times,serif\" font-size=\"14.00\" fill=\"#000000\">...</text>\n",
       "</g>\n",
       "<!-- preprocess&#45;&gt;squeezenet... -->\n",
       "<g id=\"edge2\" class=\"edge\">\n",
       "<title>preprocess&#45;&gt;squeezenet...</title>\n",
       "<path fill=\"none\" stroke=\"#000000\" d=\"M278.8848,-143.8314C278.8848,-136.131 278.8848,-126.9743 278.8848,-118.4166\"/>\n",
       "<polygon fill=\"#000000\" stroke=\"#000000\" points=\"282.3849,-118.4132 278.8848,-108.4133 275.3849,-118.4133 282.3849,-118.4132\"/>\n",
       "</g>\n",
       "<!-- squeezenet_k -->\n",
       "<g id=\"node4\" class=\"node\">\n",
       "<title>squeezenet_k</title>\n",
       "<ellipse fill=\"none\" stroke=\"#000000\" cx=\"440.8848\" cy=\"-90\" rx=\"116.7696\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"440.8848\" y=\"-85.8\" font-family=\"Times,serif\" font-size=\"14.00\" fill=\"#000000\">PyTorch Squeezenet (&lt; 50ms)</text>\n",
       "</g>\n",
       "<!-- preprocess&#45;&gt;squeezenet_k -->\n",
       "<g id=\"edge3\" class=\"edge\">\n",
       "<title>preprocess&#45;&gt;squeezenet_k</title>\n",
       "<path fill=\"none\" stroke=\"#000000\" d=\"M317.6878,-144.7542C340.3302,-134.6909 369.076,-121.915 393.1043,-111.2358\"/>\n",
       "<polygon fill=\"#000000\" stroke=\"#000000\" points=\"394.6817,-114.3649 402.3984,-107.1051 391.8387,-107.9682 394.6817,-114.3649\"/>\n",
       "</g>\n",
       "<!-- average -->\n",
       "<g id=\"node5\" class=\"node\">\n",
       "<title>average</title>\n",
       "<ellipse fill=\"none\" stroke=\"#000000\" cx=\"278.8848\" cy=\"-18\" rx=\"97.908\" ry=\"18\"/>\n",
       "<text text-anchor=\"middle\" x=\"278.8848\" y=\"-13.8\" font-family=\"Times,serif\" font-size=\"14.00\" fill=\"#000000\">Numpy Average (&lt; 1ms)</text>\n",
       "</g>\n",
       "<!-- squeezenet_1&#45;&gt;average -->\n",
       "<g id=\"edge4\" class=\"edge\">\n",
       "<title>squeezenet_1&#45;&gt;average</title>\n",
       "<path fill=\"none\" stroke=\"#000000\" d=\"M155.2765,-72.937C178.2396,-62.7312 207.5894,-49.6868 231.9271,-38.8701\"/>\n",
       "<polygon fill=\"#000000\" stroke=\"#000000\" points=\"233.6108,-41.9519 241.3275,-34.6921 230.7678,-35.5552 233.6108,-41.9519\"/>\n",
       "</g>\n",
       "<!-- squeezenet...&#45;&gt;average -->\n",
       "<g id=\"edge5\" class=\"edge\">\n",
       "<title>squeezenet...&#45;&gt;average</title>\n",
       "<path fill=\"none\" stroke=\"#000000\" d=\"M278.8848,-71.8314C278.8848,-64.131 278.8848,-54.9743 278.8848,-46.4166\"/>\n",
       "<polygon fill=\"#000000\" stroke=\"#000000\" points=\"282.3849,-46.4132 278.8848,-36.4133 275.3849,-46.4133 282.3849,-46.4132\"/>\n",
       "</g>\n",
       "<!-- squeezenet_k&#45;&gt;average -->\n",
       "<g id=\"edge6\" class=\"edge\">\n",
       "<title>squeezenet_k&#45;&gt;average</title>\n",
       "<path fill=\"none\" stroke=\"#000000\" d=\"M402.493,-72.937C379.53,-62.7312 350.1801,-49.6868 325.8424,-38.8701\"/>\n",
       "<polygon fill=\"#000000\" stroke=\"#000000\" points=\"327.0017,-35.5552 316.4421,-34.6921 324.1587,-41.9519 327.0017,-35.5552\"/>\n",
       "</g>\n",
       "</g>\n",
       "</svg>\n"
      ],
      "text/plain": [
       "<graphviz.dot.Digraph at 0x1112da390>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import graphviz\n",
    "g = graphviz.Digraph()\n",
    "g.node(\"preprocess\", \"Scikit-Image Preprocess (< 15ms)\")\n",
    "g.node(\"squeezenet_1\", \"PyTorch Squeezenet (< 50ms)\")\n",
    "g.node(\"squeezenet...\", \"...\")\n",
    "g.node(\"squeezenet_k\", \"PyTorch Squeezenet (< 50ms)\")\n",
    "g.node(\"average\", \"Numpy Average (< 1ms)\")\n",
    "squeeze = [f\"squeezenet{suffix}\" for suffix in [\"_1\", \"...\", \"_k\"]]\n",
    "[g.edge(\"preprocess\", s) for s in squeeze]\n",
    "[g.edge(s, \"average\") for s in squeeze]\n",
    "g"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Each model can be batched\n",
    "batch_size = 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model 1: Preprocess\n",
    "\n",
    "We will do a simple guassian filters and reshape it so pytorch can process it"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from skimage import filters\n",
    "import numpy as np\n",
    "arr = np.random.randn(batch_size, 224, 224, 3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocess(inp):\n",
    "    return filters.gaussian(inp).reshape(batch_size, 3, 224, 224)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "12.4 ms ± 2.36 ms per loop (mean ± std. dev. of 7 runs, 100 loops each)\n"
     ]
    }
   ],
   "source": [
    "%%timeit\n",
    "preprocess(arr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model 2: SqueezeNet\n",
    "\n",
    "- CPU Latency: 45ms"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "import torchvision"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/Users/simonmo/anaconda3/lib/python3.6/site-packages/torchvision-0.1.9-py3.6.egg/torchvision/models/squeezenet.py:94: UserWarning: nn.init.kaiming_uniform is now deprecated in favor of nn.init.kaiming_uniform_.\n",
      "/Users/simonmo/anaconda3/lib/python3.6/site-packages/torchvision-0.1.9-py3.6.egg/torchvision/models/squeezenet.py:92: UserWarning: nn.init.normal is now deprecated in favor of nn.init.normal_.\n"
     ]
    }
   ],
   "source": [
    "model = torchvision.models.squeezenet1_1()\n",
    "inp = np.random.randn(batch_size, 3, 224, 224)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def squeezenet(inp):\n",
    "    return model(torch.tensor(inp.astype(np.float32))).detach().numpy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "53.7 ms ± 9.17 ms per loop (mean ± std. dev. of 7 runs, 10 loops each)\n"
     ]
    }
   ],
   "source": [
    "%%timeit\n",
    "squeezenet(inp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model 3: Average"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "arrs = [np.random.randn(batch_size, 1000) for _ in range(20)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def average(inputs):\n",
    "    return np.mean(inputs, axis=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "63.9 µs ± 8.71 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)\n"
     ]
    }
   ],
   "source": [
    "%%timeit\n",
    "average(arrs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_image = np.random.randn(batch_size, 224, 224, 3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "def pipeline(inp, k_replica=5):\n",
    "    # stage one\n",
    "    preprocess_output = preprocess(inp)\n",
    "    \n",
    "    # stage two \n",
    "    # NOTE: this can be parallelized\n",
    "    squeezenets_output = [squeezenet(preprocess_output) for _ in range(k_replica)]\n",
    "    \n",
    "    # stage three\n",
    "    final_result = average(squeezenets_output)\n",
    "    \n",
    "    return final_result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "400 ms ± 77.9 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n"
     ]
    }
   ],
   "source": [
    "%%timeit\n",
    "pipeline(input_image)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "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.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}