notebooks/twitter_dataframes/twitter_dataframes-EMPTY.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from __future__ import print_function\n",
    "%matplotlib inline\n",
    "import matplotlib.pylab as plt\n",
    "import sys, os, glob\n",
    "import numpy as np\n",
    "import subprocess\n",
    "\n",
    "from ipywidgets import interact, interactive, fixed\n",
    "import ipywidgets as widgets\n",
    "\n",
    "from IPython.display import HTML\n",
    "import xml.etree.ElementTree as ET\n",
    "try:\n",
    "    tree = ET.parse(os.environ['HADOOP_CONF_DIR'] + '/yarn-site.xml')\n",
    "except IOError:\n",
    "    raise IOError(\"Can't find the yarn configuration -- is HADOOP_CONF_DIR set?\")\n",
    "root = tree.getroot()\n",
    "yarn_web_app = root.findall(\"./property[name='yarn.resourcemanager.webapp.address']\")[0].find('value').text\n",
    "yarn_web_app_string = \"If this works successfully, you can check the <a target='_blank' href='http://{yarn_web_app}'>YARN application scheduler</a> and you should see your app listed there. Clicking on the 'Application Master' link will bring up the familiar Spark Web UI. \"\n",
    "\n",
    "plt.rcParams['figure.figsize'] = (10,6)\n",
    "plt.rcParams['font.size'] = 18\n",
    "plt.style.use('fivethirtyeight')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Set up and launch the Spark runtime\n",
    "\n",
    "Remember from the previous notebook that we have a saved configuration in `./spark_config/` -- so all we need to do is set the `SPARK_CONF_DIR` environment variable and our default configuration will be used: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# specify the configuration directory\n",
    "os.environ['SPARK_CONF_DIR'] = os.path.realpath('./spark_config')\n",
    "\n",
    "# how many cores do we have for the driver\n",
    "ncores = int(os.environ.get('LSB_DJOB_NUMPROC', 1)) \n",
    "\n",
    "# here we set the memory we want spark to use for the driver JVM\n",
    "os.environ['SPARK_DRIVER_MEMORY'] = '%dG'%(ncores*0.7)\n",
    "\n",
    "# we have to tell spark which python executable we are using\n",
    "os.environ['PYSPARK_PYTHON'] = subprocess.check_output('which python', shell=True).rstrip()\n",
    "\n",
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "import pyspark\n",
    "from pyspark import SparkConf, SparkContext\n",
    "\n",
    "conf = SparkConf()\n",
    "\n",
    "sc = SparkContext(master='yarn-client', conf=conf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Uncovering World Events using Twitter Hashtags\n",
    "## ... and learning about Spark `DataFrames` along the way\n",
    "\n",
    "In this notebook, we will use temporal information about Twitter hashtags to uncover world events. In case someone isn't familiar with [Twitter](http://twitter.com): a \"tweet\" is a 140 character message broadcast into the world.  A hashtag is a part of a tweet, a single words preceded by a \"#\", and is used to indicate the topic of the tweet. If you want to find out about photography, for example, you might search for the [#photography](https://twitter.com/search?q=%23photography&src=typd) hashtag. \n",
    "\n",
    "The idea here is that when an event is happening and people are having a conversation about it on Twitter, a set of uniform hashtags that represent the event spontaneously evolves and Twitter users then use those hashtags to communicate with one another. Some hashtags, like \"#RT\" for \"retweet\" or just \"#retweet\" are used frequently and don't tell us much about what is going on. But a sudden appearance of a hashtag like \"#oscars\" probably indicates that the oscars are underway. For a particularly cool example of this type of analysis, check out this [blog post about earthquake detection using Twitter data](https://blog.twitter.com/2015/usgs-twitter-data-earthquake-detection) (although they search the text and not necessarily hashtags).\n",
    "\n",
    "We will do some simple statistics to try and find some of these event hashtags. This entire notebook uses the Spark `DataFrame`, which is a higher-level abstraction built on top of the `RDD` that we have grown to love. `DataFrame`s are designed to be efficient at handling tabular, structured data. Anything that has columns and rows, essentially. \n",
    "\n",
    "As we'll see below, the `DataFrame` allows us to process data efficiently and often use fewer lines of code. By using the optimized built-in functions, we can benefit from the performance saving of the pure scala implementation while still using Python. When something more complex is required that is not easily expressed in terms of `DataFrame` operations, we can temporarily switch into `RDD` mode. For some more background, see the [DataFrame API blog post](https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html). "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-info\">\n",
    "<p><strong>Viewer discretion is advised!</strong></p> \n",
    "\n",
    "<p>The internet is, in general, not a safe place. The data used in this tutorial is an unfiltered stream of what people all over the world tweet. The chances of there being something more or less offensive in this sea of tweets are pretty high, so be prepared to perhaps see some foul language and/or sexual references. </p>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initialization\n",
    "\n",
    "Using `DataFrame`s, our entry point into the Spark universe is the `SQLContext` or the equivalent `HiveContext`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext, HiveContext\n",
    "\n",
    "hc = HiveContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll load the data off the disk, but only for the last three months of the year. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "data = hc.read.parquet('/user/roskarr/twitter/2014_1*')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`data` is now a `DataFrame` object, which is essentially a collection of `Row` objects. Each `Row` object contains data for whatever columns are defined in the `DataFrame`. Here is a critical difference between `DataFrames` and `RDD`s: each column has an associated data type. While in dealing with an `RDD` we relied on Python to convert types, here each column has a specified data type. This means that:\n",
    "\n",
    "a) we have to be a bit more careful about what we are doing and \n",
    "\n",
    "b) the execution engine can optimize our calculations because the data is no longer a black box. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Importing libraries and types for `DataFrame` API\n",
    "\n",
    "The next few cells are a bit lengthy and complicated so you can just treat them as black boxes for now. First, we just import some necessary libraries and classes, then a series of functions are defined and executed on the `data` to give us a `DataFrame` of hashtags. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as func\n",
    "from pyspark.sql import Row, Window\n",
    "from pyspark.sql.types import IntegerType, ArrayType, StringType, NullType, LongType, StructField, StructType, DateType, DataType, DateConverter, DatetimeConverter, TimestampType, BooleanType\n",
    "import datetime"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Custom functions to extract hashtags\n",
    "\n",
    "The twitter data is rather complex and includes many different kinds of fields that we will not be using here. We will also change the format of the time string to something that Spark will have an easier time understanding. \n",
    "\n",
    "Here we define a few \"user defined functions\" (UDFs) -- these are crucial for working with `DataFrame`s and we will discuss them in more detail in a bit. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# change the date format to YYYY-MM-DD HH:MM:SS\n",
    "convert_date_string = func.udf(lambda date_string: datetime.date.strftime(datetime.datetime.strptime(date_string, '%a %b %d %H:%M:%S +0000 %Y'),'%Y-%m-%d %H:%M:%S'), StringType())\n",
    "\n",
    "# make UDF for converting the date string to a datetime object\n",
    "datetime_udf = func.udf(lambda date_string: datetime.strptime(date_string, '%a %b %d %H:%M:%S +0000 %Y'), DateType())\n",
    "\n",
    "# make a UDF \n",
    "hash_text_udf = func.udf(lambda row: [r.text.lower() for r in row], returnType=ArrayType(StringType()))\n",
    "\n",
    "Npartitions = sc.defaultParallelism*5\n",
    "\n",
    "# only keep the tweets with at least one hashtag\n",
    "hashtag_df = (data.select('created_at', 'entities.hashtags')\n",
    "                .filter(func.size('hashtags') > 0)\n",
    "                .withColumn('hash_text', hash_text_udf('hashtags'))\n",
    "                .select(convert_date_string('created_at').alias('date'), func.explode('hash_text').alias('hashtag'))\n",
    "                .withColumn('hashtag', func.lower(func.col('hashtag')))\n",
    "                .repartition(Npartitions))\n",
    "hashtag_df.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### A look at the raw data\n",
    "\n",
    "We now have a `DataFrame` consisting of a timestamp and hashtag. The `show()` method is extremelly useful here -- it functions like `take()` for an RDD, but it formats the output nicely so it's more easily readable. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hashtag_df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Underneath the `DataFrame` object is an RDD of `Row` objects, which are just containers used by `DataFrame`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hashtag_df.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The structure of `Row`s is revealed by the `DataFrame` schema: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hashtag_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In `DataFrame`s, you can think of the `select` method to be analogous to `map` in an `RDD`. In its simplest form, `select` is just used to, well, select the columns that are needed: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: use 'select' to return a DataFrame consisting of just the 'date' column\n",
    "hashtag_df.<FILL IN>.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now use `select` to get multiple columns, `'date'` and `'hashtag'`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hashtag_df.<FILL IN>.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In addition, we can pass functions to `select` that will operate on existing columns and define a new column. For example, lets use the `month` function to extract the month from the date string: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hashtag_df.select('hashtag', func.month('date')).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can see other pre-defined functions in the [`pyspark.sql` documentation](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#module-pyspark.sql.functions).\n",
    "\n",
    "If some special functionality is needed, we can define a custom \"user defined function\" (UDF) that uses the standard python interpreter. The function has to be registered as a UDF using [`pyspark.sql.functions.udf`](https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.functions.udf). For example: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def string_length(x):\n",
    "    return len(x)\n",
    "\n",
    "string_length_udf = func.udf(string_length, LongType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hashtag_df.select('hashtag', string_length_udf('hashtag')).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `DataFrame` superpower:  `groupBy`\n",
    "\n",
    "`DataFrame`'s `groupBy()` method to allows you to collect together rows that share a common value. This is a powerful method that allows you to perform aggregations on your data quickly. It can serve as a replacement to `reduceByKey` that is used on `RDD`s and requires considerably less wrangling with keys and values. \n",
    "\n",
    "This is also where things get a little confusing: unlike other `DataFrame` and `RDD` methods, `groupBy` doesn't return another `DataFrame`. Instead, it returns a `GroupedData` object, which has very limited functionality. It can *only* perform the dedicated aggregation functions `avg`, `count`, `max`, `mean`, `min`, and `sum`. You can specify a column (or columns) for the aggregation for all of these except `count`, which just counts the number of rows in each group. These methods all return another `DataFrame`, so the `GroupedData` is a sort of intermediate stage. \n",
    "\n",
    "Here, we will try to find out how many hashtags are tweeted each minute:\n",
    "\n",
    "1. create a custom UDF that calculates minute-of-the-year for each hashtag\n",
    "2. use `groupBy` aggregation methods to obtain counts"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# here we use all built-in functions (starting with 'func') and pass in the 'date' column\n",
    "minute_of_the_year = (func.dayofyear('date')*60*24 + func.hour('date')*60 + func.minute('date')).alias('moy')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "moy_df = hashtag_df.select('hashtag', minute_of_the_year)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we have a column 'moy' (stands for \"minute-of-year\") calculated using the `moy_df` function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "moy_df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We group the data by minute-of-year:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "moy_group = moy_df.groupBy('moy')\n",
    "moy_group"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that `moy_group` now contains groups of rows that all have the same value of `'moy'`. `groupBy` returns a `GroupedData` instance, as you can see above -- which is not quite a `DataFrame`, but has methods that return `DataFrame`s. Calling `count()` will give us back a `DataFrame` with a new column `'count'` that simply gives the number of items in each group (in this case, this means it represents tweets in each minute of the year):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "moy_count = moy_group.count()\n",
    "moy_count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "moy_count.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can collect this \"reduced\" data into a Pandas dataframe and plot it: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "moy_data = moy_count.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "fig, ax = plt.subplots()\n",
    "plt.plot(moy_data['moy'][:1000], moy_data['count'][:1000], 'o', alpha=.5)\n",
    "plt.xlabel('minute of year'); plt.ylabel('hashtags per minute')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For fun, lets diverge briefly and check for any obvious periods in there..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import scipy.fftpack\n",
    "\n",
    "N = len(moy_data['count'])\n",
    "T = (moy_data['moy'].max() - moy_data['moy'].min())/N\n",
    "\n",
    "yf = scipy.fftpack.fft(moy_data['count'])\n",
    "\n",
    "xf = np.linspace(0.0, 1.0/(2.0*T), N/2)\n",
    "\n",
    "fig, ax = plt.subplots()\n",
    "ax.plot(1./xf/60.,2.0/N * np.abs(yf[0:N/2]))\n",
    "ax.set_xlim(0,36)\n",
    "ax.set_xlabel('period (hours)'); ax.set_ylabel('power');"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A lot of structure in there! some of it might be harmonics, but looking into periodicity on Twitter [has been done before](http://theory.stanford.edu/~nmishra/Papers/groupChatsOnTwitter.pdf) (obviously)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Simple word count revisited\n",
    "\n",
    "OK, back to Spark. Now that we have a `DataFrame` consisting of `(timestamp, hashtag)` columns, lets do a simple word count (or hashtag count in this case) using the `DataFrame` API. \n",
    "\n",
    "The procedure to do this with a `DataFrame` is as follows (look at the minute-of-year counting we did above if you need a reference):\n",
    "\n",
    "1. group the `DataFrame` by 'hashtag'\n",
    "2. use `count()` to get the number of members in each group\n",
    "3. sort by the `'count'` column\n",
    "4. display a few of the top results using `show()`\n",
    "\n",
    "Remember that using RDDs, the word count was done like this: \n",
    "\n",
    "1. map all words to `(word,1)` tuples\n",
    "2. use `reduceByKey` to get the counts for each word \n",
    "\n",
    "To accomplish this with our `DataFrame`, we'll once again use the `groupBy` functionality:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time \n",
    "(hashtag_df.<FILL IN>)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here's the equivalent RDD word count:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "res = (hashtag_df.map(lambda r: (r.hashtag, 1))\n",
    "           .reduceByKey(lambda a,b:a+b)\n",
    "           .sortBy(lambda (k,count): count, ascending=False)).take(25)\n",
    "for x in res: \n",
    "    print(x[0], x[1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " As we can see here, there are lots os languages represented on Twitter! This is very interesting by itself, but for now lets focus on just plain ascii hash tags which we are more likely to understand: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# define a helper UDF that can be used to filter non-ascii hashtags\n",
    "def is_ascii(s):\n",
    "    \"\"\"simple function returning True if all of the characters in the string `s` \n",
    "    are standard ascii and False otherwise\"\"\"\n",
    "    return all(ord(c) > 0 and ord(c) < 128 for c in s)\n",
    "is_ascii_udf = func.udf(is_ascii, BooleanType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "hashtag_df = hashtag_df.filter(is_ascii_udf('hashtag'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Daily hashtag trends\n",
    "\n",
    "Now lets do something slightly more complicated and incorporate the time data into our analysis. \n",
    "\n",
    "Lets say we want to know how many times a hashtag was used per day. We can get this information in a few steps: \n",
    "\n",
    "1. first, we need to generate a 'day of year' value from each timestamp\n",
    "2. \n",
    "\n",
    "\n",
    "\n",
    "To get information about daily hashtag usage, we must first convert the date string into a number representing day of the year. The [Spark DataFrame functions module](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#module-pyspark.sql.functions) provides a function [dayofyear](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.functions.dayofyear) that will do this for us and we can use it just like we used the User Defined Functions (UDFs) above. \n",
    "\n",
    "We'll keep the week and the month around as well so we can use them later: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "daily_hashtag = hashtag_df.select(func.month('date').alias('month'),\n",
    "                                  func.weekofyear('date').alias('week'),\n",
    "                                  func.dayofyear('date').alias('day'), \n",
    "                                  'hashtag')\n",
    "daily_hashtag.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now use the same technique as above using `groupBy` to get daily counts for each hashtag. Since we want to satisfy two conditions (counts per day and counts per hashtag), we simply give `groupBy` two columns:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: group 'daily_hashtag' by 'day', 'hashtag', and 'week' (we'll need 'week' later); \n",
    "# then use the 'count()' method to get daily counts of each hashtag\n",
    "day_counts = daily_hashtag.<FILL IN>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's time how long calculating and sorting this full new `DataFrame` takes -- we'll compare this with the RDD method below: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%time day_counts.sort('day').count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Important aside concerning `count()` (read this while waiting for the above line to complete)\n",
    "\n",
    "The use of `count()` here is very confusing -- in the first of the two cells above, `count()` is a method of a [GroupedData](http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.GroupedData) object, which performs a counting of items in each group. Our grouping here is defined by week, day, and hashtag, so calling `count()` gives us the number of times a hashtag appears each day of the year (week here has no effect since days are unique anyway). To make it even more confusing, `count()` in this case actually returns a `DataFrame`, with \"count\" as one of the columns. \n",
    "\n",
    "However, we are much more used to seeing `count()` as an `RDD` method! This is done in the second of the two cells above. There, it returns a single number representing the number of elements in the `RDD` (or `DataFrame`), which is much different. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "day_counts.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### `RDD` analogy:\n",
    "\n",
    "The same counting of daily occurences of each hashtag can be done with an `RDD` operation using a `map` and `reduceByKey`, though it might look a bit messier: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "day_counts_rdd = daily_hashtag.rdd.map(lambda r: ((r.day, r.hashtag), 1)).reduceByKey(lambda a,b: a+b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%time day_counts_rdd.sortBy(lambda ((day, hashtag), count): day).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Wow, that takes way longer! The `DataFrame` API avoids shipping data back and forth to the Python intepreter (and can do some extra optimizations) so the performance difference can be significant, as we can see above. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets make sure that the results are the same by looking up a hashtag in both places:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "day_counts_rdd.lookup((274,'retweet'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "day_counts.filter((day_counts.hashtag == 'retweet') & (day_counts.day == 274)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At this point we should persist `day_counts` in memory since we'll be using it later on. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "day_counts.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can get a quick idea of the most popular hashtags by looking at the weekly averages of daily hashtag counts:\n",
    "\n",
    "1. group `day_counts` by 'week' and 'hashtag'\n",
    "2. use `avg` to get the average `count` per week\n",
    "3. sort by `avg(count)` in descending order \n",
    "4. display a few of the top results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "(day_counts.<FILL IN>\n",
    "           .<FILL IN>\n",
    "           .sort('avg(count)', ascending=False)\n",
    "           .show(20))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Apparently, the [MTV Europe Music Awards](https://en.wikipedia.org/wiki/2014_MTV_Europe_Music_Awards) happened in late 2014... (and [yet another boy band](https://en.wikipedia.org/wiki/One_Direction) is on the scene?)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using `Window` functions\n",
    "\n",
    "### top daily hashtags"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now lets say we want to know the top tweets for each day. This is a non-trivial thing to try and compute using a standard RDD (can you think of how you would do it?), but the `DataFrame` API gives us [\"window\"](https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html) functions that let us do it relatively easily. \n",
    "\n",
    "A window function defines how the data should be grouped and ordered. Then, when using either a ranking or analytic function (see the list in the [window functions article](https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html)) we can specify which window to run the function over. \n",
    "\n",
    "In this example, we will define a window function, `daily_window`, over the `'day'` column so our data will be grouped by day. Applying `daily_window` to the built-in `rank` function will give us `daily_rank`, a function that will yield a new column `'rank'` that will hold the rank of each hashtag on a given day based on its counts (in descending order).\n",
    "\n",
    "In the end, this is a fairly complicated operation achieved in just a few lines of code! "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, specify the quantity to window over -- in this case we want rankings per day:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "daily_window = Window.partitionBy('day')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Second, define a function that we want to compute over the window\n",
    "\n",
    "Here, we want to know the daily ranking of hashtags, so our function is `func.rank()` and we want the hashtags ordered by descending `'count'` column values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "daily_rank = func.rank() \\ \n",
    "                 .over(daily_window.orderBy(func.desc('count'))) \\\n",
    "                 .alias('rank')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we apply the `daily_rank` function to create a new column in `day_counts`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "day_counts.select('day', 'count', 'hashtag', daily_rank) \\\n",
    "          .filter('rank <= 5')\n",
    "          .show(20)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### A more complicated `Window` example: calculating the daily standard deviation based on a weekly window\n",
    "\n",
    "The above list of top daily hashtags is interesting, but it really only uncovers very frequent hashtags (e.g. \"ff\" for \"friend forward\" or \"rt\" for \"retweet\"). We want to try and identify \"world events\" and for that we want to detect a rapid *change* in the tweeting pattern. \n",
    "\n",
    "One very simple model to do this is to just look at the standard deviation of daily tweets. We want to have some sort of reasonably fine-grained control, so we will look at top hashtags per week. Remember standard deviation is just:\n",
    "\n",
    "$$\\sigma = \\sqrt{\\frac{1}{N}\\sum_i{(x_i-\\mu)^2}},$$\n",
    "\n",
    "where $x_i$ is the daily count of a hashtag, $\\mu$ is the weekly average and $N$ is the number of samples (here $N=7$ since there are always seven days in a week). \n",
    "\n",
    "So the procedure is something like this: \n",
    "\n",
    "1. calculate $\\mu$ for each hastag and each week\n",
    "2. compute the sum and take the square root\n",
    "3. order tweets for each week in order of decreasing $\\sigma$\n",
    "\n",
    "Using standard `RDD`s, the first step would be the most time consuming -- we would need to do an expensive `reduceByKey` on `(week, hashtag, count)` key-value pairs and then either join this result with the original `RDD` or broadcast it as a lookup-table (depending on its size). \n",
    "\n",
    "Using a `DataFrame`, we can instead use a \"window\" function to do the same thing. \n",
    "\n",
    "#### Constructing the window function and calculating $\\sigma$\n",
    "\n",
    "Our procedure to calculate $\\sigma$ using a weekly mean is as follows:\n",
    "\n",
    "1. define a `Window` function that will group the data by hashtag and by week\n",
    "2. compute `mean_diff` which is the difference between the data point and the weekly mean using the window function\n",
    "2. calculate `var_arg`, which is just $1/7~\\mathrm{mean\\_diff}^2$\n",
    "3. use `groupBy` to sum up the `sum_arg` column for each hashtag each week\n",
    "4. calculate `sigma` by taking the square root of the variance "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# define the window function, partitioned over 'week' and 'hashtag'\n",
    "weekly_window = Window.partitionBy('week', 'hashtag')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# define the mean_diff function which computes the difference between each data point 'count' \n",
    "# and the mean of 'count' over the weekly window (defined above)\n",
    "mean_diff = day_counts['count'] - func.mean('count').over(weekly_window)\n",
    "\n",
    "var_arg = 1./7*mean_diff*mean_diff"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "weekly_stats = (day_counts.withColumn('var_arg', var_arg)\n",
    "                         .groupBy('week', 'hashtag')\n",
    "                         .sum('var_arg')\n",
    "                         .withColumnRenamed('sum(var_arg)', 'variance')\n",
    "                         .withColumn('stddev', func.sqrt('variance'))\n",
    "                         .sort('stddev', ascending=False))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "weekly_stats.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Top weekly hashtags by $\\sigma$\n",
    "\n",
    "To get the hashtags that had the most abrupt changes in their usage, we need to rank the hashtags for each week by the standard deviations that we calculated above. Here we use the same pattern as before for getting the top daily tweets:\n",
    "\n",
    "1. window over the `'week'` and order by '`stddev`'\n",
    "2. create a column of rank computed over this window\n",
    "3. display the top 10 for several weeks in a row"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "weekly_stats_window = Window.partitionBy('week').orderBy(func.desc('stddev'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print('Top ten hashtags with highest variation each week')\n",
    "weekly_stats.select('week', 'hashtag', 'stddev', \n",
    "                     func.rank().over(weekly_stats_window).alias('rank')) \\\n",
    "            .filter('rank < 10') \\\n",
    "            .show(150)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A lot of these are pop-culture events and probably PR campaigns, e.g. 'votefifthharmony' and 'vote5sos'. Turns out the MTV Europe Music Awards were happening during November 2014, so these are not surprising. Also, who knew Twitter was so big in Argentina... check out what [kcaaargentina](http://kidschoiceawardsargentina.mundonick.com/) refers to... (We also see another \"world event\" -- 'halloween' pop up in week 44. \n",
    "\n",
    "Interestingly, in week 48 we finally see something that we are after: [\"ferguson\"](https://en.wikipedia.org/wiki/Ferguson_unrest). This was a huge event covered world-wide and spanned several months -- it actually started in August, but in November the riots returned when the grand jury decided not to indict the police officer who shot Michael Brown. We'll explore this more a bit further down. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Plotting the timeseries of hashtag counts\n",
    "\n",
    "These standard deviation lists we obtained above are nice, but lets actually look at some data. Below we define a simple function for plotting time series data of hashtags. You can specify a single hashtag or a series of hashtags separated by a comma (','). Note that the time-series is normalized by the standard deviation because often the counts for different hashtags differ by orders of magnitude. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "minmax = day_counts.select(func.min('day').alias('min'), func.max('day').alias('max')).map(lambda r: (r.min,r.max)).collect()[0]\n",
    "\n",
    "from ipywidgets import interact, interactive, fixed\n",
    "import ipywidgets as widgets\n",
    "\n",
    "def plot_hashtag_timeseries(hashtags): \n",
    "    \n",
    "    if ',' in hashtags : hashtags = map(lambda x: x.strip(), hashtags.split(','))\n",
    "        \n",
    "    if type(hashtags) is not list:\n",
    "        hashtags = [hashtags]\n",
    "    max_data = 0\n",
    "    for hashtag in hashtags:\n",
    "        day_data = day_counts.filter(day_counts['hashtag'] == hashtag).select('day', 'count').sort('day').toPandas()\n",
    "        max_data = day_data['count'].max()\n",
    "        plt.plot(day_data['day'],day_data['count']/day_data['count'].std(), label = hashtag, alpha=.5)\n",
    "        \n",
    "    plt.xlabel('day of year')\n",
    "    plt.legend(fontsize='small', loc='upper left')\n",
    "    plt.xlim(minmax)\n",
    "interact(plot_hashtag_timeseries, hashtags='ferguson, fergusondecision');"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Identifying hashtag trends with K-Means clustering\n",
    "\n",
    "The lists above give us some indication of what was big news on Twitter in the last 2.5 months of 2014. We can now use a clustering approach to identify groups of hashtags that may have been a part of the same viral development. \n",
    "\n",
    "Spark includes a K-Means clustering algorithm in MLLib, but it doesn't take a `DataFrame` as input. Instead, we have to pass to it an `RDD` of vectors. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import SparseVector\n",
    "\n",
    "minday, maxday = day_counts.select(func.min('day').alias('min'), func.max('day').alias('max')).map(lambda r: (r.min,r.max)).collect()[0]\n",
    "\n",
    "Ndays = maxday-minday+1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For each hashtag, we want an array of daily occurence counts -- the procedure is a bit complex, but the steps go as follows:\n",
    "\n",
    "1. select `'hashtag'`, `'day'` and `'counts'` from `day_counts`\n",
    "2. map to an RDD of the form `(hashtag, (day, counts))`\n",
    "3. use `aggregateByKey` to build a list of `(day, count)` pairs for each hashtag\n",
    "4. use this list of `(day,count)` pairs to create a `SparseVector`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def add_tuple_to_vector(vec, tup):\n",
    "    \"\"\"Adds a new (day, count) tuple to the aggregation vector\"\"\"\n",
    "    day, count = tup\n",
    "    vec[day] += count\n",
    "    return vec"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "day_counts_vec = \\\n",
    "    day_counts.withColumnRenamed('count', 'counts') \\\n",
    "              .map(lambda r: (r.hashtag, (r.day-minday, r.counts))) \\\n",
    "              .aggregateByKey(np.zeros(Ndays), add_tuple_to_vector, lambda a,b: a+b) \n",
    "\n",
    "day_counts_vec.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "day_counts.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To filter out some of the noise, we'll get rid of everything with a mean count of less than 10 and with 'followme' in the name:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "day_counts_filtered = day_counts_vec.filter(lambda (h,vec): (vec.mean() > 10) & ('followme' not in h)).mapValues(lambda v: v/v.std()).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Building the model\n",
    "\n",
    "Now we have our RDD of vectors and we can feed it directly to the [KMeans clustering](https://en.wikipedia.org/wiki/K-means_clustering) algorithm from [Spark's MLlib](http://spark.apache.org/docs/latest/mllib-guide.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.clustering import KMeans\n",
    "model = KMeans.train(day_counts_filtered.values(), 500, runs=10, maxIterations=1000, initializationMode='random')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After we compute the model, we calculate which cluster each hashtag belongs to:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "clusters = day_counts_filtered.mapValues(lambda vec: model.predict(vec)).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Find the cluster for `'ferguson'` and plot the timeseries for a few hashtags from that cluster:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ferguson_cluster = clusters.lookup('ferguson')[0]\n",
    "\n",
    "hashtags = clusters.filter(lambda (hashtag, cluster): cluster == ferguson_cluster).keys().takeSample(False, 5, None)\n",
    "\n",
    "days = range(minday, maxday+1)\n",
    "for hashtag in hashtags: \n",
    "    vec = day_counts_filtered.lookup(hashtag)[0]\n",
    "    plt.plot(days, vec, label = hashtag, alpha = .5)\n",
    "plt.ylabel('normalized frequency'); plt.xlabel('days');\n",
    "plt.legend(fontsize='small', loc='upper left');"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Depending on which hashtags are randomly selected, these may or may not make sense, but for the most part you should see that they all peak at about the same time. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def extract_hashtag_cluster(hashtag):\n",
    "    cluster_id = clusters.lookup(hashtag)[0]\n",
    "    return clusters.filter(lambda (hashtag, cluster): cluster == ferguson_cluster).keys().collect()\n",
    "\n",
    "interact(extract_hashtag_cluster, hashtag='ferguson');"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These are obviously not all related to Ferguson, but many are (it's nice that it catches the misspelling of \"ferguson\" also...). You can see hashtags like \"fergusondecision\", \"justiceformikebrown\", \"ripmikebrown\" etc. that are all clearly related. \n",
    "\n",
    "Feel free to build your own algorithm and experiment to clean up the clustering results!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Word2Vec"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-info\" style=\"margin=10px\"><p>Only do the following section if a) you have plenty of time and b) you have a driver with many cores and lots of memory</p></div>\n",
    "\n",
    "This section is here just for fun because word2vec is implemented in Spark and is fun to play with -- it takes a while to train the model, so only run these next few cells if you're willing to wait a while (~15 minutes on 20 executors). \n",
    "\n",
    "[Word2Vec](http://papers.nips.cc/paper/5021-distributed-representations-of-words-and-phrases-and-their-compositionality.pdf) is a popular natural language processing technique that projects words into a vector space that provides information about their context. It can then be used to complete such phrases as, for example, \"woman is to queen as man is to X?\" The answer obviously being \"king\" -- in vector space, the separation of \"woman\" and \"queen\" is similar to the offset between \"man\" and \"king\". "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hashtag_list_df = (data.select('created_at', 'entities.hashtags')\n",
    "                .filter(func.size('hashtags') > 0)\n",
    "                .select(hash_text_udf('hashtags').alias('hashtags'), 'created_at'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "hashtag_list_df.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import Word2Vec"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "w2v = Word2Vec(numPartitions=20, inputCol='hashtags', outputCol='model')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%time model = w2v.fit(hashtag_list_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "model.findSynonyms('ferguson', 10).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "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.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}