1-08_grid_converter_dask.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Grid Coordinate Conversion with Dask cuDF"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook you will extend your understanding and ability to work with Dask cuDF by revisiting the user-defined grid conversion function. In doing so you will learn more about how Dask distributes the work of computational graphs and will continue preparing data for GPU-accelerated machine learning in the next section of the workshop."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Objectives"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "By the time you complete this notebook you will be able to:\n",
    "\n",
    "- Use Dask cuDF to map user-defined functions over Dask cuDF dataframe partitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Imports"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create a Dask cluster before importing `dask_cudf` to ensure the latter has the right CUDA context. We will import the elements necessary for creating the Dask cluster and wait to import `dask_cudf` until after the cluster has been created."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import subprocess\n",
    "\n",
    "from dask.distributed import Client, wait, progress\n",
    "from dask_cuda import LocalCUDACluster\n",
    "\n",
    "import dask.dataframe as dd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table style=\"border: 2px solid white;\">\n",
       "<tr>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Client</h3>\n",
       "<ul style=\"text-align: left; list-style: none; margin: 0; padding: 0;\">\n",
       "  <li><b>Scheduler: </b>tcp://172.18.0.4:43259</li>\n",
       "  <li><b>Dashboard: </b><a href='http://172.18.0.4:8787/status' target='_blank'>http://172.18.0.4:8787/status</a></li>\n",
       "</ul>\n",
       "</td>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Cluster</h3>\n",
       "<ul style=\"text-align: left; list-style:none; margin: 0; padding: 0;\">\n",
       "  <li><b>Workers: </b>4</li>\n",
       "  <li><b>Cores: </b>4</li>\n",
       "  <li><b>Memory: </b>257.80 GB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: 'tcp://172.18.0.4:43259' processes=4 threads=4, memory=257.80 GB>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cmd = \"hostname --all-ip-addresses\"\n",
    "process = subprocess.Popen(cmd.split(), stdout=subprocess.PIPE)\n",
    "output, error = process.communicate()\n",
    "IPADDR = str(output.decode()).split()[0]\n",
    "\n",
    "cluster = LocalCUDACluster(ip=IPADDR)\n",
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we Import CUDA context creators after setting up the cluster so they don't lock to a single device."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import cudf\n",
    "import dask_cudf\n",
    "\n",
    "import cupy as cp"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lat/Long to Grid Coordinate Conversion with Dask"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We return again to converting latitude and longitude coordinates into grid coordinates by applying our custom `latlong2osgbgrid` function, however this time we will do so in a distributed fashion with Dask. Before we can do so, we need to discuss a little more specifically about how Dask distributes the computation of its task graphs."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dask Partitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Internally, Dask dataframes are split into a number of partitions, each being an individual cuDF dataframe. Under the hood, Dask automatically breaks up the work of dataframe methods and operations among these partitions, taking care to communicate efficiently and correctly. For this reason, in using Dask earlier today to perform Dask dataframe operations, you did not have to think explicitly about how Dask had partitioned the Dask dataframes.\n",
    "\n",
    "However, when we would like to work with Dask dataframes outside their built-in methods and operators, such as when applying custom functions, we often need to work more explicitly with the partitions of the Dask dataframe, as we will do now."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dask Grid Converter"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ultimately, we are going to map our custom function to each partition of a Dask dataframe using the dataframe's `map_partitions` method.\n",
    "\n",
    "With this in mind, let's look at `latlong2osgbgrid_dask`, noting modifications we have had to make to its CuPy counterpart in order to work effectively when mapped to Dask dataframe partitions rather than run on cuDF columns. There are 4 parts to the refactor, each with accompanying comments."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 1) Rather than passing in `lat` and `long` arguments, we pass in a dataframe partition, specifying which columns hold latitude and longitude information.\n",
    "def latlong2osgbgrid_dask(part_df, lat_col='lat', long_col='long', input_degrees=True):\n",
    "    '''\n",
    "    Converts latitude and longitude (ellipsoidal) coordinates into northing and easting (grid) coordinates, using a Transverse Mercator projection.\n",
    "    \n",
    "    Inputs:\n",
    "    part_df: the dask distributed dataframe partition\n",
    "    lat_col: the name of the column holding latitude data\n",
    "    long_col: the name of the column holding longitude data\n",
    "    input_degrees: if True (default), interprets the coordinates as degrees; otherwise, interprets coordinates as radians\n",
    "    \n",
    "    Output:\n",
    "    original dataframe with northing and easting columns concatenated to the right\n",
    "    '''\n",
    "    \n",
    "    # 2) Our previous function expected `lat` and `long` values to each be CuPy array-like, so we convert the relevant columns.\n",
    "    lat = cp.asarray(part_df[lat_col])\n",
    "    long = cp.asarray(part_df[long_col])\n",
    "    \n",
    "    # 3) At this point we reuse the previous cupy code until it is time to return values.\n",
    "    if input_degrees:\n",
    "        lat = lat * cp.pi/180\n",
    "        long = long * cp.pi/180\n",
    "\n",
    "    a = 6377563.396\n",
    "    b = 6356256.909\n",
    "    e2 = (a**2 - b**2) / a**2\n",
    "\n",
    "    N0 = -100000 # northing of true origin\n",
    "    E0 = 400000 # easting of true origin\n",
    "    F0 = .9996012717 # scale factor on central meridian\n",
    "    phi0 = 49 * cp.pi / 180 # latitude of true origin\n",
    "    lambda0 = -2 * cp.pi / 180 # longitude of true origin and central meridian\n",
    "    \n",
    "    sinlat = cp.sin(lat)\n",
    "    coslat = cp.cos(lat)\n",
    "    tanlat = cp.tan(lat)\n",
    "    \n",
    "    latdiff = lat-phi0\n",
    "    longdiff = long-lambda0\n",
    "\n",
    "    n = (a-b) / (a+b)\n",
    "    nu = a * F0 * (1 - e2 * sinlat ** 2) ** -.5\n",
    "    rho = a * F0 * (1 - e2) * (1 - e2 * sinlat ** 2) ** -1.5\n",
    "    eta2 = nu / rho - 1\n",
    "    M = b * F0 * ((1 + n + 5/4 * (n**2 + n**3)) * latdiff - \n",
    "                  (3*(n+n**2) + 21/8 * n**3) * cp.sin(latdiff) * cp.cos(lat+phi0) +\n",
    "                  15/8 * (n**2 + n**3) * cp.sin(2*(latdiff)) * cp.cos(2*(lat+phi0)) - \n",
    "                  35/24 * n**3 * cp.sin(3*(latdiff)) * cp.cos(3*(lat+phi0)))\n",
    "    I = M + N0\n",
    "    II = nu/2 * sinlat * coslat\n",
    "    III = nu/24 * sinlat * coslat ** 3 * (5 - tanlat ** 2 + 9 * eta2)\n",
    "    IIIA = nu/720 * sinlat * coslat ** 5 * (61-58 * tanlat**2 + tanlat**4)\n",
    "    IV = nu * coslat\n",
    "    V = nu / 6 * coslat**3 * (nu/rho - cp.tan(lat)**2)\n",
    "    VI = nu / 120 * coslat ** 5 * (5 - 18 * tanlat**2 + tanlat**4 + 14 * eta2 - 58 * tanlat**2 * eta2)\n",
    "\n",
    "    northing = I + II * longdiff**2 + III * longdiff**4 + IIIA * longdiff**6\n",
    "    easting = E0 + IV * longdiff + V * longdiff**3 + VI * longdiff**5\n",
    "    \n",
    "    # 4) Having calculated `northing` and `easting`, we add them as series to our partition and then return the partition.\n",
    "    part_df['northing'] = cudf.Series(northing)\n",
    "    part_df['easting'] = cudf.Series(easting)\n",
    "    \n",
    "    return(part_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Mapping Functions to Partitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Dask dataframe `map_partitions` method applies a given function to each partition. As you saw in the `latlong2osgbgrid_dask` function, at least one of the arguments to the function should be a `dask.dataframe` (in our case, `part_df`).\n",
    "\n",
    "The other requirement for `map_partitions` is a *meta*: a dataframe with the structure that we will be returning from the function. You can think of this like defining a function signature, and in fact, you will find many instances in Dask programming where a meta is required.\n",
    "\n",
    "In our case, however, Dask can automatically infer the meta from our function and its inputs, so we don't need to provide one explicitly."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using the Parquet Format"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The csv format we have using thus far has been realistic to many data scientists' experiences, but alternatives are often more efficient for our needs.\n",
    "\n",
    "Here, we will output to the columnar Apache Parquet format, a natural companion to the Apache Arrow memory format of RAPIDS. Parquet also will compress our data from about 18Gb to about 12Gb. \n",
    "\n",
    "The `to_parquet` writer will create a folder of smaller parquet files with associated metadata that can efficiently be read back in later with `read_parquet`, taking advantage of parallel I/O with multiple GPU workers in Dask."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise: Build a Dask Grid Converter Pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can now build a simple data pipeline to add OSGB36 grid coordinates to the population dataset. This will consist of three steps:\n",
    "\n",
    "1. Read the csv file at `./data/pop5x_1-07.csv` into a Dask dataframe with `read_csv`\n",
    "2. Map the function `latlong2osgbgrid_dask` over that dataframe with `map_partitions`\n",
    "3. Write the results to the parquet format in the folder `pop5x` with `to_parquet`\n",
    "\n",
    "While this is running, consider bringing up the Dask status dashboard on port 8787, as in the previous notebook, and observe how Dask is asynchronously reading, transforming, and writing data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask_cudf\n",
    "ddf = dask_cudf.read_csv('./data/pop5x_1-07.csv', dtype=['float32', 'str', 'str', 'float32', 'float32', 'str'])\n",
    "result = ddf.map_partitions(latlong2osgbgrid_dask)\n",
    "result.to_parquet('./pop5x')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# %load solutions/csv_to_parquet_pipeline\n",
    "ddf = dask_cudf.read_csv('./data/pop5x_1-07.csv')\n",
    "ddf = ddf.map_partitions(latlong2osgbgrid_dask)\n",
    "ddf.to_parquet('pop5x')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise: Compute Grid Coordinate Statistics"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can analyze the results of mapping `latlong2osgbgrid_dask` the same way as any other `dask_cudf` dataframe columns. We can also see the speed enabled by parquet in the following two steps:\n",
    "\n",
    "1. Read the `pop5x` folder of parquet files into a Dask dataframe\n",
    "2. Compute the mean of the `northing` and `easting` columns\n",
    "\n",
    "Observe how quickly Dask can read in the 12Gb of parquet files through this method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>__null_dask_index__</th>\n",
       "      <th>age</th>\n",
       "      <th>sex</th>\n",
       "      <th>county</th>\n",
       "      <th>lat</th>\n",
       "      <th>long</th>\n",
       "      <th>name</th>\n",
       "      <th>northing</th>\n",
       "      <th>easting</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>m</td>\n",
       "      <td>Darlington</td>\n",
       "      <td>54.549641</td>\n",
       "      <td>-1.493884</td>\n",
       "      <td>HARRISON</td>\n",
       "      <td>517286.587386</td>\n",
       "      <td>432733.804387</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>0.0</td>\n",
       "      <td>m</td>\n",
       "      <td>Darlington</td>\n",
       "      <td>54.523945</td>\n",
       "      <td>-1.401142</td>\n",
       "      <td>LAKSH</td>\n",
       "      <td>514474.568560</td>\n",
       "      <td>438756.297571</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2</td>\n",
       "      <td>0.0</td>\n",
       "      <td>m</td>\n",
       "      <td>Darlington</td>\n",
       "      <td>54.561127</td>\n",
       "      <td>-1.690068</td>\n",
       "      <td>MUHAMMAD</td>\n",
       "      <td>518490.784309</td>\n",
       "      <td>420039.702434</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>3</td>\n",
       "      <td>0.0</td>\n",
       "      <td>m</td>\n",
       "      <td>Darlington</td>\n",
       "      <td>54.542988</td>\n",
       "      <td>-1.543216</td>\n",
       "      <td>GRAYSON</td>\n",
       "      <td>516524.191943</td>\n",
       "      <td>429548.009214</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>4</td>\n",
       "      <td>0.0</td>\n",
       "      <td>m</td>\n",
       "      <td>Darlington</td>\n",
       "      <td>54.532101</td>\n",
       "      <td>-1.569116</td>\n",
       "      <td>FINLAY</td>\n",
       "      <td>515302.693872</td>\n",
       "      <td>427880.015924</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   __null_dask_index__  age sex      county        lat      long      name  \\\n",
       "0                    0  0.0   m  Darlington  54.549641 -1.493884  HARRISON   \n",
       "1                    1  0.0   m  Darlington  54.523945 -1.401142     LAKSH   \n",
       "2                    2  0.0   m  Darlington  54.561127 -1.690068  MUHAMMAD   \n",
       "3                    3  0.0   m  Darlington  54.542988 -1.543216   GRAYSON   \n",
       "4                    4  0.0   m  Darlington  54.532101 -1.569116    FINLAY   \n",
       "\n",
       "        northing        easting  \n",
       "0  517286.587386  432733.804387  \n",
       "1  514474.568560  438756.297571  \n",
       "2  518490.784309  420039.702434  \n",
       "3  516524.191943  429548.009214  \n",
       "4  515302.693872  427880.015924  "
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ddf = dask_cudf.read_parquet('./pop5x')\n",
    "ddf.head()\n",
    "mean_northing = ddf[northing]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/read_parquet"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Next"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This concludes the first section of the workshop. You've already learned how to use cuDF and Dask_cuDF to explore and modify data, including datasets larger than a single GPU's memory, and have successfully prepped several datasets for GPU-accelerated machine learning.\n",
    "\n",
    "In the next section of the workshop, you will use the data you have prepped in the context of several GPU-accelerated machine learning algorithms, before moving onto the final section of the workshop where you will apply both your GPU-accelerated data manipulation and machine learning skills to help address an emergency scenario of national scale."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<br>\n",
    "<div align=\"center\"><h2>Optional: Restart the Kernel</h2></div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you plan to continue work in other notebooks, please clear GPU memory:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import IPython\n",
    "app = IPython.Application.instance()\n",
    "app.kernel.do_shutdown(True)"
   ]
  }
 ],
 "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.7.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}