work/2-hacker-news.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction\n",
    "**Goal**: This notebook is meant to give a more thorough example of how to write code in a notebook. We will create a basic scaffolding for our future notebooks in this demo.\n",
    "\n",
    "[hacker news](https://news.ycombinator.com/) is a site full of technical knowledge, often considered a one-stop-shop for trending news. However, there might be some benefit to looking at relations to current trending items. Or potentially some useful links can be found in the comments. This notebook will be used to explore these relations and expose them in a way to be consumed by outside users. Specifically, we will look at the following questions:\n",
    "\n",
    "* What are the links in the comments?\n",
    "* What words are mentioned the most in the comments?\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Add Dependencies\n",
    "To get started we will need to add some additional libraries. These libraries will be added locally to this kernel, as well as our Spark cluster.\n",
    "* __jsoup__ - Used to strip out any links or raw text found within html snippets.\n",
    "* __hackernew4s__ - Used to get the top items from hacker news, the comments on these items, and information about the users."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "%adddeps org.jsoup jsoup 1.9.2 --transitive\n",
    "%adddeps com.github.seratch hackernews4s_2.10 0.6.0 --transitive"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Getting Hacker News Articles"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We need to get a reference to the SQL context created by Toree."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val sqlC = sqlContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now import all of the classes we need to create our application."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import hackernews4s.v0._\n",
    "import sqlC.implicits._\n",
    "import org.apache.spark.sql.functions._\n",
    "import org.apache.spark.sql.Row\n",
    "import org.jsoup.Jsoup\n",
    "import org.jsoup.nodes.Document\n",
    "import org.jsoup.nodes.Element\n",
    "import scala.collection.JavaConversions._\n",
    "import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}\n",
    "import org.apache.spark.ml.feature.StopWordsRemover\n",
    "import org.apache.spark.ml.feature.{CountVectorizer, CountVectorizerModel}\n",
    "import org.apache.spark.sql.DataFrame\n",
    "import org.apache.spark.rdd.RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "case class Comment(story: Long, itemId: Long, text: String)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "// A function to transform an item into a tuple of that item and a list of comments on that item\n",
    "val getComments: (Item) => Seq[Comment] = (story: Item) => {\n",
    "    def _getComments:  (Item) => Seq[Comment] = (item: Item) => {\n",
    "        val commentIds = item.commentIds\n",
    "        if(commentIds.size == 0){\n",
    "            Seq(Comment(story.id.id, item.id.id, item.text))\n",
    "        } else {\n",
    "            val comments: Seq[Comment] = commentIds.flatMap((itemId: ItemId) => { \n",
    "                _getComments(HackerNews.getItem(itemId).get)\n",
    "            })\n",
    "            if(\"Story\".equals(item.itemType.toString)){\n",
    "                comments\n",
    "            } else {\n",
    "                Comment(story.id.id, item.id.id, item.text) +: comments\n",
    "            }\n",
    "            \n",
    "        }   \n",
    "    }\n",
    "    \n",
    "    _getComments(story)\n",
    "}\n",
    "\n",
    "val getItemText: (Comment) => String = (comment: Comment) => {\n",
    "    Jsoup.parse(comment.text).text()\n",
    "}\n",
    "val getItemLinks: (Comment) => Seq[String] = (comment: Comment) => {\n",
    "    val aTags: List[Element] = Jsoup.parse(comment.text).select(\"a\").toList\n",
    "    aTags.map((link: Element) => {\n",
    "        link.attr(\"href\")\n",
    "    })\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This function will take in a story id as an argument and will return a Spark RDD of all the comments. This allows us to parallelize our work in the Spark Cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def getStoryComments(storyId: Int) = {\n",
    "    val story = Seq(HackerNews.getItem(ItemId(storyId)).get)\n",
    "    sc.parallelize(story).flatMap((item: Item) => {\n",
    "        getComments(item)\n",
    "    })\n",
    "}   "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This function will take in a comments RDD and will return a new Spark RDD with all of the links for the comments."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def getCommentLinks(comments: RDD[Comment]) = {\n",
    "    comments.flatMap((comment:Comment) => {\n",
    "        getItemLinks(comment)\n",
    "    })\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**`tokenizer`** and **`remover`** are objects from the Apache Spark ML API. They will be used to tokenize the comments and filter out words we do not want to count.\n",
    "\n",
    "**NOTE:** This is an example of a core Spark API being exposed through Toree. Another third party library, like Apache System ML, could be plugged in at this point."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val tokenizer = new Tokenizer().setInputCol(\"_1\").setOutputCol(\"words\")\n",
    "val remover = new StopWordsRemover().setInputCol(\"words\").setOutputCol(\"filteredWords\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This function will take in a comments RDD and will return a new Spark RDD with words and the number of times they appear in the comments."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def getCommentWordCounts(comments: RDD[Comment]) = {\n",
    "    val textDF = comments.map((comment:Comment) => {\n",
    "        getItemText(comment)\n",
    "    }).toDF\n",
    "    val tokenizedComments = tokenizer.transform(textDF)\n",
    "    val filteredWordCountsDF = remover.transform(tokenizedComments)\n",
    "    val terms = filteredWordCountsDF.flatMap((row: Row) =>{\n",
    "        row.getSeq[String](2)\n",
    "    })\n",
    "    val wordCounts = terms.map((word: String) => {\n",
    "        (word, 1)\n",
    "    }).reduceByKey(_+_)\n",
    "    wordCounts\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def getStoryInfo(storyId: Int) = {\n",
    "    val commentsRDD = getStoryComments(storyId)    \n",
    "    val comments = commentsRDD.collect()\n",
    "    val links = getCommentLinks(commentsRDD).collect()\n",
    "    val counts = getCommentWordCounts(commentsRDD).sortBy((wordCount: (String, Int)) => {\n",
    "        wordCount._2\n",
    "    }, ascending=false).take(50)\n",
    "    (comments, counts, links)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Testing\n",
    "We can test our functions out by calling them and inspecting the output value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val storyInfo = getStoryInfo(12476597)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "storyInfo._1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "storyInfo._2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "storyInfo._3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Apache Toree - Scala",
   "language": "scala",
   "name": "apache_toree_scala"
  },
  "language_info": {
   "name": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}