tutorial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Spark word count tutorial\n",
    "This Apache-Toree Jupyter notebook demonstrates performing word count with [Spark](http://spark.apache.org/docs/latest/index.html) (and [Scala](http://www.scala-lang.org/)). The content is adapted slightly from the [Spark getting started guide](http://spark.apache.org/docs/latest/quick-start.html). Users can execute the same commands in the Spark REPL, which is launched by running in bash \n",
    "```bash\n",
    "$SPARK_HOME/bin/spark-shell\n",
    "```\n",
    "where the environment variable `$SPARK_HOME` points to an installation of Spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<iframe width=\"100%\" height=\"550\" \n",
       "src=\"http://spark.apache.org/docs/latest/quick-start.html\"></iframe>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%html\n",
    "<iframe width=\"100%\" height=\"550\" \n",
    "src=\"http://spark.apache.org/docs/latest/quick-start.html\"></iframe>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Read in a text file using the SparkContext\n",
    "The SparkContext ```sc``` is automatically instantiated by the Apache-Toree notebook.  In standalone Spark applications, the SparkContext must be manually instantiated. The value ```textFile``` becomes an RDD (Resilient Distributed Dataset)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [],
   "source": [
    "val textFile = sc.textFile(\"spark_read_me.txt\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "source": [
    "The RDD cannot be view directly in the notebook (in practice it will be distributed across many nodes!!). Thus, in order to view all the data, we have to gather the data at a single node using ```collect```. A summary of RDD functions can be found [here](http://spark.apache.org/docs/latest/programming-guide.html#rdd-operations)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "# Apache Spark\n",
      "\n",
      "Spark is a fast and general cluster computing system for Big Data. It provides\n",
      "high-level APIs in Scala, Java, Python, and R, and an optimized engine that\n",
      "supports general computation graphs for data analysis. It also supports a\n",
      "rich set of higher-level tools including Spark SQL for SQL and DataFrames,\n",
      "MLlib for machine learning, GraphX for graph processing,\n",
      "and Spark Streaming for stream processing.\n",
      "\n",
      "<http://spark.apache.org/>\n",
      "\n",
      "\n",
      "## Online Documentation\n",
      "\n",
      "You can find the latest Spark documentation, including a programming\n",
      "guide, on the [project web page](http://spark.apache.org/documentation.html)\n",
      "and [project wiki](https://cwiki.apache.org/confluence/display/SPARK).\n",
      "This README file only contains basic setup instructions.\n",
      "\n",
      "## Building Spark\n",
      "\n",
      "Spark is built using [Apache Maven](http://maven.apache.org/).\n",
      "To build Spark and its example programs, run:\n",
      "\n",
      "    build/mvn -DskipTests clean package\n",
      "\n",
      "(You do not need to do this if you downloaded a pre-built package.)\n",
      "More detailed documentation is available from the project site, at\n",
      "[\"Building Spark\"](http://spark.apache.org/docs/latest/building-spark.html).\n",
      "\n",
      "## Interactive Scala Shell\n",
      "\n",
      "The easiest way to start using Spark is through the Scala shell:\n",
      "\n",
      "    ./bin/spark-shell\n",
      "\n",
      "Try the following command, which should return 1000:\n",
      "\n",
      "    scala> sc.parallelize(1 to 1000).count()\n",
      "\n",
      "## Interactive Python Shell\n",
      "\n",
      "Alternatively, if you prefer Python, you can use the Python shell:\n",
      "\n",
      "    ./bin/pyspark\n",
      "\n",
      "And run the following command, which should also return 1000:\n",
      "\n",
      "    >>> sc.parallelize(range(1000)).count()\n",
      "\n",
      "## Example Programs\n",
      "\n",
      "Spark also comes with several sample programs in the `examples` directory.\n",
      "To run one of them, use `./bin/run-example <class> [params]`. For example:\n",
      "\n",
      "    ./bin/run-example SparkPi\n",
      "\n",
      "will run the Pi example locally.\n",
      "\n",
      "You can set the MASTER environment variable when running examples to submit\n",
      "examples to a cluster. This can be a mesos:// or spark:// URL,\n",
      "\"yarn\" to run on YARN, and \"local\" to run\n",
      "locally with one thread, or \"local[N]\" to run locally with N threads. You\n",
      "can also use an abbreviated class name if the class is in the `examples`\n",
      "package. For instance:\n",
      "\n",
      "    MASTER=spark://host:7077 ./bin/run-example SparkPi\n",
      "\n",
      "Many of the example programs print usage help if no params are given.\n",
      "\n",
      "## Running Tests\n",
      "\n",
      "Testing first requires [building Spark](#building-spark). Once Spark is built, tests\n",
      "can be run using:\n",
      "\n",
      "    ./dev/run-tests\n",
      "\n",
      "Please see the guidance on how to\n",
      "[run tests for a module, or individual tests](https://cwiki.apache.org/confluence/display/SPARK/Useful+Developer+Tools).\n",
      "\n",
      "## A Note About Hadoop Versions\n",
      "\n",
      "Spark uses the Hadoop core library to talk to HDFS and other Hadoop-supported\n",
      "storage systems. Because the protocols have changed in different versions of\n",
      "Hadoop, you must build Spark against the same version that your cluster runs.\n",
      "\n",
      "Please refer to the build documentation at\n",
      "[\"Specifying the Hadoop Version\"](http://spark.apache.org/docs/latest/building-spark.html#specifying-the-hadoop-version)\n",
      "for detailed guidance on building for a particular distribution of Hadoop, including\n",
      "building for particular Hive and Hive Thriftserver distributions.\n",
      "\n",
      "## Configuration\n",
      "\n",
      "Please refer to the [Configuration Guide](http://spark.apache.org/docs/latest/configuration.html)\n",
      "in the online documentation for an overview on how to configure Spark.\n"
     ]
    }
   ],
   "source": [
    "(textFile collect) foreach println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Filter lines containing *Spark*\n",
    "The RDD (and Scala collections) support filtering."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "# Apache Spark\n",
      "Spark is a fast and general cluster computing system for Big Data. It provides\n",
      "rich set of higher-level tools including Spark SQL for SQL and DataFrames,\n",
      "and Spark Streaming for stream processing.\n",
      "You can find the latest Spark documentation, including a programming\n",
      "## Building Spark\n",
      "Spark is built using [Apache Maven](http://maven.apache.org/).\n",
      "To build Spark and its example programs, run:\n",
      "[\"Building Spark\"](http://spark.apache.org/docs/latest/building-spark.html).\n",
      "The easiest way to start using Spark is through the Scala shell:\n",
      "Spark also comes with several sample programs in the `examples` directory.\n",
      "    ./bin/run-example SparkPi\n",
      "    MASTER=spark://host:7077 ./bin/run-example SparkPi\n",
      "Testing first requires [building Spark](#building-spark). Once Spark is built, tests\n",
      "Spark uses the Hadoop core library to talk to HDFS and other Hadoop-supported\n",
      "Hadoop, you must build Spark against the same version that your cluster runs.\n",
      "in the online documentation for an overview on how to configure Spark.\n"
     ]
    }
   ],
   "source": [
    "val linesWithSpark = textFile.filter(line => line.contains(\"Spark\"))\n",
    "(linesWithSpark collect) foreach println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Map lines from String to Array[String]\n",
    "The RDD (and Scala collections) support mapping. For example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(a, line, with, Spark)"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val foo = \"a line with Spark\"\n",
    "foo.split(\" \") "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Scala Bonus!\n",
    "The dot operator can be omitted in Scala, so that splitting operation can be written as:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(a, line, with, Spark)"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "foo split (\" \")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "source": [
    "Here, we've split the line (of type `String`) into an `Array` of `String`s by splitting the original line on whitespace (\" \"). We'll consider each element of the "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3\n",
      "14\n",
      "12\n",
      "6\n",
      "10\n",
      "3\n",
      "6\n",
      "8\n",
      "2\n",
      "12\n",
      "11\n",
      "6\n",
      "7\n",
      "10\n",
      "13\n",
      "13\n",
      "12\n"
     ]
    }
   ],
   "source": [
    "val arraysWithSpark = linesWithSpark.map(line => line split(\" \"))\n",
    "val wordsPerLine = arraysWithSpark map (a => a.size)\n",
    "(wordsPerLine collect) foreach println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Counting words per line\n",
    "Spark and Scala allow for chaining operations together. Thus, we can just write:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3\n",
      "14\n",
      "12\n",
      "6\n",
      "10\n",
      "3\n",
      "6\n",
      "8\n",
      "2\n",
      "12\n",
      "11\n",
      "6\n",
      "7\n",
      "10\n",
      "13\n",
      "13\n",
      "12\n"
     ]
    }
   ],
   "source": [
    "val wordsPerLine = linesWithSpark.map(line => line split(\" \")).\n",
    "  map(a => a.size)\n",
    "(wordsPerLine collect) foreach println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Or equivalently"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3\n",
      "14\n",
      "12\n",
      "6\n",
      "10\n",
      "3\n",
      "6\n",
      "8\n",
      "2\n",
      "12\n",
      "11\n",
      "6\n",
      "7\n",
      "10\n",
      "13\n",
      "13\n",
      "12\n"
     ]
    }
   ],
   "source": [
    "val wordsPerLine = linesWithSpark map(_ split \" \") map(_ size)\n",
    "(wordsPerLine collect) foreach println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "source": [
    "Here, Scala is smart enough to understand that the underscore implies that the map function is taking a single argument, thus avoiding the need for the ```foo => foo split(\" \")``` pattern. (Scala Bonus! this pattern defines an anonymous function.)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Counting total *Spark* occurrences\n",
    "If we wanted to count total occurrence of *Spark* we could filter our `Array`s from the previous step to keep only those words that match *Spark*. Since we don't actually care about which line contains occurrences of *Spark* but rather how many *Spark*s are in our entire document, we can consider all the individual words at once. Spark provides a mechanism to do this called `flatMap`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r",
      "#\n",
      "Apache\n",
      "Spark\n",
      "Spark\n",
      "is\n",
      "a\n",
      "fast\n",
      "and\n",
      "general\n",
      "cluster\n",
      "computing\n",
      "system\n",
      "for\n",
      "Big\n",
      "Data.\n",
      "It\n",
      "provides\n",
      "rich\n",
      "set\n",
      "of\n",
      "higher-level\n",
      "tools\n",
      "including\n",
      "Spark\n",
      "SQL\n",
      "for\n",
      "SQL\n",
      "and\n",
      "DataFrames,\n",
      "and\n",
      "Spark\n",
      "Streaming\n",
      "for\n",
      "stream\n",
      "processing.\n",
      "You\n",
      "can\n",
      "find\n",
      "the\n",
      "latest\n",
      "Spark\n",
      "documentation,\n",
      "including\n",
      "a\n",
      "programming\n",
      "##\n",
      "Building\n",
      "Spark\n",
      "Spark\n",
      "is\n",
      "built\n",
      "using\n",
      "[Apache\n",
      "Maven](http://maven.apache.org/).\n",
      "To\n",
      "build\n",
      "Spark\n",
      "and\n",
      "its\n",
      "example\n",
      "programs,\n",
      "run:\n",
      "[\"Building\n",
      "Spark\"](http://spark.apache.org/docs/latest/building-spark.html).\n",
      "The\n",
      "easiest\n",
      "way\n",
      "to\n",
      "start\n",
      "using\n",
      "Spark\n",
      "is\n",
      "through\n",
      "the\n",
      "Scala\n",
      "shell:\n",
      "Spark\n",
      "also\n",
      "comes\n",
      "with\n",
      "several\n",
      "sample\n",
      "programs\n",
      "in\n",
      "the\n",
      "`examples`\n",
      "directory.\n",
      "\n",
      "\n",
      "\n",
      "\n",
      "./bin/run-example\n",
      "SparkPi\n",
      "\n",
      "\n",
      "\n",
      "\n",
      "MASTER=spark://host:7077\n",
      "./bin/run-example\n",
      "SparkPi\n",
      "Testing\n",
      "first\n",
      "requires\n",
      "[building\n",
      "Spark](#building-spark).\n",
      "Once\n",
      "Spark\n",
      "is\n",
      "built,\n",
      "tests\n",
      "Spark\n",
      "uses\n",
      "the\n",
      "Hadoop\n",
      "core\n",
      "library\n",
      "to\n",
      "talk\n",
      "to\n",
      "HDFS\n",
      "and\n",
      "other\n",
      "Hadoop-supported\n",
      "Hadoop,\n",
      "you\n",
      "must\n",
      "build\n",
      "Spark\n",
      "against\n",
      "the\n",
      "same\n",
      "version\n",
      "that\n",
      "your\n",
      "cluster\n",
      "runs.\n",
      "in\n",
      "the\n",
      "online\n",
      "documentation\n",
      "for\n",
      "an\n",
      "overview\n",
      "on\n",
      "how\n",
      "to\n",
      "configure\n",
      "Spark.\n"
     ]
    }
   ],
   "source": [
    "val allSparkWords = linesWithSpark flatMap (line => line split \" \")\n",
    "(allSparkWords collect) foreach println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "To get the Spark occurrences, simply filter:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n",
      "Spark\n"
     ]
    }
   ],
   "source": [
    "val allSparkWords = linesWithSpark flatMap (_ split \"\\\\s+\") filter (_ == \"Spark\")\n",
    "(allSparkWords collect) foreach println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "source": [
    "Scala Bonus! `\"\\\\s+\"` is a regex that matches one or more consecutive whitespace characters."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Spark supports *actions*\n",
    "Actions return a value to the driver program after running a computation on the dataset. A common action is `count`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "13\n"
     ]
    }
   ],
   "source": [
    "val totalSparks = allSparkWords count ()\n",
    "println(totalSparks)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Counting occurrences for each word\n",
    "This is the classic MapReduce (as popularized by [Hadoop](http://hadoop.apache.org/)) example. We split each line into words and before and map each word into a word value *pair* (aka *tuple*). The first element of the pair is the *key* which serves as identifier. The second element of the pair is the value `1` which signifies that each word has occurred one time. Since each word is not unique, we need to group them together and count the occurrences per group. When we perform an *action* on an RDD, all the pairs with identical keys are sent to the same node and then we can aggregate these together. This is precisely what the `reduceByKey` function does:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(package,1)\n",
      "(this,1)\n",
      "(Version\"](http://spark.apache.org/docs/latest/building-spark.html#specifying-the-hadoop-version),1)\n",
      "(Because,1)\n",
      "(Python,2)\n",
      "(cluster.,1)\n",
      "(its,1)\n",
      "([run,1)\n",
      "(general,2)\n",
      "(have,1)\n",
      "(pre-built,1)\n",
      "(YARN,,1)\n",
      "(locally,2)\n",
      "(changed,1)\n",
      "(locally.,1)\n",
      "(sc.parallelize(1,1)\n",
      "(only,1)\n",
      "(several,1)\n",
      "(This,2)\n",
      "(basic,1)\n"
     ]
    }
   ],
   "source": [
    "val wordFrequencies = textFile flatMap (_ split (\"\\\\s+\") map (word => (word, 1))) reduceByKey (_ + _)\n",
    "(wordFrequencies take 20) foreach println"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "-"
    }
   },
   "source": [
    "The argument to `reduceByKey` is a function describing how to combine values (which must have the same type, otherwise see `aggregateByKey`). To print the output, we've usen the `take` function to take the first 20 results. We can also use the aptly named `takeOrdered` function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(,43)\n",
      "(the,21)\n",
      "(to,14)\n",
      "(Spark,13)\n",
      "(for,11)\n",
      "(and,10)\n",
      "(a,8)\n",
      "(##,8)\n",
      "(run,7)\n",
      "(is,6)\n",
      "(can,6)\n",
      "(on,5)\n",
      "(in,5)\n",
      "(of,5)\n",
      "(also,4)\n",
      "(if,4)\n",
      "(you,4)\n",
      "(Please,3)\n",
      "(use,3)\n",
      "(documentation,3)\n",
      "(build,3)\n",
      "(with,3)\n",
      "(Hadoop,3)\n",
      "(example,3)\n",
      "(or,3)\n",
      "(You,3)\n",
      "(including,3)\n",
      "(an,3)\n",
      "(Hadoop,,2)\n",
      "(command,,2)\n",
      "(This,2)\n",
      "(cluster,2)\n",
      "(Scala,2)\n",
      "(./bin/run-example,2)\n",
      "(Hive,2)\n",
      "([project,2)\n",
      "(general,2)\n",
      "(be,2)\n",
      "(It,2)\n",
      "(1000:,2)\n",
      "(following,2)\n",
      "(For,2)\n",
      "(SparkPi,2)\n",
      "(set,2)\n",
      "(To,2)\n",
      "(should,2)\n",
      "(do,2)\n",
      "(tests,2)\n",
      "(which,2)\n",
      "(class,2)\n"
     ]
    }
   ],
   "source": [
    "(wordFrequencies.takeOrdered(50)(Ordering[Int].reverse.on(_._2))) foreach println"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "kernelspec": {
   "display_name": "Apache Toree - Scala",
   "language": "scala",
   "name": "apache_toree_scala"
  },
  "language_info": {
   "name": "scala",
   "version": "2.10.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}