demo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Zappy Demo"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook demonstrates Zappy, a library for distributed processing of chunked NumPy arrays. We'll start with the `direct` engine, which is the simplest engine. Processing is carried out locally, in-memory.\n",
    "\n",
    "First, create a Zappy array filled with integer 1s."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import zappy.direct\n",
    "a = zappy.direct.ones((10, 2), chunks=(2, 2), dtype='i4')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can look at the contents of the array by calling its `asndarray` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([[1, 1],\n",
       "       [1, 1],\n",
       "       [1, 1],\n",
       "       [1, 1],\n",
       "       [1, 1],\n",
       "       [1, 1],\n",
       "       [1, 1],\n",
       "       [1, 1],\n",
       "       [1, 1],\n",
       "       [1, 1]], dtype=int32)"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "a.asndarray()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that the chunking is hidden, since it's an implementation detail.\n",
    "\n",
    "Now we have a Zappy array, we can run a computation on it. Zappy exposes the standard NumPy API, accessible via the `zappy.base` module. Here we simply sum all columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([10, 10])"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import zappy.base as np\n",
    "b = np.sum(a, axis=0)\n",
    "b.asndarray()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can save an array in Zarr format. If the array does not match the chunk size supplied -- either because it is different to the original chunk size, or the number of rows in some partitions has changed in the course of applying NumPy operations -- then the rows are repartitioned before being written out."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.to_zarr(\"/tmp/a.zarr\", chunks=(2, 2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/tmp/a.zarr\r\n",
      "/tmp/a.zarr/.zarray\r\n",
      "/tmp/a.zarr/1.0\r\n",
      "/tmp/a.zarr/2.0\r\n",
      "/tmp/a.zarr/3.0\r\n",
      "/tmp/a.zarr/0.0\r\n",
      "/tmp/a.zarr/4.0\r\n"
     ]
    }
   ],
   "source": [
    "! find /tmp/a.zarr"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It's easy to read Zappy arrays from Zarr. In this case there's no need to specify a chunk size since the array automatically inherits Zarr's chunk size."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([10, 10])"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "c = zappy.direct.from_zarr(\"/tmp/a.zarr\")\n",
    "np.sum(c, axis=0).asndarray()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Serverless Processing with Pywren\n",
    "\n",
    "[Pywren](http://pywren.io/) allows you to run Python code at scale using AWS Lambda. Zappy has a Pywren engine called `executor` that makes it easy to perform NumPy calculations at scale.\n",
    "\n",
    "Before starting, [install Pywren](http://pywren.io/pages/gettingstarted.html) in the same Python virtual environment you created to run this demo.\n",
    "\n",
    "Zappy uses a custom Pywren runtime for Zarr support, which is enabled by editing the `runtime` section in *~/.pywren_config* to be:\n",
    "\n",
    "```\n",
    "runtime:\n",
    "    s3_bucket: tom-pywren-runtimes\n",
    "    s3_key: pywren.runtime/pywren_runtime-3.6-default.meta.json\n",
    "```\n",
    "\n",
    "Now that Pywren is set up, we can create a Zappy array of 1s, just like for the `direct` engine above. This time we need to pass in a Python [Executor](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Executor) object, so we create one using Pywren as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pywren\n",
    "import zappy.executor\n",
    "executor = zappy.executor.PywrenExecutor()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the `executor` we can create the Zappy array..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "a = zappy.executor.ones(executor, (10, 2), chunks=(2, 2), dtype='i4')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "... and save it as a Zarr array in S3 cloud storage. (You will need to change the S3 path to be a bucket that you have write access to.)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "import s3fs.mapping\n",
    "s3 = s3fs.S3FileSystem()\n",
    "path = 'sc-tom-test-data/ones.zarr'\n",
    "output_zarr = s3fs.mapping.S3Map(path, s3=s3)\n",
    "a.to_zarr(output_zarr, a.chunks)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This may take tens of seconds to a minute to run. What's happening behind the scenes is that Pywren serializes the execution graph and input data values, then runs each input as an AWS Lambda invocation. The execution graph in this case is the `ones` function that creates an array chunk of 1s, followed by a write operation that writes the array chunk to Zarr in S3. The data values are five `(2, 2)` chunk sizes that are inputs to the `ones` function.\n",
    "\n",
    "The important point is that the processing all happened in the cloud, not in the local Python process.\n",
    "\n",
    "The result is that there is a Zarr array stored on S3 as five `(2, 2)` chunks. We can see the raw files (_0.0_, _1.0_, ... _4.0_) by listing the S3 bucket:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['sc-tom-test-data/ones.zarr/.zarray',\n",
       " 'sc-tom-test-data/ones.zarr/0.0',\n",
       " 'sc-tom-test-data/ones.zarr/1.0',\n",
       " 'sc-tom-test-data/ones.zarr/2.0',\n",
       " 'sc-tom-test-data/ones.zarr/3.0',\n",
       " 'sc-tom-test-data/ones.zarr/4.0']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "s3.ls(path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, delete the data from S3:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3.rm(path, recursive=True)"
   ]
  },
  {
   "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
}