{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Table Joins Tutorial\n",
    "\n",
    "This tutorial walks through some ways to join Hail tables. We'll use a simple movie dataset to illustrate. The movie dataset comes in multiple parts. Here are a few questions we might naturally ask about the dataset:\n",
    "\n",
    "- What is the mean rating per genre?\n",
    "- What is the favorite movie for each occupation?\n",
    "- What genres are most preferred by women vs men?\n",
    "\n",
    "We'll use joins to combine datasets in order to answer these questions. \n",
    "\n",
    "Let's initialize Hail, fetch the tutorial data, and load three tables: users, movies, and ratings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import hail as hl\n",
    "\n",
    "hl.utils.get_movie_lens('data/')\n",
    "\n",
    "users = hl.read_table('data/users.ht')\n",
    "movies = hl.read_table('data/movies.ht')\n",
    "ratings = hl.read_table('data/ratings.ht')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The Key to Understanding Joins\n",
    "\n",
    "To understand joins in Hail, we need to revisit one of the crucial properties of tables: the key.\n",
    "\n",
    "A table has an ordered list of fields known as the key. Our `users` table has one key, the `id` field. We can see all the fields, as well as the keys, of a table by calling `describe()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "users.describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`key` is a struct expression of all of the key fields, so we can refer to the key of a table without explicitly specifying the names of the key fields."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "users.key.describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Keys need not be unique or non-missing, although in many applications they will be both.\n",
    "\n",
    "When tables are joined in Hail, they are joined based on their keys. In order to join two tables, they must share the same number of keys, same key types (i.e. string vs integer), and the same order of keys."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's look at a simple example of a join. We'll use the `Table.parallelize()` method to create two small tables, `t1` and `t2`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1 = hl.Table.parallelize([\n",
    "    {'a': 'foo', 'b': 1},\n",
    "    {'a': 'bar', 'b': 2},\n",
    "    {'a': 'bar', 'b': 2}],\n",
    "    hl.tstruct(a=hl.tstr, b=hl.tint32),\n",
    "    key='a')\n",
    "t2 = hl.Table.parallelize([\n",
    "    {'t': 'foo', 'x': 3.14},\n",
    "    {'t': 'bar', 'x': 2.78},\n",
    "    {'t': 'bar', 'x': -1},\n",
    "    {'t': 'quam', 'x': 0}],\n",
    "    hl.tstruct(t=hl.tstr, x=hl.tfloat64),\n",
    "    key='t')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t2.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we can join the tables. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "j = t1.annotate(t2_x = t2[t1.a].x)\n",
    "j.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's break this syntax down. \n",
    "\n",
    "`t2[t1.a]` is an expression referring to the row of table `t2` with value `t1.a`. So this expression will create a map between the keys of `t1` and the rows of `t2`. You can view this mapping directly:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t2[t1.a].show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since we only want the field `x` from `t2`, we can select it with `t2[t1.a].x`. Then we add this field to `t1` with the `anntotate_rows()` method. The new joined table `j` has a field `t2_x` that comes from the rows of `t2`. The tables could be joined, because they shared the same number of keys (1) and the same key type (string). The keys do not need to share the same name. Notice that the rows with keys present in `t2` but not in `t1` do not show up in the final result. This join syntax performs a left join. Tables also have a SQL-style inner/left/right/outer [join()](https://hail.is/docs/0.2/hail.Table.html#hail.Table.join) method.\n",
    "\n",
    "The magic of keys is that they can be used to create a mapping, like a Python dictionary, between the keys of one table and the row values of another table: `table[expr]` will refer to the row of `table` that has a key value of `expr`. If the row is not unique, one such row is chosen arbitrarily.\n",
    "\n",
    "Here's a subtle bit: if `expr` is an expression indexed by a row of `table2`, then `table[expr]` is also an expression indexed by a row of `table2`.\n",
    "\n",
    "Also note that while they look similar, `table['field']` and `table1[table2.key]` are doing very different things!\n",
    "\n",
    "`table['field']` selects a field from the table, while `table1[table2.key]` creates a mapping between the keys of `table2` and the rows of `table1`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t1['a'].describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t2[t1.a].describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Joining Tables\n",
    "\n",
    "Now that we understand the basics of how joins work, let's use a join to compute the average movie rating per genre.\n",
    "\n",
    "We have a table `ratings`, which contains `user_id`, `movie_id`, and `rating` fields. Group by `movie_id` and aggregate to get the mean rating of each movie. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = (ratings.group_by(ratings.movie_id) \n",
    "     .aggregate(rating = hl.agg.mean(ratings.rating)))\n",
    "t.describe()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To get the mean rating by genre, we need to join in the genre field from the `movies` table. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = t.annotate(genres = movies[t.movie_id].genres)\n",
    "t.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We want to group the ratings by genre, but they're packed up in an array. To unpack the genres, we can use [explode](https://hail.is/docs/0.2/hail.Table.html#hail.Table.explode). \n",
    "\n",
    "`explode` creates a new row for each element in the value of the field, which must be a collection (array or set)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = t.explode(t.genres)\n",
    "t.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, we can get group by genre and aggregate to get the mean rating per genre."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "t = (t.group_by(t.genres)\n",
    "     .aggregate(rating = hl.agg.mean(t.rating)))\n",
    "t.show(n=100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's do another example. This time, we'll see if we can determine what the highest rated movies are, on average, for each occupation. We start by joining the two tables `movies` and `users.`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "movie_data = ratings.annotate(\n",
    "    movie = movies[ratings.movie_id].title,\n",
    "    occupation = users[ratings.user_id].occupation)\n",
    "\n",
    "movie_data.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we'll use `group_by` along with the aggregator `hl.agg.mean` to determine the average rating of each movie by occupation. Remember that the `group_by` operation is always associated with an aggregation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings_by_job = movie_data.group_by(\n",
    "    movie_data.occupation, movie_data.movie).aggregate(\n",
    "    mean = hl.agg.mean(movie_data.rating))\n",
    "\n",
    "ratings_by_job.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can use another `group_by` to determine the highest rated movie, on average, for each occupation.\n",
    "\n",
    "The syntax here needs some explaining. The second step in the cell below is just to clean up the table created by the preceding step. If you examine the intermediate result (for example, by giving a new name to the output of the first step), you will see that there are two columns corresponding to occupation, `occupation` and `val.occupation`. This is an artifact of the aggregator syntax and the fact that we are retaining the entire row from `ratings_by_job`. So in the second line, we use `select` to keep those columns that we want, and also rename them to drop the `val.` syntax.  Since `occupation` is a key of this table, we don't need to select for it. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "highest_rated = ratings_by_job.group_by(\n",
    "    ratings_by_job.occupation).aggregate(\n",
    "    val = hl.agg.take(ratings_by_job.row,1, ordering = -ratings_by_job.mean)[0]\n",
    ")\n",
    "\n",
    "highest_rated = highest_rated.select(movie = highest_rated.val.movie,\n",
    "                                    mean = highest_rated.val.mean)\n",
    "\n",
    "highest_rated.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's try to get a deeper understanding of this result. Notice that every movie displayed has an *average* rating of 5, which means that every person gave these movies the highest rating. Is that unlikely? We can determine how many people rated each of these movies by working backwards and filtering our original `movie_data` table by fields in `highest_rated`.\n",
    "\n",
    "Note that in the second line below, we are taking advantage of the fact that Hail tables are keyed. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "highest_rated = highest_rated.key_by(\n",
    "    highest_rated.occupation, highest_rated.movie)\n",
    "\n",
    "counts_temp = movie_data.filter(\n",
    "    hl.is_defined(highest_rated[movie_data.occupation, movie_data.movie]))\n",
    "\n",
    "counts = counts_temp.group_by(counts_temp.occupation, counts_temp.movie).aggregate(\n",
    "    counts = hl.agg.count())\n",
    "\n",
    "counts.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So it looks like the highest rated movies, when computed naively, mostly have a single viewer rating them. To get a better understanding of the data, we can recompute this list but only include movies which have more than 1 viewer (left as an exercise).  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercises\n",
    "\n",
    "- What is the favorite movie for each occupation, conditional on there being more than one viewer?\n",
    "- What genres are rated most differently by men and women?\n",
    " "
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}