Chapter3 - Programming with RDDs - scala.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h1 id=\"tocheading\">Table of Contents</h1>\n",
    "<div id=\"toc\"></div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "application/javascript": [
       "$.getScript('https://kmahelona.github.io/ipython_notebook_goodies/ipython_notebook_toc.js')"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%javascript\n",
    "$.getScript('https://kmahelona.github.io/ipython_notebook_goodies/ipython_notebook_toc.js')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Chapter 3: Programming with RDDs\n",
    "## RDD Basics\n",
    "* RDD: An immutable distributed collection of objects. Each split into multiple partitions, which pay be computed on different nodes of the cluster\n",
    "* Created in 2 ways:\n",
    "    1. Loading an external dataset\n",
    "    2. Distributing a collection of objects (e.g. list or set) in their driver program\n",
    "* RDDs offer 2 types of operations:\n",
    "    1. Transformations: construct a new RDD from a previous one\n",
    "    2. Actions: Compute a result based on an RDD, either return it to the driver program or save it to an external storage system (e.g. HDFS)\n",
    "* Difference: Although you can define new RDDs any time, Spark only computes them in a <i> lazy</i> fashion, the first time they are used in an action\n",
    "* RDDs are by default recomputed each time you run an action on them\n",
    "    - If you would like to reuse RDD in multiple actions, you can ask Spark to <i>persist</i> it using <code>RDD.persist()</code>. to store the RDD contents in memory (partitioned across the machines in your cluster) and reuse them in future actions\n",
    "    - It does not persist by default because in the context of big data if you do not use the RDD, there is no reason to waste storage space when Spark could instead stream through the data once and just compute the result\n",
    "\n",
    "### Example: Filter data that matches a predicate\n",
    "Create a new RDD holding just the strings that contain \"Python\":\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of lines: 95\n",
      "First line: # Apache Spark\n",
      "First line with python in it: high-level APIs in Scala, Java, Python, and R, and an optimized engine that\n"
     ]
    }
   ],
   "source": [
    "val lines = sc.textFile(\"README_spark.md\") \n",
    "println(\"number of lines: \" + lines.count())\n",
    "println(\"First line: \" + lines.first())\n",
    "val pythonLines = lines.filter(line => line.contains(\"Python\"))\n",
    "println(\"First line with python in it: \" + pythonLines.first())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Benefits of lazy evaluation:\n",
    "* Computations only executed the first time they are in exected in an action\n",
    "* In this example, if Spark were to load and store all the lines in the file at:\n",
    "        lines = sc.textFile(...)\n",
    "  it would waste a lot of storage space, given that we then immediately filter out many lines. \n",
    "* Instead, once Spark sees the whole chain of transformations, it can compute just the data needed for its result\n",
    "    - In fact, for the </code>first()</code> action, Spark only scans the file until it finds the first matching line; <b>it does not read the whole file</b>\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Example: Practical Use of Persist\n",
    "\n",
    "In practice, you will often use <code>persist</code> to load a subset of your data into memory and query it repeatedly. For example, if we knew that we wanted to compute multiple results from the README lines that contain \"Python\", we could write:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Map(3 -> MapPartitionsRDD[3] at filter at <console>:22)\n",
      "Map()\n"
     ]
    }
   ],
   "source": [
    "pythonLines.persist()\n",
    "println(sc.getPersistentRDDs)\n",
    "pythonLines.unpersist()\n",
    "println(sc.getPersistentRDDs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating RDDs\n",
    "\n",
    "Two options:\n",
    "1. Load external dataset\n",
    "2. Parallelize a collection in your driver program\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "external dataset: # Apache Spark\n",
      "parallelize your own: Pandas\n"
     ]
    }
   ],
   "source": [
    "// load external dataset (1):\n",
    "val lines = sc.textFile(\"README_spark.md\")\n",
    "println(\"external dataset: \" + lines.first())\n",
    "\n",
    "// parallelize a collection in your own driver program (2):\n",
    "val lines = sc.parallelize(List(\"Pandas\", \"I Like Pandas\"))\n",
    "println(\"parallelize your own: \" + lines.first())\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transformations Filter Example\n",
    "* <code>filter</code> does not mutate the existing <code>inputRDD</code> \n",
    "* Instead, it returns a pointer to an entirely new RDD\n",
    "* <code>inputRDD</code> can be used again to search for lines with the word \"warning\" in them\n",
    "* We can then use another transformation - <code>union</code> - to print out the number of lines that contain either \"error\" or \"warning\"\n",
    "    - <code>union</code> is different than filter in that it operates on two RDDs opposed to one\n",
    "    - Transformations can actually operate on any number of input RDDs\n",
    "* As you derive new RDDs from each other using transformation, Spark keeps track of the set of dependencies between different RDDs called <code>lineage graph</code>\n",
    "    - It uses this information to copute each RDD on demand and to recover los data if part of a persistent RDD is lost\n",
    "    \n",
    "<img src =\"figures/lineage_graph.png\">"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "inputRDD: \n",
      "first warning\n",
      "cats\n",
      "second warning\n",
      "cats\n",
      "first warning and error\n",
      "cats\n",
      "\n",
      "badLinesRDD: \n",
      "first warning and error\n",
      "first warning\n",
      "second warning\n",
      "first warning and error\n"
     ]
    }
   ],
   "source": [
    "val inputRDD = sc.textFile(\"log.txt\")\n",
    "val errorsRDD = inputRDD.filter(line => line.contains(\"error\"))\n",
    "val warningsRDD = inputRDD.filter(line => line.contains(\"warning\"))\n",
    "val badLinesRDD = errorsRDD.union(warningsRDD)\n",
    "\n",
    "// print error lines\n",
    "println(\"inputRDD: \")\n",
    "inputRDD.collect().foreach(println)\n",
    "\n",
    "println(\"\\nbadLinesRDD: \")\n",
    "badLinesRDD.collect().foreach(println)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Actions Example\n",
    "* Actions <i>do something</i> with our dataset\n",
    "* <code>count()</code>: returns the count as a number\n",
    "* <code>take()</code>: collects a number of elements from the RDD\n",
    "* <code>collect()</code>: Retrieves the entire RDD; useful if your program filters RDDs down to a very small size and you'd like to deal with it locally. \n",
    "    - <b> Your entire dataset must fit in memory on a single machine to use <code>collect()</code></b>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### MapReduce vs. Spark Exerpt \n",
    "<i>Spark uses lazy evaluation to reduce the number of passes it has to take over our data\n",
    "by grouping operations together. In MapReduce systems like Hadoop, developers often\n",
    "have to spend a lot time considering how to group together operations to minimize the\n",
    "number of MapReduce passes. In Spark, there is no substantial benefit to writing a single\n",
    "complex map instead of chaining together many simple operations. Thus, users are free\n",
    "to organize their program into smaller, more manageable operations</i>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Passing Functions to Spark\n",
    "* When you pass a function that is the member of an object (e.g. <code>self.field</code>), Spark sends the <i> entire</i> object to worker nodes, which can be much larger than the bit of information you need\n",
    "* This can also cause your program to fail, if your class contains objects that Python can't figure out how to pickle\n",
    "\n",
    "<b> Don't do this:</b>\n",
    "\n",
    "    class Word Functions(object):\n",
    "        ...\n",
    "        def getMatchesNoReference(self,rdd):\n",
    "            return rdd.filter(lambda x: self.query in x)\n",
    "\n",
    "<b> Do this:</b>\n",
    "\n",
    "    class Word Functions(object):\n",
    "        ...\n",
    "        def getMatchesNoReference(self,rdd):\n",
    "            query = self.query\n",
    "            return rdd.filter(lambda x: query in x)\n",
    "            \n",
    "            \n",
    "If <code>NotSerializableException</code> occurs in Scala, a reference to a method or field in a nonserializable class is usually the problem. Note that passing in local serializable variables or functions that are members of a top-level object is always safe. \n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Common Transformations and Actions\n",
    "* The two most <b>common transformations</b> you will likely be using are <code>map()</code> and <code>filter()</code>\n",
    "    - <code>map()</code> takes in a function and applies it to each element in the RDD with the result of the function being a new value of each element\n",
    "        - The return type does not have to be the same as its input type\n",
    "    - <code>filter()</code> takes in a function and returns an RDD that only has elements that pass the <code>filter()</code> function\n",
    "* <code>flatmap()</code> produces multiple output elements for each input element \n",
    "    - Called individually for each element in our input RDD\n",
    "    - Instead of returning a single element, we return an iterator with our return values\n",
    "    - Rather than producing an RDD of iterators, we return an RDD that consists of the elements from all the iterators\n",
    "    \n",
    "   \n",
    "### Pseudo Set Operators \n",
    "* RDDs support many of the operations of mathematical sets, such as union, intersection, even when the RDDs themselves are not properly sets \n",
    "* The set property most frequently missing from our RDDs is the uniqueness of elements\n",
    "    - If we only want unique elements we can use <code>RDD.distinct</code> transformation to produce a new RDD with only distinct items\n",
    "    - Note that <code>distinct()</code> is expensive as it requires shuffling all the data over the network to ensure that we only receive one copy of each element \n",
    "* <code>union()</code>: returns an RDD consisting of data from both sources, but if there are duplicates in the input RDDs, the result will also contain duplicates\n",
    "* <code>intersection()</code>: only returns elements in both RDDs and removes all duplicates (including duplicates from a single RDD)\n",
    "    - Performance of <code>intersection</code> is much worse than <code>union</code> since it requires a shuffle over the network to identify the common elements\n",
    "        - In a shuffle operation we have to send the results to different machines rather than processing them locally\n",
    "* <code>subtract(other)</code> takes in another RDD and returns an RDD that only has values present in teh first RDD and not the second RDD \n",
    "* <code>cartesian(other)</code> transformation results in possible pairs of (a,b) where a is the source RDD and b is the other RDD\n",
    "\n",
    "<img src='figures/cartesian_product_figure.png'>\n",
    "\n",
    "### Examples of Transformations\n",
    "\n",
    "#### Map and Flat Map Example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "//map example\n",
    "println(\"Map Example\")\n",
    "val input = sc.parallelize(List(1,2,3,4))\n",
    "val result = input.map(x => x*x)\n",
    "println(\"original input: \" + input.collect().mkString(\",\"))\n",
    "println(\"mapped result: \" + result.collect().mkString(\",\"))\n",
    "\n",
    "\n",
    "//Flat Map Example - splitting up an input string into words\n",
    "println(\"\\nFlat Map Example\")\n",
    "val lines = sc.parallelize(List(\"hello world\",\"hi\"))\n",
    "val words = lines.flatMap(line => line.split(\" \"))\n",
    "println(\"original lines: \" + lines.collect().mkString(\",\"))\n",
    "println(\"mapped results: \" + words.collect().mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Map and Flat Map Example\n",
    "* <code>sample(withReplacement,fraction,[seed])</code>\n",
    "    - <code>withReplacement: {true, false}</code>\n",
    "    - <code> fraction: </code> the probability that each element is chosen\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "original: 1,2,3,3\n",
      "map output: 2,3,4,4\n",
      "sample output: 1,3\n",
      "rdd1: 1,2,3\n",
      "rdd2: 3,4,5\n",
      "union output: 1,2,3,3,4,5\n",
      "intersection output: 3\n",
      "subtraction output: 1,2\n",
      "Cartesian product: (1,3),(1,4),(1,5),(2,3),(2,4),(2,5),(3,3),(3,4),(3,5)\n"
     ]
    }
   ],
   "source": [
    "// map example\n",
    "val rdd = sc.parallelize(List(1,2,3,3))\n",
    "println(\"original: \" + rdd.collect().mkString(\",\"))\n",
    "println(\"map output: \" + rdd.map(x => x+1).collect().mkString(\",\"))\n",
    "\n",
    "// sample\n",
    "println(\"sample output: \" + rdd.sample(false,0.5).collect().mkString(\",\"))\n",
    "\n",
    "\n",
    "val rdd1 = sc.parallelize(List(1,2,3))\n",
    "val rdd2 = sc.parallelize(List(3,4,5))\n",
    "println(\"rdd1: \" + rdd1.collect().mkString(\",\"))\n",
    "println(\"rdd2: \" + rdd2.collect().mkString(\",\"))\n",
    "\n",
    "\n",
    "// union\n",
    "println(\"union output: \" + rdd1.union(rdd2).collect().mkString(\",\"))\n",
    "\n",
    "// intersection \n",
    "println(\"intersection output: \" + rdd1.intersection(rdd2).collect().mkString(\",\"))\n",
    "\n",
    "// subtract\n",
    "println(\"subtraction output: \" + rdd1.subtract(rdd2).collect().mkString(\",\"))\n",
    "println(\"Cartesian product: \" + rdd1.cartesian(rdd2).collect().mkString(\",\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Actions\n",
    "* <code>reduce()</code>: Takes in a function with operates on two elements of the same type of your RDD and returns a new element of the same type \n",
    "    - <code>reduce()</code> is somewhat special. The \"worker function\" for this one must accept two arguments (we've called them x and y here), not just one. The function is called with the first two elements from the list, then with the result of that call and the third element, and so on, until all of the list elements have been handled. This means that our function is called n-1 times if the list contains n elements. The return value of the last call is the result of the reduce() construct. In the above example, it simply adds the arguments, so we get the sum of all elements. \n",
    "* <code>fold()</code>: takes a function with the same signature needed for <code>reduce</code>, but in addition takes a \"zero value\" to be used for the initial call on each partition. \n",
    "    - The zero value you provide should be the identity element for your operation \n",
    "    - That is, applying it multiple times with your function should not change the value (e.g. 0 for +, 1 for *, or an empty list for concatenation)\n",
    "* <code>aggregate()</code>: this function frees us from the constraint of having the return be the same type as the RDD we are working with. <br>We supply:\n",
    "    - An initial value of the type we want to return\n",
    "    - A function to combine the elements from our RDD with the accumulator \n",
    "    - A second function to merge two accumulators, given that each node accumulates its own results locally"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Example of Fold and Reduce"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "reduce output: 10\n",
      "fold output: __2_1_4_3\n"
     ]
    }
   ],
   "source": [
    "println(\"reduce output: \" + sc.parallelize(List(1,2,3,4)).reduce((x,y) => x+y))\n",
    "println(\"fold output: \" +  sc.parallelize(List(\"1\",\"2\",\"3\",\"4\")).fold(\"_\")((x,y) => x+y))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Testing Same Type Output Claim:\n",
    "<i>Both <code>fold()</code> and <code>reduce()</code> require that the return type of our result be the same type as that of the elements in the RDD we are operating over "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0"
      ]
     },
     "execution_count": 130,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//sc.parallelize(List(1,2,3,4)).reduce((x,y) => x.asInstanceOf[Long]/y) // this fails\n",
    "sc.parallelize(List(1,2,3,4)).reduce((x,y) => x/y) // rounded "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Aggregate Example\n",
    "<code>rdd.aggregate((init_val0, init_val1),combiner_function,merger_function)</code>\n",
    "- <code>combiner_function</code>: combine the elements from our RDD with the accumulator\n",
    "- <code>merger_function</code>: merge two accumulators, given that each node accumulates its own results locally\n",
    "\n",
    "<b>Note:</b> The number of partitions affects how the elements are combined as shown in the \"intermediate\" step\n",
    "\n",
    "<b> Use <code>aggregate</code> to compute the mean</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 158,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "results: (10,4)\n",
      "average:2.5\n"
     ]
    }
   ],
   "source": [
    "val partitions = 2\n",
    "val rdd = sc.parallelize(List(1,2,3,4),partitions)\n",
    "val results = rdd.aggregate((0,0))((x,y) => (x._1 + y, x._2+1),(x,y) => (x._1 + y._1,x._2 + y._2))\n",
    "println(\"results: \" + results)\n",
    "println(\"average:\"+  (results._1.asInstanceOf[Double]/results._2.asInstanceOf[Double]))"
   ]
  }
 ],
 "metadata": {
  "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
}