github-on-ec2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We've been working on some fun stuff lately, namely dask and anaconda cluster. So we have been experimenting with analyzing all the github data for 2015 on an EC2 cluster (a distributed memory environment). We use anaconda cluster to set up a 50 node cluster on EC2, then use dask for analysis.\n",
    "\n",
    "*  [dask](http://dask.pydata.org/en/latest/) ([github](https://github.com/continuumIO/dask)) for analysis\n",
    "*  [anaconda-cluster](https://github.com/continuumIO/dask) for cluster management. \n",
    "\n",
    "**Dask** is a tool for out-of-core, parallel data analysis. Recently we added a [distributed memory scheduler](http://dask.pydata.org/en/latest/distributed.html) for running dask on clusters. We will also be using [dask.bag](http://dask.pydata.org/en/latest/bag.html), which provides an api for operations on unordered lists (like sets but with duplicates). It is great for dealing with semi-structured data like JSON blobs or log files. More blogposts about dask can be found [here](http://www.continuum.io/blog/tags/dask) or [here](http://matthewrocklin.com/blog/tags.html#dask-ref).\n",
    "\n",
    "**Anaconda Cluster** lets us easily setup clusters and manage the packages in them with conda. Running the cluster for this demo is just a few lines.\n",
    "\n",
    "```bash\n",
    "acluster create my-cluster -p aws_profile  # create the cluster\n",
    "acluster install notebook dask-cluster  # install plugins that setup an ipython notebook and dask cluster\n",
    "acluster conda install boto ujson # install conda packages we need for the demo\n",
    "acluster open notebook  # open ipython-notebook in the browser to interact with the cluster\n",
    "```\n",
    "\n",
    "While dask.distributed is well integrated with [Anaconda cluster](http://continuumio.github.io/anaconda-cluster/) it isn't restricted to it.  [This blogpost](http://matthewrocklin.com/blog/work/2015/06/23/Distributed/) shows how to set up a dask.distributed network manually and [these docs](https://dask.readthedocs.io/en/latest/distributed.html#ipython-parallel) show how to set up dask.distributed from any IPyParallel cluster.\n",
    "\n",
    "### Related Projects\n",
    "Projects for python analytics in a distributed memory environment\n",
    "* [`ipyparallel`](https://github.com/ipython/ipyparallel)\n",
    "* [`luigi`](https://github.com/spotify/luigi)\n",
    "* [`pyspark`](https://spark.apache.org/docs/0.9.0/python-programming-guide.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Github Archive data on S3\n",
    "\n",
    "We took data from [githubarchive.com](githubarchive.com), from January 2015 to May 2015, and put this on S3. We choose S3 because there are nice python libraries for interacting with it, and we can get awesome bandwidth from EC2 to S3. (The script for gathering this data is [here](https://gist.github.com/cowlicks/7973b68e34808ddf97e2)). \n",
    "\n",
    "Lets inspect the data first so we can find something to analyze and learn the data schema. You can inspect the data yourself in the `githubarchive-data` S3 bucket."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "28.289080878"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "### cut me out\n",
    "import boto\n",
    "bucket = boto.connect_s3().get_bucket('githubarchive-data')  # this is a public bucket\n",
    "\n",
    "from fnmatch import fnmatchcase\n",
    "keys = bucket.list()\n",
    "def data_size(key_str):\n",
    "    \"\"\"Check the size of all the keys in our bucket that match pattern\n",
    "    that can contain ? and * for matching.\n",
    "    \"\"\"\n",
    "    matches = [k for k in keys if fnmatchcase(k.name, key_str)]\n",
    "    return sum([k.size for k in matches]) * 1e-9  # in GB\n",
    "\n",
    "data_size('2015-*')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inspect S3 data with `dask.bag`\n",
    "\n",
    "We have approximately 28 GB of data. One file per hour, averaging around 7.8 MB each (compressed). So what is the schema and how can we inspect it? We take **one** file and turn it into a `dask.Bag` for analysis on our **local machine**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import dask.bag as db\n",
    "import ujson as json\n",
    "\n",
    "# take one file from the bucket load it as a json object, not gz decompression\n",
    "# happens automatically at compute time.\n",
    "b = db.from_s3('githubarchive-data', '2015-01-01-0.json.gz').map(json.loads)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{u'actor': {u'avatar_url': u'https://avatars.githubusercontent.com/u/9152315?',\n",
       "  u'gravatar_id': u'',\n",
       "  u'id': 9152315,\n",
       "  u'login': u'davidjhulse',\n",
       "  u'url': u'https://api.github.com/users/davidjhulse'},\n",
       " u'created_at': u'2015-01-01T00:00:00Z',\n",
       " u'id': u'2489368070',\n",
       " u'payload': {u'before': u'86ffa724b4d70fce46e760f8cc080f5ec3d7d85f',\n",
       "  u'commits': [{u'author': {u'email': u'david.hulse@live.com',\n",
       "     u'name': u'davidjhulse'},\n",
       "    u'distinct': True,\n",
       "    u'message': u'Altered BingBot.jar\\n\\nFixed issue with multiple account support',\n",
       "    u'sha': u'a9b22a6d80c1e0bb49c1cf75a3c075b642c28f81',\n",
       "    u'url': u'https://api.github.com/repos/davidjhulse/davesbingrewardsbot/commits/a9b22a6d80c1e0bb49c1cf75a3c075b642c28f81'}],\n",
       "  u'distinct_size': 1,\n",
       "  u'head': u'a9b22a6d80c1e0bb49c1cf75a3c075b642c28f81',\n",
       "  u'push_id': 536740396,\n",
       "  u'ref': u'refs/heads/master',\n",
       "  u'size': 1},\n",
       " u'public': True,\n",
       " u'repo': {u'id': 28635890,\n",
       "  u'name': u'davidjhulse/davesbingrewardsbot',\n",
       "  u'url': u'https://api.github.com/repos/davidjhulse/davesbingrewardsbot'},\n",
       " u'type': u'PushEvent'}"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first = b.take(1)[0]  # take the first json object from the file\n",
    "first"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'payload', u'created_at', u'actor', u'public', u'repo', u'type', u'id']"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first.keys()  # top level keys in this json object"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Type looks interesting. What are possible types and how often does each occur? We can inspect this with `dask.bag.frequencies`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 8 ms, sys: 76 ms, total: 84 ms\n",
      "Wall time: 1.28 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[(u'ReleaseEvent', 24),\n",
       " (u'PublicEvent', 2),\n",
       " (u'PullRequestReviewCommentEvent', 85),\n",
       " (u'ForkEvent', 213),\n",
       " (u'MemberEvent', 16),\n",
       " (u'PullRequestEvent', 315),\n",
       " (u'IssueCommentEvent', 650),\n",
       " (u'PushEvent', 4280),\n",
       " (u'DeleteEvent', 141),\n",
       " (u'CommitCommentEvent', 56),\n",
       " (u'WatchEvent', 642),\n",
       " (u'IssuesEvent', 373),\n",
       " (u'CreateEvent', 815),\n",
       " (u'GollumEvent', 90)]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%time b.pluck('type').frequencies().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Top Committers\n",
    "----------------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So most events are pushes, that is not surprising. Lets ask \"Who pushes the most?\".\n",
    "\n",
    "We do this by filtering out `PushEvent`s. Then we count the frequencies of usernames for the pushes. Then take the top 5."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 12 ms, sys: 64 ms, total: 76 ms\n",
      "Wall time: 1.26 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[(u'KenanSulayman', 79),\n",
       " (u'mirror-updates', 42),\n",
       " (u'cm-gerrit', 35),\n",
       " (u'qdm', 29),\n",
       " (u'greatfire', 24)]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pushes = b.filter(lambda x: x['type'] == 'PushEvent')  # filter out the push events\n",
    "names = pushes.pluck('actor').pluck('login') # get the login names\n",
    "top_5 = names.frequencies().topk(5, key=lambda (name, count): count)  # List top 5 pushers\n",
    "%time top_5.compute()  # run the above computations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These users *pushed* the most, but push can have multiple commits. So we can ask \"who pushed the most *commits*?\".\n",
    "\n",
    "We can figure this out by grouping by username, then summing the number of commits from every push, for each user. More technically speaking, we want to `GroupBy` on usernames, so for each username we get a list their of PushEvents. Then reduce each `PushEvent` by taking a `count` of their commits. Then reducing these `count`s by `sum`ing them for each user. So we are grouping then reducing.\n",
    "\n",
    "However there are algorithms for grouping and reducing simultaneously which avoid expensive shuffle operations and are much faster. In dask bag we have `foldby`. Analogous methods: [`toolz.reduceby`]( https://toolz.readthedocs.io/en/latest/api.html#toolz.itertoolz.reduceby), and in pyspark [`RDD.combineByKey`](https://spark.apache.org/docs/latest/api/python/pyspark.html?#pyspark.RDD.combineByKey)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 12 ms, sys: 64 ms, total: 76 ms\n",
      "Wall time: 1.28 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[(u'mirror-updates', 413),\n",
       " (u'jrmarino', 88),\n",
       " (u'javra', 80),\n",
       " (u'KenanSulayman', 79),\n",
       " (u'chcholman', 51)]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def get_logins(x):\n",
    "    \"\"\"The key for foldby, like a groupby key. Get the username from a PushEvent\"\"\"\n",
    "    return x['actor']['login']\n",
    "\n",
    "def binop(total, x):\n",
    "    \"\"\"Count the number of commits in a PushEvent\"\"\"\n",
    "    return total + len(x['payload']['commits'])\n",
    "\n",
    "def combine(total1, total2):\n",
    "    \"\"\"This combines commit counts from PushEvents\"\"\"\n",
    "    return total1 + total2\n",
    "\n",
    "commits = pushes.foldby(get_logins, binop, initial=0, combine=combine)\n",
    "top_commits = commits.topk(5, key=lambda (name, count): count)\n",
    "%time top_commits.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Recall this `dask.Bag` had *one* file. Now that we know how to get the top committers, we'll gradually load more data, and benchmark the `dask.distributed` scheduler against the default `dask.multiprocessing` scheduler.\n",
    "\n",
    "### Benchmarking dask.distributed\n",
    "First we setup the distributed scheduler. Then write a benchmarking script, the benchmarking script is omitted but those interested can find `both_benchmark` [here](https://gist.github.com/cowlicks/5e9c7ceceed0e490712d). Basically it does time the analysis and prints the results nicely."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# dask.distributed setup\n",
    "import dask\n",
    "from dask.distributed import Client\n",
    "\n",
    "dc = Client('tcp://localhost:9000') # client connected to 50 nodes, 2 workers per node.\n",
    "# pass dc.get to compute functions to use the distributed scheduler.\n",
    "\n",
    "## cut out the rest of this cell\n",
    "# make a top5 committers function\n",
    "import time\n",
    "from pprint import pprint\n",
    "def both_benchmark(data_pattern):\n",
    "    bag = db.from_s3('githubarchive-data', data_pattern).map(json.loads)\n",
    "    pushes = bag.filter(lambda x: x['type'] == 'PushEvent')\n",
    "    commits =  pushes.foldby(get_logins, binop, initial=0, combine=combine)\n",
    "    top5 = commits.topk(5, lambda x: x[1])\n",
    "    \n",
    "    # time the default comptue and the distributed compute\n",
    "    default_start = time.time()\n",
    "    default_result = top5.compute()\n",
    "    default_time = time.time() - default_start\n",
    "    \n",
    "    dist_start = time.time()\n",
    "    dist_result = top5.compute(get=dc.get)\n",
    "    dist_time = time.time() - dist_start\n",
    "    \n",
    "    # assert we have the same result\n",
    "    assert default_result == dist_result\n",
    "    \n",
    "    # size of the computed data\n",
    "    size = data_size(data_pattern)\n",
    "    \n",
    "    # general details\n",
    "    print(\"To compute {0:.4f} GB of data the default scheduler took {1:.2f} seconds, the distributed scheduler took {2:.2f} seconds\".format(size, default_time, dist_time))\n",
    "    print(\"\")\n",
    "    # speedup default_time / dist_time\n",
    "    print(\"Distributed scheduler is \\t\\t\\t\\t\\t{:.2f} times faster.\".format(default_time / dist_time))\n",
    "    \n",
    "    # single node bandwidth = size / default_time\n",
    "    print(\"Default scheduler compute bandwidth: \\t\\t\\t\\t{:.2f} MB/s\".format(1e3 * size / default_time))\n",
    "    # dist bandwidth = size / dist_time\n",
    "    print(\"Distributed scheduler compute bandwidth: \\t\\t\\t{:.2f} MB/s\".format(1e3 * size / dist_time))\n",
    "    \n",
    "    # dist node bandwidth per node = size / (time * node)\n",
    "    print(\"Compute bandwidth per node with distributed scheduler: \\t\\t{:.3f} MB/(s node)\".format(1e3 * size / (dist_time * 50)))\n",
    "    print('')\n",
    "    print(\"Analysis results:\")\n",
    "    pprint(dist_result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets benchmark a single file first."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "To compute 0.0026 GB of data the default scheduler took 1.25 seconds, the distributed scheduler took 1.09 seconds\n",
      "\n",
      "Distributed scheduler is \t\t\t\t\t1.15 times faster.\n",
      "Default scheduler compute bandwidth: \t\t\t\t2.10 MB/s\n",
      "Distributed scheduler compute bandwidth: \t\t\t2.41 MB/s\n",
      "Compute bandwidth per node with distributed scheduler: \t\t0.048 MB/(s node)\n",
      "\n",
      "Analysis results:\n",
      "[(u'mirror-updates', 413),\n",
      " (u'jrmarino', 88),\n",
      " (u'javra', 80),\n",
      " (u'KenanSulayman', 79),\n",
      " (u'chcholman', 51)]\n"
     ]
    }
   ],
   "source": [
    "both_benchmark('2015-01-01-0.json.gz')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask distributed is comparable with the default scheduler, that is not suprising for this small amount of data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1 day of data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "To compute 0.1952 GB of data the default scheduler took 36.58 seconds, the distributed scheduler took 4.29 seconds\n",
      "\n",
      "Distributed scheduler is \t\t\t\t\t8.52 times faster.\n",
      "Default scheduler compute bandwidth: \t\t\t\t5.34 MB/s\n",
      "Distributed scheduler compute bandwidth: \t\t\t45.49 MB/s\n",
      "Compute bandwidth per node with distributed scheduler: \t\t0.910 MB/(s node)\n",
      "\n",
      "Analysis results:\n",
      "[(u'mirror-updates', 9912),\n",
      " (u'KenanSulayman', 1848),\n",
      " (u'peff', 1140),\n",
      " (u'bors', 972),\n",
      " (u'dougclarknc', 887)]\n"
     ]
    }
   ],
   "source": [
    "both_benchmark('2015-01-15-*.json.gz')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Already a good speedup."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 10 days of data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "To compute 1.5871 GB of data the default scheduler took 258.54 seconds, the distributed scheduler took 14.82 seconds\n",
      "\n",
      "Distributed scheduler is \t\t\t\t\t17.45 times faster.\n",
      "Default scheduler compute bandwidth: \t\t\t\t6.14 MB/s\n",
      "Distributed scheduler compute bandwidth: \t\t\t107.11 MB/s\n",
      "Compute bandwidth per node with distributed scheduler: \t\t2.142 MB/(s node)\n",
      "\n",
      "Analysis results:\n",
      "[(u'mirror-updates', 98297),\n",
      " (u'KenanSulayman', 18556),\n",
      " (u'qdm', 7012),\n",
      " (u'mAAdhaTTah', 6893),\n",
      " (u'greatfire', 5563)]\n"
     ]
    }
   ],
   "source": [
    "both_benchmark('2015-01-1?-*.json.gz')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Computing this on one node is possible, but it is annoying to wait so long. So we continue with just the distributed scheduler. `distributed_benchmark` can be found [here](https://gist.github.com/cowlicks/5e9c7ceceed0e490712d#file-dist_benchmark-py)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "### cut me out\n",
    "def distributed_benchmark(data_pattern):\n",
    "    bag = db.from_s3('githubarchive-data', data_pattern).map(json.loads)\n",
    "    pushes = bag.filter(lambda x: x['type'] == 'PushEvent')\n",
    "    commits =  pushes.foldby(get_logins, binop, initial=0, combine=combine)\n",
    "    top5 = commits.topk(5, lambda x: x[1])\n",
    "\n",
    "    dist_start = time.time()\n",
    "    dist_result = top5.compute(get=dc.get)\n",
    "    dist_time = time.time() - dist_start\n",
    "\n",
    "    # size of the computed data\n",
    "    size = data_size(data_pattern)\n",
    "    \n",
    "    # general details\n",
    "    print(\"To compute {0:.4f} GB of data the distributed scheduler took {1:.2f} seconds\".format(size, dist_time))\n",
    "    print('')\n",
    "   \n",
    "    # dist bandwidth = size / dist_time\n",
    "    print(\"Distributed scheduler compute bandwidth: \\t\\t\\t{:.2f} MB/s\".format(1e3 * size / dist_time))\n",
    "    \n",
    "    # dist node bandwidth per node = size / (time * node)\n",
    "    print(\"Compute bandwidth per node with distributed scheduler: \\t\\t{:.3f} MB/(s node)\".format(1e3 * size / (dist_time * 50)))\n",
    "    print('')\n",
    "    \n",
    "    print(\"Analysis results:\")\n",
    "    pprint(dist_result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### January 2015"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "To compute 5.0187 GB of data the distributed scheduler took 43.14 seconds\n",
      "\n",
      "Distributed scheduler compute bandwidth: \t\t\t116.35 MB/s\n",
      "Compute bandwidth per node with distributed scheduler: \t\t2.327 MB/(s node)\n",
      "\n",
      "Analysis results:\n",
      "[(u'mirror-updates', 302755),\n",
      " (u'greatfire', 57019),\n",
      " (u'KenanSulayman', 56557),\n",
      " (u'qdm', 20964),\n",
      " (u'greatfire-martin', 19594)]\n"
     ]
    }
   ],
   "source": [
    "distributed_benchmark('2015-01-*.json.gz')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### January - May 2015"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "To compute 28.2891 GB of data the distributed scheduler took 246.21 seconds\n",
      "\n",
      "Distributed scheduler compute bandwidth: \t\t\t114.90 MB/s\n",
      "Compute bandwidth per node with distributed scheduler: \t\t2.298 MB/(s node)\n",
      "\n",
      "Analysis results:\n",
      "[(u'mirror-updates', 1463019),\n",
      " (u'KenanSulayman', 235300),\n",
      " (u'greatfirebot', 167558),\n",
      " (u'rydnr', 133323),\n",
      " (u'markkcc', 127625)]\n"
     ]
    }
   ],
   "source": [
    "distributed_benchmark('2015-*.json.gz')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Final Thoughts\n",
    "----------------\n",
    "\n",
    "This is experimental work. We had the following problems when doing this experiment:\n",
    "\n",
    "*  Hard to deploy - solved by making a \"dask-cluster\" plugin for aconda-cluster\n",
    "*  Investigating the state of distributed network is hard - partiall solved by providing clients views to workers\n",
    "*  Profiling distributed computation is hard - in the future we'll try applying new dask [profiling methods](https://github.com/ContinuumIO/dask/pull/363) to the distributed scheduler.\n",
    "\n",
    "We also have some lingering issues regarding performance:\n",
    "\n",
    "*  Why does the distributed cluster perform worse than the single-node scheduler per node?  This computation should be embarrassingly parallel.\n",
    "*  6MB/s of compressed data throughput on a single node is nice but we can probably do better.  As always we should think first about single-core performance before we \"go big\" with a cluster."
   ]
  }
 ],
 "metadata": {
  "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}