W3-03-Pyspark-Tutorial-Joins.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark Tutorial - Joins\n",
    "<div>\n",
    " <h2> CSCI 4253 / 5253 \n",
    "  <IMG SRC=\"https://www.colorado.edu/cs/profiles/express/themes/cuspirit/logo.png\" WIDTH=50 ALIGN=\"right\"/> </h2>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "import numpy as np\n",
    "import operator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf=SparkConf().setAppName(\"pyspark tutorial\").setMaster(\"local[*]\")\n",
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Key Value Data, Grouping and Joins"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Key-Value or (k,v) datatype is fundemental to many operations including grouping and joins. We need to be able to:\n",
    "* Create keys from non-KV data\n",
    "* Group or organize data according to keys\n",
    "* Operate on data according to keys\n",
    "* Form standard joins"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating Key-Pair Values"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can directly create K-V pairs using lists of pairs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "visits = sc.parallelize([ (\"index.html\", \"1.2.3.4\"),\n",
    "                         (\"about.html\", \"3.4.5.6\"),\n",
    "                         (\"index.html\", \"1.3.3.1\") ])\n",
    "pageNames = sc.parallelize([ (\"index.html\", \"Home\"),\n",
    "                            (\"about.html\", \"About\") ])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When we `collect()` the K-V pairs, we'll get the full list of keys -- this pulls the data to the front-end machine:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "visits.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The first position of the K-V pair is the key. We'll cover grouping in more detail later, but let's take a look at grouping visits by the keys:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "visits.groupByKey().collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The KV pairs have now been *grouped* meaning that we have an RDD of all the keys and the values for each key are a \"list\" of of the values corresponding to that key in the original KV RDD. Because the values are scattered across your cluster, a `ResultIterable` is used to represented their distributed type.\n",
    "\n",
    "We reify the results by converting it to a list. You wouldn't do this in practice because this brings all the values back to the front-end machine across the whole networking, but lets see what this produces to understand what `groupByKey` is doing.\n",
    "\n",
    "We'll `map` a lambda function that simply \"flattens\" the items by converting the `ResultIterable` into a list."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "visits.groupByKey().map(lambda x: (x[0], list(x[1]))).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When we `map` across a grouped KV the mapped function takes an argument which is the pair \"(key, value\") -- that's why we're referring to `x[0]` for the value and `x[1]` for the value in the sample `map`.\n",
    "\n",
    "It's more typicaly that you want to just `map` across the values and there is a corresponding `mapValues` function that does precisely this. In the example below, we are mapping `list` across the values in the groups. The results should be the same as the "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "visits.groupByKey().mapValues(list).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`keyBy` is a function to efficiently create a key-value pair from an RDD. The elements of the original RDD are the \"values\" and a function provides the associated key.\n",
    "\n",
    "For example, assume we want $K = V^2$ for values $V$ in an RDD:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "r = sc.parallelize( [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3] )\n",
    "r.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "rsq = r.keyBy(lambda x : x*x)\n",
    "rsq.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For example, we might want to create keys for the **passwd** data using the shell name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "passwd = sc.textFile(\"/etc/passwd\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The password is the 7th field (6th index)  in the `/etc/passwd` file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "passwd.take(1)[0].split(':')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Thus, if we want the `key` to be the 6th element, we provide a function that extracts that from each entry:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "byShell = passwd.keyBy( lambda x : x.split(':')[6] )\n",
    "byShell.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is more or less equivilent to a `map` that returns pairs of values. For example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "passwd.map( lambda x : (x.split(':')[6], x) ).take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using Key-Value Pairs\n",
    "\n",
    "We can return a dictionary containing the number of logins using each shell using the `countByKey` function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "byShell.countByKey()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And we can extract keys and values in parallel across the cluster:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "shellKeys = byShell.keys()\n",
    "shellKeys.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "byShell.values().take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Grouping and Joins"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once you have (k,v) pairs, you can group the key -- the values are iterable ( distributed lists) of the values for that key. Earlier, we shows that you can `map` a function over the `ResultIterable` -- it's unlikely you want want to return them to a native `list` because that will pull all the values back to the front end machine."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "byShell.groupByKey().take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also group KV pairs by other attributes using the groupBy() method. For example, here we're going to group the password data by the user name (the 0'th field of the values)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "byShell.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "def getLogin(x):\n",
    "    return x.split(':')[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "byShell.groupBy (lambda x : getLogin(x[1]) ).take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once items are grouped, you can iterate over the values associated with a key.\n",
    "\n",
    "Let's group the `/etc/passwd` entries by the shell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "grpdShell = byShell.groupByKey()\n",
    "grpdShell.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now the key are differ Unix shells (*e.g.* `/bin/bash`) and the values are `ResultIterable`s of the values having that key in the original KV list."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Again, we typically wouldn't want to pull in all the value using a `list` because this will bring everything to the front-end machine in the cluster. But, let's convert the `ResultIterable` to a list just to see the structure:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "grpdShell.map(lambda x: (x[0], list(x[1])) ).take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The better way to process this would be to would `map` a function over each KV pair in the grouped list. The key is `x[0]` and the `ResultIterable` is `x[1]`. We can then map a function over each of the  `ResultIterable` to extract some field, such as the login information."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "grpdShell.mapValues(list).take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "shellAndLogins = grpdShell.map( \n",
    "    lambda x: (x[0], \",\".join( [ getLogin(y) for y in x[1] ]) ))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "shellAndLogins.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Rather than first group the keys and then combine the values into the string above, we can use **foldByKey** to do more or less the same thing -- this combines the mapping phase implicit in the list comprehension above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "byShell.foldByKey( \"\", lambda x,y: x + getLogin(y) + ',' ).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Internally, this is done using a \"combiner(createCombiner, mergeValue, mergeCombiners)\", which turns an RDD[(K, V)] into a result of type RDD[(K, C)], for a \"combined type\" C.  Note that V and C can be different -- for example, one might group an RDD of type (Int, Int) into an RDD of type (Int, List[Int]).\n",
    "\n",
    "This example takes the byShell (K,V) list and constructs a new V that is the name of the users of that shell. Entries within the same RDD partition are joined by a comma and between partitions by \"AND\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "byShell.combineByKey( getLogin,\n",
    "                        lambda xs, x: xs + ',' + getLogin(x),\n",
    "                        lambda xs, ys: xs + ' AND ' + ys ).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**combineByKey** is used to develop \"reduceByKey\" functions that can e.g. sum up the items associated with a key. This is effectively doing a **groupByKey** followed by a **reduce** on each list of values"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "c = sc.parallelize([ (1,2), (2,3), (1, 99), (3, 44), (2, 1), (4,5), (3, 19) ] )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `groupByKey` operator gives us `iterable` items."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "c.groupByKey().collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can reify (make concrete) the iterable item by convert it to a list:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "c.groupByKey().map(lambda x : (x[0], list(x[1]))).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `reduceByKey` groups identical keys and then applys a function over the iterable results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "c.reduceByKey( operator.add ).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## CoGroup - the basis for joins\n",
    "\n",
    "As in Pig, joins are done performing \"co-groups\" where multiple data sets are grouped by the same key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "s1 = c\n",
    "s1.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "s2 = sc.parallelize( [ (2, -99), ( 4, 199), (19, 23) ] )\n",
    "s2.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "co = s1.cogroup(s2)\n",
    "co.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's use our \"convert it to a list trick\" to see what's in each cogroup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "co.map(lambda x: (x[0], list(x[1][0]), list(x[1][1])) ).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is used to build different kinds of joins"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "s1.join(s2).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "s1.leftOuterJoin(s2).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "s1.rightOuterJoin(s2).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "x = sc.parallelize( [ (\"NY\", 10), (\"OH\", 20), (\"OH\", 99), (\"CO\", 88) ] )\n",
    "y = sc.parallelize( [ (\"NY\", 30), (\"CO\", 40), (\"NY\", 22 )] )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "x.join(y).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "x.join(y).flatMap(lambda kv: ((kv[0],x) for x in kv[1])).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "x.cogroup(y).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "x.cogroup(y).flatMap(lambda kv: ((kv[0],x,y) for x in kv[1][0] for y in kv[1][1])).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}