004-h5py-and-arrays.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data management on the cloud and dask arrays\n",
    "\n",
    "The way this works: data is stored in one place (our GCS bucket) and we pull it down from there and push it back there. \n",
    "\n",
    "As much as possible, avoid creating data locally on the machine that is running your notebook. This will help us: \n",
    "\n",
    "1. Collaborate more readily -- everyone's code points to one place.\n",
    "2. Work more rapidly -- worker nodes on the cluster get their own data.\n",
    "3. Avoid losing data -- GCS is solid, but nodes are kinda ephemeral.\n",
    "\n",
    "In the following code, we're going to read and write data into GCS using `h5py`, which is a library for reading and writing hdf5 files. The `gcsfs` library lets us interact with GCS as though it was the file-system of the machine that we are working  on (and later on as though it was mounted on all nodes of our cluster)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import h5py\n",
    "import gcsfs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create a file system object based on our project name on GCP:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs = gcsfs.GCSFileSystem(project='learning-2-learn-221016')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can do different file-sytem operations with this object. For example, we can get a listing of the objects inside of our bucket. Our project is the top-level or root of the file-system:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs.ls('/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And we can drill down further into the file-system:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs.ls('/learning2learn')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs.ls('/learning2learn/tutorial')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For the tutorial's sake, we have an example of some data stored in our bucket in an hdf5 file. \n",
    "\n",
    "As long as the file is open, we can operate on the data within it. For this purpose, we will create a context manager within which the file is open:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "with fs.open('learning2learn/tutorial/random.hdf5', 'rb') as f:\n",
    "    f = h5py.File(f,'r')\n",
    "    dset = f['/x']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once outside of the context manager, we can do certain operations with the data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(dset)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But not others:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dset[1:10]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For data this size, it's reasonable to operate in blocks:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "10000000 == 10e6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "sums = []\n",
    "lengths = []\n",
    "\n",
    "with fs.open('learning2learn/tutorial/random.hdf5', 'rb') as f:\n",
    "    f = h5py.File(f,'r')\n",
    "    dset = f['/x']\n",
    "    # Compute sum of large, O(10e8)-element array, 10e6 numbers at a time\n",
    "    for i in range(0, int(10e8), int(10e6)):\n",
    "        chunk = dset[i: i + int(10e6)]  # pull out numpy array\n",
    "        sums.append(chunk.sum())\n",
    "        lengths.append(len(chunk))\n",
    "\n",
    "mean = sum(sums) / sum(lengths)\n",
    "print(mean)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But that's pretty slow... \n",
    "\n",
    "You can probably already see for yourself how we might use the delayed interface to write our own parallelized version of this. In addition to the tools you've already seen, dask provides a specialized API for representation and computation with arrays, which we will introduce here.\n",
    "\n",
    "Let's start by firing up a cluster:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client, progress\n",
    "from dask_kubernetes import KubeCluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster = KubeCluster(n_workers=20)\n",
    "cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll need the dask `delayed` function and we'll need the dask `array` module"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask import delayed\n",
    "import dask.array as da"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Because our workers are distributed, we need to take a radically different approach here. Each node in our cluster will have to open the file on its own  and access its own part of the data. We do that by first writing a function that knows how to read the data and extract the needed part of it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def read_chunk(i):\n",
    "    with fs.open('learning2learn/tutorial/random.hdf5', 'rb') as f:\n",
    "        f = h5py.File(f, 'r')\n",
    "        dset = f['/x'] \n",
    "        chunk = dset[i: i + int(10e6)]\n",
    "    return chunk"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then, we create delayed arrays. Each element in this list is just the instructions for creating an array from a delayed function. The arrays have not been materialized yet. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "chunks = []\n",
    "\n",
    "for i in range(0, int(10e8), int(10e6)):\n",
    "    this_chunk = delayed(read_chunk)(i)\n",
    "    chunks.append(da.from_delayed(this_chunk, \n",
    "                                  shape=(int(10e6), ), \n",
    "                                  dtype=chunk.dtype))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We concatenate all of these arrays together to form one large array:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "x = da.concatenate(chunks)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Question: to calculate the mean, do you need to ever have all of the data in one node's memory? Why is that?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "m = x.mean()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "the_mean = m.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our setup allows us to read data directly from our bucket, without needing to further authenticate. To be able to upload data into our bucket, we'll need to authenticate. The method for authentication is `'browser'`, which means that you'll get a link to use for authentication and then you can \n",
    "\n",
    "    fs = gcsfs.GCSFileSystem(project='learning-2-learn-221016', token='browser')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once you've entered the token, the following code should work: \n",
    "\n",
    "    fs.put('path/to/local/file', '/learning2learn/path/to/remote/file')"
   ]
  },
  {
   "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}