lab_exercises/lab2_1_dataframes.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# **Exercise 2: Spark SQL**\n",
    " This second exercise will introduce database operations with Spark. \n",
    "\n",
    "\n",
    "##### During the exercises, the following resources might come in handy:\n",
    "*  Documentation of the [PySpark API](https://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD)\n",
    "*  Documentation of the [Python API](https://docs.python.org/2.7/)\n",
    "*  Documentation of the [Spark SQL API](http://spark.apache.org/docs/latest/sql-programming-guide.html)\n",
    "*  Documentation of [Hive SQL](https://cwiki.apache.org/confluence/display/Hive/LanguageManual+UDF)\n",
    "\n",
    "##### To run code in Jupyter, press: \n",
    "*  `Ctrl-Enter` to run the code in the currently selected cell\n",
    "*  `Shift-Enter` to run the code in the currently selected cell and jump to the next cell"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### **Helper: Displays rows from a Spark SQL object as HTML**\n",
    " The following code can display arrays of rows or dataframes as a html table. This creates a human-friendly output of our data. It is enough to browse through this code, as it is not important to know it in detail for this exercise."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.display import display, HTML\n",
    "import warnings\n",
    "\n",
    "def displayRows(rowDf):\n",
    "    headers = []\n",
    "    rows = []\n",
    "    if(str(type(rowDf)) == \"<class 'pyspark.sql.dataframe.DataFrame'>\"):\n",
    "        rows = rowDf.limit(10000).collect() #Let's limit the output just in case!\n",
    "        if(len(rows) == 10000):\n",
    "            if(rowDf.limit(10001).count() == 10001):\n",
    "                warnings.warn(\"More than 10 000 rows was returned, only showing the first 10 000.\")\n",
    "                \n",
    "        headers = list(rowDf.columns)\n",
    "    else:\n",
    "        rows = rowDf\n",
    "        if(len(rows) > 10000):\n",
    "            warnings.warn(\"Rows has {0} elements, only showing the first 10 000.\".format(len(rows)))\n",
    "            rows = rows[0:10000]\n",
    "            \n",
    "        #Computes the unique set of keys\n",
    "        headers = list(sorted(reduce(lambda x,y: x.union(set(y.asDict().iterkeys())), rows, set())))\n",
    "            \n",
    "    tableHead = [\"<th>{0}</th>\".format(key) for key in headers]\n",
    "    tableBody = [\"<tr>{0}</tr>\".format(\n",
    "                    \"\".join([\"<td>{0}</td>\".format(rowDict.get(header)) \n",
    "                            for rowDict \n",
    "                            in (row.asDict(),) \n",
    "                            for header \n",
    "                            in headers])\n",
    "                    ) for row in rows]\n",
    "    \n",
    "    display(HTML(\n",
    "    u\"\"\"<table>\n",
    "    <thead><tr>{0}</tr></thead>\n",
    "    <tbody>{1}</tbody>\n",
    "    </table>\n",
    "    \"\"\".format(\"\".join(tableHead), \"\".join(tableBody))))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Part 1: Learning Spark SQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " This part will introduce you the Spark SQL by writing SQL queries.\n",
    "\n",
    "The cell below generates data which you will write queries for."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_list = ['apple', 'banana', 'grapes', 'pear']\n",
    "counter_list = list(enumerate(my_list, 1))\n",
    "print(counter_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "#Top 20 boy and girl names 2014 in random order.\n",
    "names = [\"Caden\", \"Kaylee\", \"Lucas\", \"Ethan\", \"Alexander\", \"Jackson\", \n",
    "         \"Aiden\", \"Madelyn\", \"Michael\", \"Avery\", \"Luke\", \"Isabella\", \n",
    "         \"Chloe\", \"Elijah\", \"Abigail\", \"Madison\", \"Jacob\", \"Zoe\", \"Emily\", \n",
    "         \"Jayden\", \"Liam\", \"Mason\", \"Mia\", \"Sophia\", \"Benjamin\", \"Layla\", \n",
    "         \"Emma\", \"Lily\", \"Charlotte\", \"Caleb\", \"James\", \"Noah\", \"Ella\", \n",
    "         \"Jack\", \"Jayce\", \"Aubrey\", \"Olivia\", \"Harper\", \"Logan\", \"Ava\"]\n",
    "#A-G in phonetic alphabet\n",
    "groups = [\"Alpha\",\"Bravo\", \"Charlie\", \"Delta\", \"Echo\", \"Foxtrot\", \"Golf\"]\n",
    "\n",
    "#Some numeric magic to generate not so uniform random data.\n",
    "tblUserRdd = sc.parallelize(map(lambda i: (i, ((i*104729)^131) % 7, 26500 + ((i*104729)^96587) % 6367), range(1,51)))\n",
    "# Parallelize data using 4 partitions\n",
    "tblNamesRdd = sc.parallelize(enumerate(names, 1), 4)\n",
    "# Parallelize data using 2 partitions\n",
    "tblGroupNamesRdd = sc.parallelize(enumerate(groups), 2)\n",
    "\n",
    "#Create dataframes from the RDDs\n",
    "tblNames      = sqlContext.createDataFrame(tblNamesRdd,      [\"userId\", \"name\"])\n",
    "tblUsers      = sqlContext.createDataFrame(tblUserRdd,       [\"id\", \"groupId\", \"salary\"])\n",
    "tblGroupNames = sqlContext.createDataFrame(tblGroupNamesRdd, [\"id\", \"name\"])\n",
    "\n",
    "#Register them for use with SQL-TRADITIONAL.\n",
    "sqlContext.registerDataFrameAsTable(tblGroupNames, \"tblGroupNames\")\n",
    "sqlContext.registerDataFrameAsTable(tblUsers, \"tblUsers\")\n",
    "sqlContext.registerDataFrameAsTable(tblNames, \"tblNames\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, lets get some basic information about each dataframe\n",
    "\n",
    "Dataframes are structured meaning that types and columns are well-defined; if you have read the data generation cell you might have noticed that the types were not specified. These are inferred by Spark."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dataframes provide a very handy function called `printSchema()`. As its name implies, it shows the schema of the data, including column names and types."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tblUsers.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It is possible to call a number of operations on dataframe, similar to RDDs dataframes have a `count()` action to display the number of rows in the dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tblUsers.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tblGroupNames.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tblGroupNames.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tblUsers.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tblNames.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " Next 3 cells will display the content of the dataframe by using the helper function *displayRows*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "displayRows(tblUsers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "displayRows(tblNames)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "displayRows(tblGroupNames)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There is a basic function for displaying the contents of an Dataframe by using *show()*\n",
    "However, the output is limited and gives a limited view of a long column. It is useful for debugging."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tblUsers.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, the first query you will write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 1.a) Write a query that selects all user ids in the group with id 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Replace <FILL IN> with the proper code\n",
    "q1a = sqlContext.sql(\"\"\"\n",
    "SELECT id \n",
    "FROM tblUsers \n",
    "WHERE <FILL IN>\n",
    "\"\"\")\n",
    "\n",
    "displayRows(q1a)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert set(map(lambda row: row.id, q1a.collect())) == set([11,26,27])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 1.b) Write a query that finds the min and max userId grouped by groupId\n",
    "\n",
    "The result should have the following columns:\n",
    "\n",
    "1. minUserId: The min user id per group\n",
    "2. maxUserId: The max user id per group\n",
    "2. groupId: The group id\n",
    "\n",
    "**Hint:** Use GROUP BY, MIN, MAX"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q1b = sqlContext.sql(\"\"\"\n",
    "SELECT \n",
    "    <FILL IN> AS minUserId, \n",
    "    <FILL IN> AS maxUserId,\n",
    "    <FILL IN> \n",
    "FROM tblUsers \n",
    "<FILL IN>\n",
    "\"\"\")\n",
    "\n",
    "displayRows(q1b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "minIds = {0: 11,\n",
    " 1: 12,\n",
    " 2: 6,\n",
    " 3: 8,\n",
    " 4: 24,\n",
    " 5: 1,\n",
    " 6: 4}\n",
    "\n",
    "maxIds = {0: 27,\n",
    " 1: 43,\n",
    " 2: 46,\n",
    " 3: 39,\n",
    " 4: 40,\n",
    " 5: 47,\n",
    " 6: 50}\n",
    "\n",
    "assert all(map(lambda row: minIds[row.groupId] == row.minUserId, q1b.collect()))\n",
    "assert all(map(lambda row: maxIds[row.groupId] == row.maxUserId, q1b.collect()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 1.c) Compute the global average salary"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When you do not specify any group by columns and use aggregating functions such as **AVG**(column) then the aggregation will be performed over the entire result and return a single row."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "avgSalary = sqlContext.sql(\"\"\"\n",
    "SELECT <FILL IN> AS avgSalary \n",
    "FROM tblUsers\n",
    "\"\"\").collect()[0].avgSalary\n",
    "\n",
    "avgSalary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assert avgSalary == 29707.34"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### 1.d) Aggregate salaries per group\n",
    "\n",
    "Group per groupId and compute the minimum, average, maximum salary and sort by average salary descending.\n",
    "\n",
    "**Hint:** Use MIN, AVG, MAX, GROUP BY, you can sort by computed columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q1d = sqlContext.sql(\"\"\"\n",
    "SELECT \n",
    "    groupId,\n",
    "    COUNT(id) AS NumUsers,\n",
    "    <FILL IN> AS MinSalary, \n",
    "    <FILL IN> AS AvgSalary,\n",
    "    <FILL IN> AS MaxSalary,\n",
    "    AVG(salary) - {} AS GlobalAvgDelta\n",
    "FROM tblUsers\n",
    "<FILL IN>\n",
    "ORDER BY <FILL IN>\n",
    "\"\"\".format(avgSalary))\n",
    "\n",
    "displayRows(q1d)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "groups = [\n",
    "    (5, 26623, 30573, 32862), \n",
    "    (4, 26923, 29898, 31849),\n",
    "    (1, 26600, 29833, 32234),\n",
    "    (2, 27784, 29537, 32796),\n",
    "    (6, 26973, 29490, 32245),\n",
    "    (3, 27858, 29447, 32531),\n",
    "    (0, 26784, 28369, 30346)\n",
    "]\n",
    "\n",
    "q1dresult = q1d.collect()\n",
    "assert len(q1dresult) == 7\n",
    "assert map(lambda i: q1dresult[i].groupId == groups[i][0], range(0,len(q1dresult)-1)), \"GroupID column does not match.\"\n",
    "assert map(lambda i: q1dresult[i].MinSalary == groups[i][1], range(0,len(q1dresult)-1)), \"MinSalary column does not match.\"\n",
    "assert map(lambda i: int(q1dresult[i].AvgSalary) == groups[i][2], range(0,len(q1dresult)-1)), \"AvgSalary column does not match.\"\n",
    "assert map(lambda i: q1dresult[i].MaxSalary == groups[i][3], range(0,len(q1dresult)-1)), \"MaxSalary column does not match.\""
   ]
  }
 ],
 "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.14"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}