work/3-hacker-news-rest.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RESTful Service\n",
    "**Goal:** The goal of this notebook is to show how we can use Apache Toree for other purposes than exploratory reasons. This example will transform our previous notebook into a backend service we can invoke from the browser or other applications.\n",
    "\n",
    "This notebook exposes our hacker news data explorations as RESTful service. This is done externally through a Jupyter extension called the [Jupyter Kernel Gateway](https://github.com/jupyter-incubator/dashboards). If you are running this demo with `docker-compose` you should be able to access the endpoints at:\n",
    "\n",
    "* http://localhost:9999/story/12476597/comments\n",
    "* http://localhost:9999/story/12476597/links\n",
    "* http://localhost:9999/story/12476597/words\n",
    "\n",
    "You can substitute the id for any valid story in case you would like the verify the demo :D."
   ]
  },
  {
   "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": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val sqlC = sqlContext"
   ]
  },
  {
   "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.{StopWordsRemover, Tokenizer}\n",
    "import org.apache.spark.sql.DataFrame\n",
    "import org.apache.spark.rdd.RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The code in the next cell is the same from the previous notebook. The interesting bits are going to be after this cell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "case class Comment(story: Long, id: Long, text: String)\n",
    "\n",
    "val tokenizer = new Tokenizer().setInputCol(\"_1\").setOutputCol(\"words\")\n",
    "val remover = new StopWordsRemover().setInputCol(\"words\").setOutputCol(\"filteredWords\")\n",
    "\n",
    "// 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",
    "}\n",
    "\n",
    "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",
    "}   \n",
    "\n",
    "def getCommentLinks(comments: RDD[Comment]) = {\n",
    "    comments.flatMap((comment:Comment) => {\n",
    "        getItemLinks(comment)\n",
    "    })\n",
    "}\n",
    "\n",
    "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": "markdown",
   "metadata": {},
   "source": [
    "This JsonHelper object is what we will use to serialize the results from our previous code. This class will output our data structures as JSON, allowing for other applications and users to consume the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "object JsonHelper extends Serializable {\n",
    "    import play.api.libs.json._\n",
    "    \n",
    "    implicit val commentWrites = new Writes[Comment] {\n",
    "        def writes(comment: Comment) = Json.obj(\n",
    "        \"story\" -> comment.story,\n",
    "        \"id\" -> comment.id,\n",
    "        \"text\" -> comment.text\n",
    "        )\n",
    "    }\n",
    "    \n",
    "    implicit val tupleWrites = new Writes[(String, Int)] {\n",
    "        def writes(tuple: (String, Int)) = Json.obj(\n",
    "            tuple._1 -> tuple._2\n",
    "        )\n",
    "    }\n",
    "    \n",
    "    def jsonComments(REQUEST: String) = {\n",
    "        val req = Json.parse(REQUEST)\n",
    "        val storyId = (req \\ \"path\" \\ \"story_id\").as[String].toInt\n",
    "        val comments = getStoryComments(storyId).collect()\n",
    "        Json.toJson(comments)\n",
    "    }\n",
    "    \n",
    "    def jsonLinks(REQUEST: String) = {\n",
    "        val req = Json.parse(REQUEST)\n",
    "        val storyId = (req \\ \"path\" \\ \"story_id\").as[String].toInt\n",
    "        val links = getCommentLinks(getStoryComments(storyId)).collect()\n",
    "        Json.toJson(links)\n",
    "    }\n",
    "    \n",
    "    def jsonWords(REQUEST: String) = {\n",
    "        val req = Json.parse(REQUEST)\n",
    "        val storyId = (req \\ \"path\" \\ \"story_id\").as[String].toInt\n",
    "        val words = getCommentWordCounts(getStoryComments(storyId)).sortBy((wordCount: (String, Int)) => {\n",
    "        wordCount._2\n",
    "    }, ascending=false).take(50)\n",
    "        Json.toJson(words)\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Below we have annotated our cells with comments to register them as RESTful endpoints. The three endpoints take a __`story_id`__ as a path parameter. This is parsed out in our **`JSONHelper`** class above. The three endpoints are:\n",
    "\n",
    "* GET /story/:story_id/comments: Gets all of the comments for a story\n",
    "* GET /story/:story_id/links: Gets all of the links in the comments of a story\n",
    "* GET /story/:story_id/words: Gets all of the word counts for the comments of a story.\n",
    "\n",
    "The Jupyter Kernel Gateway automatically injects a JSON string stored in the variable **`REQUEST`**. This is done before each cell is invoked.\n",
    "\n",
    "If we wanted to test this code locally we could create **`REQUEST`** inline in a code cell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "// GET /story/:story_id/comments\n",
    "println(JsonHelper.jsonComments(REQUEST))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "// GET /story/:story_id/links\n",
    "println(JsonHelper.jsonLinks(REQUEST))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "// GET /story/:story_id/words\n",
    "println(JsonHelper.jsonWords(REQUEST))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Apache Toree - Scala",
   "language": "scala",
   "name": "apache_toree_scala"
  },
  "language_info": {
   "name": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}