Lectures/15b-Spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SPARK"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import pandas as pd\n",
    "import seaborn as sns\n",
    "pd.set_option('display.width', 500)\n",
    "pd.set_option('display.max_columns', 100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#Table of Contents\n",
    "* [SPARK](#SPARK)\n",
    "* [Spark Context](#Spark-Context)\n",
    "\t* &nbsp;\n",
    "\t\t* [Create A RDD](#Create-A-RDD)\n",
    "\t\t* [Call `collect` on an RDD: Lazy Spark](#Call-collect-on-an-RDD:-Lazy-Spark)\n",
    "\t\t* [Operations on RDDs](#Operations-on-RDDs)\n",
    "\t\t* [Word Examples](#Word-Examples)\n",
    "\t\t* [Key Value Pairs](#Key-Value-Pairs)\n",
    "\t\t* [word count 1](#word-count-1)\n",
    "\t\t* [word count 2:  `reduceByKey()`](#word-count-2:--reduceByKey%28%29)\n",
    "\t\t* [Nested Syntax](#Nested-Syntax)\n",
    "\t\t* [Using Cache](#Using-Cache)\n",
    "\t\t* [Fun with words](#Fun-with-words)\n",
    "\t\t* [DataFrames](#DataFrames)\n",
    "\t\t* [Machine Learning](#Machine-Learning)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "With shameless stealing of some code and text from:\n",
    "\n",
    "- https://github.com/tdhopper/rta-pyspark-presentation/blob/master/slides.ipynb\n",
    "- Databricks and Berkeley Spark MOOC: https://www.edx.org/course/introduction-big-data-apache-spark-uc-berkeleyx-cs100-1x\n",
    "\n",
    "which you should go check out."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Installing (Py)Spark locally\n",
    "For Mac users using Homebrew:\n",
    "\n",
    "```\n",
    "$ brew install apache-spark\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Install Java SDK\n",
    "\n",
    "OR: Use the Vagrant based solution which will be talked about in the lab.\n",
    "\n",
    "```\n",
    "$ IPYTHON=1 pyspark\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "$ $ IPYTHON_OPTS=\"notebook --matplotlib inline\" pyspark\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "➜  ~  /anaconda/bin/pip install findspark\n",
    "You are using pip version 7.0.3, however version 7.1.2 is available.\n",
    "You should consider upgrading via the 'pip install --upgrade pip' command.\n",
    "Collecting findspark\n",
    "  Downloading findspark-0.0.5-py2.py3-none-any.whl\n",
    "Installing collected packages: findspark\n",
    "Successfully installed findspark-0.0.5\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "#Spark Context\n",
    "\n",
    "You can also use it directly from the notebook interface on the mac if you installed `apache-spark` using `brew` and also installed `findspark` above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "import pyspark\n",
    "sc = pyspark.SparkContext()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It also output's a bunch of stuff on my terminal. This is because the entire java context is started up.\n",
    "\n",
    "```Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
    "15/10/21 14:46:15 INFO SparkContext: Running Spark version 1.4.0\n",
    "2015-10-21 14:46:15.774 java[30685:c003] Unable to load realm info from SCDynamicStore\n",
    "15/10/21 14:46:15 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
    "15/10/21 14:46:15 INFO SecurityManager: Changing view acls to: rahul\n",
    "15/10/21 14:46:15 INFO SecurityManager: Changing modify acls to: rahul\n",
    "15/10/21 14:46:15 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(rahul); users with modify permissions: Set(rahul)\n",
    "15/10/21 14:46:16 INFO Slf4jLogger: Slf4jLogger started\n",
    "15/10/21 14:46:16 INFO Remoting: Starting remoting\n",
    "15/10/21 14:46:16 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriver@10.251.101.163:64359]\n",
    "15/10/21 14:46:16 INFO Utils: Successfully started service 'sparkDriver' on port 64359.\n",
    "15/10/21 14:46:16 INFO SparkEnv: Registering MapOutputTracker\n",
    "15/10/21 14:46:16 INFO SparkEnv: Registering BlockManagerMaster\n",
    "15/10/21 14:46:16 INFO DiskBlockManager: Created local directory at /private/var/folders/_f/y76rs29s3c57ykwyz9c8z12c0000gn/T/spark-00a4e09e-e5db-485f-81dc-2e5016e9a27e/blockmgr-8966e07c-223b-4c38-9273-11543aa9d3c1\n",
    "15/10/21 14:46:16 INFO MemoryStore: MemoryStore started with capacity 273.0 MB\n",
    "15/10/21 14:46:16 INFO HttpFileServer: HTTP File server directory is /private/var/folders/_f/y76rs29s3c57ykwyz9c8z12c0000gn/T/spark-00a4e09e-e5db-485f-81dc-2e5016e9a27e/httpd-6af0a9e0-1cfe-42c4-a1bd-e01715b98436\n",
    "15/10/21 14:46:16 INFO HttpServer: Starting HTTP Server\n",
    "15/10/21 14:46:17 INFO Utils: Successfully started service 'HTTP file server' on port 64360.\n",
    "15/10/21 14:46:17 INFO SparkEnv: Registering OutputCommitCoordinator\n",
    "15/10/21 14:46:18 INFO Utils: Successfully started service 'SparkUI' on port 4040.\n",
    "15/10/21 14:46:18 INFO SparkUI: Started SparkUI at http://10.251.101.163:4040\n",
    "15/10/21 14:46:18 INFO Executor: Starting executor ID driver on host localhost\n",
    "15/10/21 14:46:18 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 64361.\n",
    "15/10/21 14:46:18 INFO NettyBlockTransferService: Server created on 64361\n",
    "15/10/21 14:46:18 INFO BlockManagerMaster: Trying to register BlockManager\n",
    "15/10/21 14:46:18 INFO BlockManagerMasterEndpoint: Registering block manager localhost:64361 with 273.0 MB RAM, BlockManagerId(driver, localhost, 64361)\n",
    "15/10/21 14:46:18 INFO BlockManagerMaster: Registered BlockManager\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.context.SparkContext at 0x10441cf10>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "385"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]).map(lambda x: x**2).sum()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Create A RDD\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.rdd.RDD'>\n"
     ]
    }
   ],
   "source": [
    "wordsList = ['cat', 'elephant', 'rat', 'rat', 'cat']\n",
    "wordsRDD = sc.parallelize(wordsList, 4)\n",
    "# Print out the type of wordsRDD\n",
    "print type(wordsRDD)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Call `collect` on an RDD: Lazy Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Spark is lazy. Until you `collect`, nothing is actually run.\n",
    "\n",
    ">Instead, they just remember the transformations applied to some base dataset (e.g. a file). The transformations are only computed when an action requires a result to be returned to the driver program."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['cat', 'elephant', 'rat', 'rat', 'cat']"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "wordsRDD.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "15/10/21 14:59:59 INFO SparkContext: Starting job: collect at <ipython-input-6-dee494da0714>:1\n",
    "15/10/21 14:59:59 INFO DAGScheduler: Got job 0 (collect at <ipython-input-6-dee494da0714>:1) with 4 output partitions (allowLocal=false)\n",
    "15/10/21 14:59:59 INFO DAGScheduler: Final stage: ResultStage 0(collect at <ipython-input-6-dee494da0714>:1)\n",
    "15/10/21 14:59:59 INFO DAGScheduler: Parents of final stage: List()\n",
    "15/10/21 14:59:59 INFO DAGScheduler: Missing parents: List()\n",
    "15/10/21 14:59:59 INFO DAGScheduler: Submitting ResultStage 0 (ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:396), which has no missing parents\n",
    "15/10/21 15:00:00 INFO MemoryStore: ensureFreeSpace(1224) called with curMem=0, maxMem=286300569\n",
    "15/10/21 15:00:00 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 1224.0 B, free 273.0 MB)\n",
    "15/10/21 15:00:00 INFO MemoryStore: ensureFreeSpace(777) called with curMem=1224, maxMem=286300569\n",
    "15/10/21 15:00:00 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 777.0 B, free 273.0 MB)\n",
    "15/10/21 15:00:00 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on localhost:64361 (size: 777.0 B, free: 273.0 MB)\n",
    "15/10/21 15:00:00 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:874\n",
    "15/10/21 15:00:00 INFO DAGScheduler: Submitting 4 missing tasks from ResultStage 0 (ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:396)\n",
    "15/10/21 15:00:00 INFO TaskSchedulerImpl: Adding task set 0.0 with 4 tasks\n",
    "15/10/21 15:00:00 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, PROCESS_LOCAL, 1379 bytes)\n",
    "15/10/21 15:00:00 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1, localhost, PROCESS_LOCAL, 1384 bytes)\n",
    "15/10/21 15:00:00 INFO TaskSetManager: Starting task 2.0 in stage 0.0 (TID 2, localhost, PROCESS_LOCAL, 1379 bytes)\n",
    "15/10/21 15:00:00 INFO TaskSetManager: Starting task 3.0 in stage 0.0 (TID 3, localhost, PROCESS_LOCAL, 1403 bytes)\n",
    "15/10/21 15:00:00 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)\n",
    "15/10/21 15:00:00 INFO Executor: Running task 2.0 in stage 0.0 (TID 2)\n",
    "15/10/21 15:00:00 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)\n",
    "15/10/21 15:00:00 INFO Executor: Running task 3.0 in stage 0.0 (TID 3)\n",
    "15/10/21 15:00:00 INFO Executor: Finished task 1.0 in stage 0.0 (TID 1). 646 bytes result sent to driver\n",
    "15/10/21 15:00:00 INFO Executor: Finished task 3.0 in stage 0.0 (TID 3). 665 bytes result sent to driver\n",
    "15/10/21 15:00:00 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 641 bytes result sent to driver\n",
    "15/10/21 15:00:00 INFO Executor: Finished task 2.0 in stage 0.0 (TID 2). 641 bytes result sent to driver\n",
    "15/10/21 15:00:00 INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 61 ms on localhost (1/4)\n",
    "15/10/21 15:00:00 INFO TaskSetManager: Finished task 3.0 in stage 0.0 (TID 3) in 60 ms on localhost (2/4)\n",
    "15/10/21 15:00:00 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 96 ms on localhost (3/4)\n",
    "15/10/21 15:00:00 INFO TaskSetManager: Finished task 2.0 in stage 0.0 (TID 2) in 63 ms on localhost (4/4)\n",
    "15/10/21 15:00:00 INFO DAGScheduler: ResultStage 0 (collect at <ipython-input-6-dee494da0714>:1) finished in 0.120 s\n",
    "15/10/21 15:00:00 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool\n",
    "15/10/21 15:00:00 INFO DAGScheduler: Job 0 finished: collect at <ipython-input-6-dee494da0714>:1, took 0.872367 s\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###Operations on RDDs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From the Spark Programming Guide:\n",
    "\n",
    ">RDDs support two types of operations: transformations, which create a new dataset from an existing one, and actions, which return a value to the driver program after running a computation on the dataset. For example, map is a transformation that passes each dataset element through a function and returns a new RDD representing the results. On the other hand, reduce is an action that aggregates all the elements of the RDD using some function and returns the final result to the driver program (although there is also a parallel reduceByKey that returns a distributed dataset)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###Word Examples"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "cats\n"
     ]
    }
   ],
   "source": [
    "def makePlural(word):\n",
    "    return word + 's'\n",
    "\n",
    "print makePlural('cat')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Transform one RDD into another."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "cats\n",
      "['cats', 'elephants']\n"
     ]
    }
   ],
   "source": [
    "pluralRDD = wordsRDD.map(makePlural)\n",
    "print pluralRDD.first()\n",
    "print pluralRDD.take(2)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['cats']"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pluralRDD.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['cats', 'elephants', 'rats', 'rats', 'cats']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pluralRDD.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Key Value Pairs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('cat', 1), ('elephant', 1), ('rat', 1), ('rat', 1), ('cat', 1)]\n"
     ]
    }
   ],
   "source": [
    "wordPairs = wordsRDD.map(lambda w: (w, 1))\n",
    "print wordPairs.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "➜  sparklect  ps auxwww | grep pyspark\n",
    "rahul           30685   0.4  0.8  3458120  68712 s012  S+    2:46PM   2:00.21 /System/Library/Java/JavaVirtualMachines/1.6.0.jdk/Contents/Home/bin/java -cp /usr/local/opt/apache-spark/libexec/conf/:/usr/local/opt/apache-spark/libexec/lib/spark-assembly-1.4.0-hadoop2.6.0.jar:/usr/local/opt/apache-spark/libexec/lib/datanucleus-api-jdo-3.2.6.jar:/usr/local/opt/apache-spark/libexec/lib/datanucleus-core-3.2.10.jar:/usr/local/opt/apache-spark/libexec/lib/datanucleus-rdbms-3.2.9.jar -Xms512m -Xmx512m -XX:MaxPermSize=128m org.apache.spark.deploy.SparkSubmit pyspark-shell\n",
    "rahul           31520   0.0  0.0  2432784    480 s011  R+    6:42PM   0:00.00 grep --color=auto --exclude-dir=.bzr --exclude-dir=.cvs --exclude-dir=.git --exclude-dir=.hg --exclude-dir=.svn pyspark\n",
    "rahul           31494   0.0  0.7  2548972  57288 s012  S     6:41PM   0:00.10 python -m pyspark.daemon\n",
    "rahul           31493   0.0  0.7  2548972  57308 s012  S     6:41PM   0:00.10 python -m pyspark.daemon\n",
    "rahul           31492   0.0  0.7  2548972  57288 s012  S     6:41PM   0:00.11 python -m pyspark.daemon\n",
    "rahul           31446   0.0  0.8  2548972  68460 s012  S     6:35PM   0:01.34 python -m pyspark.daemon\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### WORD COUNT!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('rat', 2), ('elephant', 1), ('cat', 2)]\n"
     ]
    }
   ],
   "source": [
    "wordsList = ['cat', 'elephant', 'rat', 'rat', 'cat']\n",
    "wordsRDD = sc.parallelize(wordsList, 4)\n",
    "wordCountsCollected = (wordsRDD\n",
    "                       .map(lambda w: (w, 1))\n",
    "                       .reduceByKey(lambda x,y: x+y)\n",
    "                       .collect())\n",
    "print wordCountsCollected"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "![Tons of shuffling](https://databricks.gitbooks.io/databricks-spark-knowledge-base/content/images/reduce_by.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(4) PythonRDD[19] at RDD at PythonRDD.scala:43 []\n",
      " |  MapPartitionsRDD[18] at mapPartitions at PythonRDD.scala:346 []\n",
      " |  ShuffledRDD[17] at partitionBy at NativeMethodAccessorImpl.java:-2 []\n",
      " +-(4) PairwiseRDD[16] at reduceByKey at <ipython-input-13-c1214cba7909>:3 []\n",
      "    |  PythonRDD[15] at reduceByKey at <ipython-input-13-c1214cba7909>:3 []\n",
      "    |  ParallelCollectionRDD[9] at parallelize at PythonRDD.scala:396 []\n"
     ]
    }
   ],
   "source": [
    "print (wordsRDD\n",
    "    .map(lambda w: (w, 1))\n",
    "    .reduceByKey(lambda x,y: x+y)).toDebugString()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Using Cache"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ParallelCollectionRDD[20] at parallelize at PythonRDD.scala:396\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "5"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "wordsList = ['cat', 'elephant', 'rat', 'rat', 'cat']\n",
    "wordsRDD = sc.parallelize(wordsList, 4)\n",
    "print wordsRDD\n",
    "wordsRDD.count()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#this is rerun from the start\n",
    "wordsRDD.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[20] at parallelize at PythonRDD.scala:396"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#default storage level (MEMORY_ONLY)\n",
    "wordsRDD.cache()#nothing done this is still lazy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#parallelize is rerun and cached because we told it to cache\n",
    "wordsRDD.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#this `sc.parallelize` is not rerun in this case\n",
    "wordsRDD.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Where is this useful: it is when you have branching parts or loops, so that you dont do things again and again. Spark, being \"lazy\" will rerun the chain again. So `cache` or `persist` serves as a checkpoint, breaking the RDD chain or the *lineage*."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'cat': 'mammal',\n",
       " 'elephant': 'mammal',\n",
       " 'heron': 'bird',\n",
       " 'owl': 'bird',\n",
       " 'rat': 'mammal'}"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "birdsList=['heron','owl']\n",
    "animList=wordsList+birdsList\n",
    "animaldict={}\n",
    "for e in wordsList:\n",
    "    animaldict[e]='mammal'\n",
    "for e in birdsList:\n",
    "    animaldict[e]='bird'\n",
    "animaldict"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5 2\n"
     ]
    }
   ],
   "source": [
    "animsrdd = sc.parallelize(animList, 4)\n",
    "animsrdd.cache()\n",
    "#below runs the whole chain but causes cache to be populated\n",
    "mammalcount=animsrdd.filter(lambda w: animaldict[w]=='mammal').count()\n",
    "#now only the filter is carried out\n",
    "birdcount=animsrdd.filter(lambda w: animaldict[w]=='bird').count()\n",
    "print mammalcount, birdcount"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fun with words"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Read http://spark.apache.org/docs/latest/programming-guide.html ."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "stopwords=[e.strip() for e in open(\"./sparklect/english.stop.txt\").readlines()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "juliusrdd=sc.textFile(\"./sparklect/shakes/juliuscaesar.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "21245"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "juliusrdd.flatMap(lambda line: line.split()).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'1599',\n",
       " u'the',\n",
       " u'tragedy',\n",
       " u'of',\n",
       " u'julius',\n",
       " u'caesar',\n",
       " u'by',\n",
       " u'william',\n",
       " u'shakespeare',\n",
       " u'dramatis',\n",
       " u'personae',\n",
       " u'julius',\n",
       " u'caesar,',\n",
       " u'roman',\n",
       " u'statesman',\n",
       " u'and',\n",
       " u'general',\n",
       " u'octavius,',\n",
       " u'triumvir',\n",
       " u'after']"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(juliusrdd.flatMap(lambda line: line.split())\n",
    " .map(lambda word: word.strip().lower())\n",
    "  .take(20)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'1599',\n",
       " u'tragedy',\n",
       " u'julius',\n",
       " u'caesar',\n",
       " u'william',\n",
       " u'shakespeare',\n",
       " u'dramatis',\n",
       " u'personae',\n",
       " u'julius',\n",
       " u'caesar,',\n",
       " u'roman',\n",
       " u'statesman',\n",
       " u'general',\n",
       " u'octavius,',\n",
       " u'triumvir',\n",
       " u\"caesar's\",\n",
       " u'death,',\n",
       " u'augustus',\n",
       " u'caesar,',\n",
       " u'emperor']"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(juliusrdd.flatMap(lambda line: line.split())\n",
    " .map(lambda word: word.strip().lower())\n",
    " .filter(lambda word: word not in stopwords)\n",
    " .take(20)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'1599', 1),\n",
       " (u'tragedy', 1),\n",
       " (u'julius', 1),\n",
       " (u'caesar', 1),\n",
       " (u'william', 1),\n",
       " (u'shakespeare', 1),\n",
       " (u'dramatis', 1),\n",
       " (u'personae', 1),\n",
       " (u'julius', 1),\n",
       " (u'caesar,', 1),\n",
       " (u'roman', 1),\n",
       " (u'statesman', 1),\n",
       " (u'general', 1),\n",
       " (u'octavius,', 1),\n",
       " (u'triumvir', 1),\n",
       " (u\"caesar's\", 1),\n",
       " (u'death,', 1),\n",
       " (u'augustus', 1),\n",
       " (u'caesar,', 1),\n",
       " (u'emperor', 1)]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(juliusrdd.flatMap(lambda line: line.split())\n",
    " .map(lambda word: word.strip().lower())\n",
    " .filter(lambda word: word not in stopwords)\n",
    " .map(lambda word: (word, 1))\n",
    " .take(20)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'fawn', 2),\n",
       " (u'pardon', 5),\n",
       " (u'shouted.', 1),\n",
       " (u'foul', 1),\n",
       " (u'cause;', 1),\n",
       " (u'cognizance.', 1),\n",
       " (u'sleep', 6),\n",
       " (u'abide', 2),\n",
       " (u\"friend's\", 1),\n",
       " (u'muffling', 1),\n",
       " (u'hate', 1),\n",
       " (u'up.', 3),\n",
       " (u'up,', 3),\n",
       " (u'pardon!', 1),\n",
       " (u'venom', 1),\n",
       " (u'presents', 1),\n",
       " (u'whatsoever', 1),\n",
       " (u'sway', 1),\n",
       " (u'hats', 1),\n",
       " (u'reveler!', 1)]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(juliusrdd.flatMap(lambda line: line.split())\n",
    " .map(lambda word: word.strip().lower())\n",
    " .filter(lambda word: word not in stopwords)\n",
    " .map(lambda word: (word, 1))\n",
    " .reduceByKey(lambda a, b: a + b)\n",
    " .take(20)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'brutus.', 211),\n",
       " (u'cassius.', 152),\n",
       " (u'thou', 107),\n",
       " (u'caesar', 96),\n",
       " (u'brutus', 75),\n",
       " (u'antony.', 73),\n",
       " (u'citizen.', 68),\n",
       " (u'good', 66),\n",
       " (u'caesar.', 62),\n",
       " (u'thy', 54),\n",
       " (u'brutus,', 54),\n",
       " (u'caesar,', 46),\n",
       " (u'casca.', 44),\n",
       " (u'\"', 44),\n",
       " (u'men', 41),\n",
       " (u'you,', 41),\n",
       " (u'enter', 40),\n",
       " (u\"caesar's\", 40),\n",
       " (u'lucius.', 38),\n",
       " (u'cassius,', 38)]"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(juliusrdd.flatMap(lambda line: line.split())\n",
    " .map(lambda word: word.strip().lower())\n",
    " .filter(lambda word: word not in stopwords)\n",
    " .map(lambda word: (word, 1))\n",
    " .reduceByKey(lambda a, b: a + b)\n",
    " .takeOrdered(20, lambda x: -x[1])\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "image/png": "iVBORw0KGgoAAAANSUhEUgAAAeMAAAFxCAYAAABA2cBwAAAABHNCSVQICAgIfAhkiAAAAAlwSFlz\nAAALEgAACxIB0t1+/AAAIABJREFUeJzt3Xl4VdWh/vH3hCTYDMgUqIUYMMYCKiASrjhQwYqgCAqE\nMAXKpKVCKWmRWRCEoEyicEWt1WtkVloRvV4VBy5ipVYRIRAkIOAFIZBAkyBJIPv3B7+cElSmsxbr\nHPh+nofnISF5s9hPznn3sPbaPs/zPAEAAGfCXA8AAIBLHWUMAIBjlDEAAI5RxgAAOEYZAwDgGGUM\nAIBj4af7x9LSUo0ZM0Z79uxRSUmJBg8erJ///Od68MEHVa9ePUlSz5491b59ey1dulRLlixReHi4\nBg8erNtvv/0CDB8AgNDnO919xsuXL1d2drZGjx6tw4cPq1OnTnrooYdUWFiofv36+b8uNzdX/fv3\n1/Lly1VcXKwePXrotddeU2Rk5AX5TwAAEMpOe2Tcrl073XXXXZKksrIyhYeHa9OmTdqxY4dWrVql\nhIQEjRkzRhs2bFCzZs0UERGhiIgIJSQkKDs7W9dff/0F+U8AABDKTlvGUVFRkqTCwkINGzZMw4cP\nV3Fxsbp166ZGjRpp/vz5mjt3rho2bKjY2Fj/90VHR6uwsNDuyAEAuEiccQLX3r171bdvX9133326\n5557dOedd6pRo0aSpDvvvFObN29WTEyMioqK/N9TVFSkKlWqnDb32LHjAQ4dAICLw2mPjA8cOKD+\n/ftrwoQJuummmyRJAwcO1NixY9W4cWOtXbtW1113nRo3bqzZs2erpKRExcXFysnJUVJS0ml/cH7+\nkTMOLi4uVrm5Befw3zk7tnJtZodars3sUMu1mR1quTazQy3XZnao5drMDrbcuLjYH/38act4/vz5\nKigo0Lx58zRv3jxJ0pgxY5SRkaHw8HDVqlVLkyZNUnR0tPr06aOePXuqrKxM6enpTN4CAOAsnbaM\nx40bp3Hjxv3g84sWLfrB51JSUpSSkmJuZAAAXCJY9AMAAMcoYwAAHKOMAQBwjDIGAMAxyhgAAMco\nYwAAHKOMAQBwjDIGAMAxyhgAAMcoYwAAHKOMAQBwjDIGAMAxyhgAAMcoYwAAHKOMAQBwjDIGAMAx\nyhgAAMcoYwAAHKOMAQBwjDIGAMAxyhgAAMcoYwAAHKOMAQBwjDIGAMAxyhgAAMcoYwAAHKOMAQBw\njDIGAMAxyhgAAMcoYwAAHKOMAQBwjDIGAMAxyhgAAMcoYwAAHKOMAQBwjDIGAMAxyhgAAMcoYwAA\nHKOMAQBwjDIGAMAxyhgAAMcoYwAAHKOMAQBwLNzVD87J+fqMX5OfH6O8vMLTfk18fIIiIyNNDQsA\ngAvOWRkPm75CUZfXCijjyOH9mjOioxITkwyNCgCAC89ZGUddXksx1eq4+vEAAAQNrhkDAOAYZQwA\ngGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4dtpFP0pLSzVmzBjt2bNHJSUlGjx4sBIT\nEzVq1CiFhYUpKSlJEyZMkM/n09KlS7VkyRKFh4dr8ODBuv322y/QfwEAgNB22jJ+4403VL16dU2f\nPl2HDx9Wp06d1LBhQ6Wnpys5OVkTJkzQqlWr1KRJE2VmZmr58uUqLi5Wjx49dPPNN7NmNAAAZ+G0\nZdyuXTvdddddkqSysjKFh4crKytLycnJkqRWrVrp448/VlhYmJo1a6aIiAhFREQoISFB2dnZuv76\n6+3/DwAACHGnvWYcFRWl6OhoFRYWatiwYfrDH/6gsrIy/79HR0eroKBAhYWFio2NrfD5wsLTP20J\nAACccMYHRezdu1dDhgxRr1691KFDB02fPt3/b4WFhapSpYpiYmJUVFTk/3xRUZGqVKliZ8SnqF49\nRnFxsWf+wlOcz/e4zg61XJvZoZZrMzvUcm1mh1quzexQy7WZHQq5py3jAwcOqH///powYYJuuukm\nSVLDhg21bt06tWjRQqtXr1bLli3VuHFjzZ49WyUlJSouLlZOTo6Ski7MYw3z8gqVm1twTt8TFxd7\nzt/jOjvUcm1mh1quzexQy7WZHWq5NrNDLddmdrDl/lSBn7aM58+fr4KCAs2bN0/z5s2TJI0dO1ZT\npkxRaWmpEhMT1a5dO/l8PvXp00c9e/ZUWVmZ0tPTmbwFAMBZOm0Zjxs3TuPGjfvB5zMzM3/wuZSU\nFKWkpJgbGQAAlwgW/QAAwDHKGAAAxyhjAAAco4wBAHCMMgYAwDHKGAAAxyhjAAAco4wBAHCMMgYA\nwDHKGAAAxyhjAAAco4wBAHCMMgYAwDHKGAAAxyhjAAAco4wBAHCMMgYAwDHKGAAAxyhjAAAco4wB\nAHCMMgYAwDHKGAAAxyhjAAAco4wBAHCMMgYAwDHKGAAAxyhjAAAco4wBAHCMMgYAwDHKGAAAxyhj\nAAAco4wBAHCMMgYAwDHKGAAAxyhjAAAco4wBAHCMMgYAwDHKGAAAxyhjAAAco4wBAHCMMgYAwDHK\nGAAAxyhjAAAco4wBAHCMMgYAwDHKGAAAxyhjAAAco4wBAHCMMgYAwDHKGAAAxyhjAAAco4wBAHCM\nMgYAwLGzKuMvv/xSaWlpkqSsrCy1atVKaWlpSktL03//939LkpYuXaouXbooNTVVH374obUBAwBw\nsQk/0xc8//zzWrFihaKjoyVJmzZtUr9+/dSvXz//1+Tm5iozM1PLly9XcXGxevTooZtvvlmRkZH2\nRv4TSkpKtHv3ztN+TX5+jPLyCk/7NfHxCU7GDwC49JyxjBMSEjR37lw9/PDDkqSNGzfqm2++0apV\nq5SQkKAxY8Zow4YNatasmSIiIhQREaGEhARlZ2fr+uuvt/4fONXu3Ts1bPoKRV1e67wzjhzerzkj\nOioxMcngyAAA+HFnLOO2bdvq22+/9X/cpEkTpaamqlGjRpo/f77mzp2rhg0bKjY21v810dHRKiw8\n/ZGnTVGX11JMtTrOfj4AAOfijGV8qjvvvNNfvHfeeacmT56s5ORkFRUV+b+mqKhIVapUMTfK06he\nPUZxcf/eEcjPj7GSey7O9/sutlyb2aGWazM71HJtZodars3sUMu1mR0KuedcxgMHDtTYsWPVuHFj\nrV27Vtddd50aN26s2bNnq6SkRMXFxcrJyVFS0oU5xZuXV6jc3IIKH9vIPVtxcbHn9X0XW67N7FDL\ntZkdark2s0Mt12Z2qOXazA623J8q8LMuY5/PJ0l69NFH9eijjyo8PFy1atXSpEmTFB0drT59+qhn\nz54qKytTeno6k58AADhLZ1XGdevW1eLFiyVJDRo00KJFi37wNSkpKUpJSTE7OgAALgEs+gEAgGOU\nMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAY\nZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4\nRhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAA\njlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwA\ngGOUMQAAjlHGAAA4RhkDAOBYuOsBhIqSkhLt3r3zjF+Xnx+jvLzC035NfHyCIiMjTQ0NABDiKOOz\ntHv3Tg2bvkJRl9cKKOfI4f2aM6KjEhOTDI0MABDqKONzEHV5LcVUq+N6GACAiwzXjAEAcIwyBgDA\nsbMq4y+//FJpaWmSpJ07d6pHjx7q1auXJk6cKM/zJElLly5Vly5dlJqaqg8//NDagAEAuNicsYyf\nf/55jRs3TqWlpZKkjIwMpaena8GCBfI8T6tWrVJubq4yMzO1ePFivfDCC5o5c6ZKSkqsDx4AgIvB\nGcs4ISFBc+fO9R8BZ2VlKTk5WZLUqlUrrV27Vl999ZWaNWumiIgIxcTEKCEhQdnZ2XZHDgDAReKM\nZdy2bVtVqlTJ/3F5KUtSdHS0CgoKVFhYqNjY2AqfLyw8/b22AADghHO+tSks7N/9XVhYqCpVqigm\nJkZFRUX+zxcVFalKlSpmRngG1avHKC7u3zsC+fkxQZ37Y9ln63y+x2WuzexQy7WZHWq5NrNDLddm\ndqjl2swOhdxzLuOGDRtq3bp1atGihVavXq2WLVuqcePGmj17tkpKSlRcXKycnBwlJV2YRS3y8gqV\nm1tQ4eNgzv2x7LMRFxd7zt/jMtdmdqjl2swOtVyb2aGWazM71HJtZgdb7k8V+FmXsc/nkySNGjVK\n48ePV2lpqRITE9WuXTv5fD716dNHPXv2VFlZmdLT01nuEQCAs3RWZVy3bl0tXrxYklSvXj1lZmb+\n4GtSUlKUkpJidnQAAFwCWA7TMR5AAQCgjB3jARQAAMo4CPAACgC4tLE2NQAAjlHGAAA4RhkDAOAY\nZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4\nRhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOBYuOsBwI6SkhLt\n3r3zjF+Xnx+jvLzC035NfHyCIiMjTQ0NAHAKyvgitXv3Tg2bvkJRl9cKKOfI4f2aM6KjEhOTDI0M\nAHAqyvgiFnV5LcVUq+N6GACAM6CMcc44BQ4AZlHGOGecAgcAsyhjnBdOgQOAOdzaBACAY5QxAACO\nUcYAADjGNWMEDWZpA7hUUcYIGszSBnCpoowRVGzM0uaIG0Cwo4xx0eOIG0Cwo4xxSeC+aADBjNnU\nAAA4RhkDAOAYp6mBADA5DIAJlDEQACaHATCBMgYCxOQwAIHimjEAAI5RxgAAOEYZAwDgGGUMAIBj\nlDEAAI5RxgAAOEYZAwDgGPcZA0HI1sperBgGBCfKGAhCtlb2YsUwIDiddxnff//9iomJkSTFx8fr\nwQcf1KhRoxQWFqakpCRNmDBBPp/P2ECBS42tlb1YMQwIPudVxsXFxZKkzMxM/+d++9vfKj09XcnJ\nyZowYYJWrVqlX//612ZGCQDARey8JnBt2bJF33//vQYMGKC+fftq/fr1ysrKUnJysiSpVatWWrt2\nrdGBAgBwsTqvI+Of/exnGjBggFJSUvTNN99o4MCBFf49KipKBQUFRgYIAMDF7rzKuF69ekpISPD/\nvWrVqtq8ebP/34uKilSlShUzIzyD6tVjFBcX6/84Pz8mqHNPzQ61XJvZoZZrMzvUcs/V+X7fxZZr\nMzvUcm1mh0LueZXx8uXLlZ2drQkTJmjfvn0qKirSLbfconXr1qlFixZavXq1WrZsaWyQp5OXV6jc\n3IIKHwdz7qnZoZZrMzvUcm1mh1ruuYiLiz2v77vYcm1mh1quzexgy/2pAj+vMu7atatGjx6tXr16\nSZIyMjJUtWpVjR8/XqWlpUpMTFS7du3OJxoAgEvOeZVxeHi4pk+f/oPPnzy7GgAAnB0W/QBgBKt7\nAeePMgZgBKt7AeePMgZgjI3VvVinG5cCyhhAUGOdblwKKGMAQY91unGx43nGAAA4xpExABjGdW6c\nK8oYAAwLtevcNkveVPbFvvNAGQOABaF0ndvmZDYT2ZfCJDnKGABgdTIbE+XOjDIGAISci+26PGUM\nAAg5oXZd/kwoYwBASAql6/Jnwn3GAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOU\nMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAY\nZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4\nRhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAAjlHGAAA4RhkDAOAYZQwAgGOUMQAA\njoWbDCsrK9PEiRO1detWRUREaMqUKbryyitN/ggAAC46Ro+M33vvPZWWlmrx4sX605/+pGnTppmM\nBwDgomS0jD///HPddtttkqQmTZpo48aNJuMBALgoGT1NXVhYqJiYGP/HlSpVUllZmcLCftj5Rw7v\nD/jn/VRGoNm2cn8qI9RybWaHWq7N7FDLtZkdark2s0Mt10T2xbKNT8fneZ4X8E/9/6ZNm6YmTZqo\nffv2kqRf/epX+uijj0zFAwBwUTJ6mrpZs2ZavXq1JGn9+vX65S9/aTIeAICLktEjY8/zNHHiRGVn\nZ0uSMjIyVL9+fVPxAABclIyWMQAAOHcs+gEAgGOUMQAAjlHGAAA4RhkDwEn27NmjkpIS18PAJSbo\ny/jYsWOuhwDgEtK7d28NGDBAeXl5roeCS0jQl/GDDz5oJffRRx81ktOmTZsKfzp16mQkV5JycnKM\nZZ3su+++09dff63t27dr9OjR2rx5s5Wfs3TpUiu5kvT+++8H9P1paWk/+qdPnz6GRljRvHnzjOTk\n5uYayTlbaWlpmjp1qr777jujuc8995yeffZZlZaWGsvcsGGDkZz7779fHTp00M9+9jMjeSdbs2aN\nBg4caP337cCBA/r++++NZKWnpxvJOdXHH3+sjz76SB9++KHuuOMOrVixwsrPMfVe/1OmTp2q2bNn\nq7CwMLAg7xK1bds2IzlHjx71jh496n3//ffeZ5995j322GNGcj3P81JTU41lnaxnz57eJ5984g0Z\nMsR74403vN69exvJ3b9/f4WP33jjDSO5P+bFF18M6PsPHjzoHTx40Bs5cqS3Zs0a7+jRo97f//53\nb/z48UbGd+rv1yeffGIkNzU11Rs8eLD3/vvve8ePHzeSeSGcuj1Wr17tHTx40Dtw4ICxn/HEE08Y\nyXnqqae8pUuXet9//72RvJPdfffd3scff+xt27bN/8eG4cOHe7NmzfL27NkTcNaQIUO8zZs3e0eP\nHvWKi4u94uJiAyP0vC5dung7d+70+vfv7+3bt8/r0aOHkdxT2drG5bKysrwjR454BQUFAeUEXRnv\n3bvX27p1q5eTk+ONGjXKy8rKMpJbVlbm/etf//IKCgq8v/71r96hQ4eM5J7K5C9Uv379vClTpngL\nFizwFi1a5C1evNhIbu/evb3S0lKvb9++nud5Xp8+fYzkhmJRnLojYmrHxNaOlOd53tatW71p06Z5\nKSkp3qxZs7xdu3YZy7b1+rO5PULJwIEDXQ/hnN1zzz1e69at/X/atGljJLd3795eQUGB98ADD3ie\n53m9evUykmv7vd5WvtEHRZjwxz/+UUOHDtWCBQt01113aerUqcrMzAw4d/jw4br99tv1xRdfyPM8\nvfvuu0ZOHc6cOdP/99zcXFWqVCngzHI33HCDfD6f8WtXpaWlmjFjhpo3b66///3vxk4VLl68WF9/\n/bWWL1+uZ555Ri1btlTXrl0VHx8fcPaaNWv00ksvqbi4WJLk8/n08ssvB5xbqVIlLVu2TNdff73+\n+c9/Gjs1GRUVpalTp6pevXoKCwuTz+dTamqqkezatWsrPj5eGzdu1NatWzV16lRdddVVGjFiRMDZ\ntl5/NrbHF198oeXLl+vYsWPyPE+5ubl64YUXAh6rTTVq1NAjjzyihg0byufzGfu9mDt37g8+N2TI\nkIBzJWnlypWSpPz8fFWtWlU+n89IbnR0tAYOHKju3btrwYIFql69upFcW+/1tvODrozDwsLUvHlz\nzZ8/Xx06dNCyZcuM5O7fv1/33XefXnvtNWVmZuo3v/mNkdz69ev7fzkbNmzof4SkCUOHDtW+ffv8\nbzb79wf+JBHpxDKla9euVUpKit577z09/vjjRnIle0WRkZGhsWPHqnbt2oZGesKMGTM0f/58vf32\n20pMTNQTTzxhJNfWjtSwYcO0detWdezYUTNmzPBvj86dOxvJt/X6s7E9Jk6cqEGDBul//ud/dM01\n1+iKK64wlm1LnTp15PP5dPDgQaO5NWrUkM/nU1lZmbKyslRWVmYse926dZo0aZKOHz+u9u3b64or\nrlBKSkrAuXPmzNGuXbuUlJSkrVu3GsmU7L3X284PujK2ddR27NgxvfPOO7r66quVl5enoqIiI7kd\nO3bUkiVL9PXXX6t+/fqqXLmykVxJGj16tL788ksdOXJER48eVZMmTfTss88GnJubm6ukpCR9+eWX\nqlmzpr777jsjR682i+IXv/iFbr755oBzTlWzZk3deuutql27turXr6+qVasaybW1I9WtWzfdcsst\nP/j8woULjeTbev3Z2B7VqlVThw4dtGbNGg0dOlSDBg0yMFK7hg4dqrVr12rXrl1q2rSp6tWrZyS3\nR48eFT4eMGCAkVxJevLJJ5WZmanf//736t+/v9LS0gIqzrS0NNWuXVtjx45VUlKSJOmaa64xNVxr\n7/W284OujG0dtQ0cOFBvvvmmRo8erczMTP3ud78zkjt+/HhVqVJFt956qz799FONGzfO2NFVdna2\nVq5cqQkTJmj48OGaMmWKkdxFixbJ5/PJ8zxt27ZNderUUXJycsC5NovC1um9GTNmaOfOnbrxxhv1\nt7/9TZ999plGjRoVcK6tHamSkhI98MAD/pmy5afrL7vssoCzJXuvPxvbo1KlStq6dauOHj2qnJwc\n4zO+bZg5c6b27dunbdu2KTw8XM8995xmzZoVcO6OHTv8f9+/f7/27t0bcGa5sLAwVatWTZIUExNT\n4Zn152PIkCGKi4tTXl6eP9ckW+/1tvODroxtHbW1bdtWbdu2lXTiCM6UnTt3+svm17/+tbHrgpJU\ntWpVhYWF6ciRI6pevbqx21pmz57t/3tJSYmx7VGtWjVNmDChwoIJGRkZRorC1um9zz77TIsXL5Yk\n9e3b19ipMls7UnPmzNGYMWNUo0YNI3mnsvX6s7E9Ro4cqW3btql3794aMWKEunTpEnCmbf/85z+1\ncOFCpaWlqWvXrsZu/3vkkUf8f69cubJGjhxpJFeSrrzySs2YMUOHDh3Ss88+q1/84hcB5f3Hf/yH\nJKl79+7+155Jtt7rbecHXRnbOmq79dZbJZ14zOPhw4dVt25dvf322wHnlpSU6MiRI4qKitL3339v\n9FrNtddeqz//+c+qVauWhg8friNHjhjLLnfs2DF9++23RrJGjRrlPwUlydhED8ne6b1jx47p+PHj\nqlSpksrKyhQWZubWe1s7UlWrVlWLFi2MZP0YW68/G9ujZs2ays3NVfPmzdWlSxfde++9AWfaVlZW\n5p+EeOzYMWO/b5mZmcrPz9fu3btVt25dY5OhJGnSpElaunSpbrzxRkVFRWny5MlGcm1NcrT1Xm87\nP+jK2NZR25o1a/x//7//+78fnX14Pvr06aP77rtPV199tXJycjR06FAjudKJma2FhYW67LLLtHr1\najVu3NhI7i233OIvytLSUvXt29dIblxcnLEjy1PZOr139913q0ePHmrSpIk2bNigu+++28Boze9I\nlR9BREREaPz48WrUqJHR0/XlbL3+bOxYpqen+xfNuPzyyzVixAgjlwJs6tu3rzp37qy8vDx169ZN\n/fr1M5L71ltvac6cOUpMTNTWrVs1ZMgQ3XfffUayp06dWuHI++GHHzZyKc7WJEdb7/W284OujE9m\n8qjtZHXq1DG2ulXHjh3VqlUr/x6pyWsgW7Zs0ZEjRxQWFqaXXnpJv/3tb1WzZs2Ac1999dUKM09N\nbYs6deroueeeU8OGDSWdODIu34sMlK3Te/3799ett96q7du3KyUlxdhEEtM7UgcOHJB04o3goYce\n8p+uLz/KssHk68/GjuXRo0fVpk0bSVKHDh20ZMmSgDNti4yMVHx8vP91/Prrrxs5on/ppZe0fPly\nRUdHq7Cw0H+QEIhXXnlF8+fP16FDh/TOO+9IOnE0mJiYGPB4JXuTHE9m8r3edn7QlbGto7aTl3TL\nzc01UmqS9MEHH2jRokU/mFBjwoQJE/TII4/oqaee0vDhwzV9+vSAZhRnZ2dr//79mjFjhv9Wo+PH\nj2vmzJlGlqIrKSnRjh07KkwmMVXGJ5/eO378uLHTe3v37tXcuXO1bds21a9fX6NHj1bdunUDzjW9\nI1W7dm0tW7ZMUVFR+t///V9JJ7ZJaWmp/vjHPwY83nK2Xn82dizDw8O1Zs0aNW3aVF999ZXRe/xt\neeKJJzRp0iRVqVJFkrlLOWFhYYqOjpZ0YpKViXkavXv3Vu/evfXMM89o8ODBAeedytYkR1vv9bbz\ng66MbR21de/e3f/3ypUr67rrrjOSa3NCTWRkpK6++modO3ZMN9xwQ8BvNgUFBXrzzTd14MABvfnm\nm5JOvBn06tXLxHA1bdo0bd26Vdu2bVO9evXUqFEjI7lSxdN7KSkpxu7tGzdunHr27KnmzZvrH//4\nh8aOHav/+q//CjjX9I5Up06d1LJlS82fP1+DBw+W53mqVKmS8d87W68/09tDkh577DE9/vjjmjJl\nihITEzVp0iQjY7UpKSnJP4HJpPj4eE2bNk3NmzfXZ599piuvvNJYdlpamt56660KEzNNnAK3NcnR\n1nu97fygKWNbR23vv/++2rRpo+3bt1f4/JYtW4xca7M5ocbn8+nhhx9Wq1at9NZbbykiIiKgvObN\nm6t58+batGmTrr32WkOj/LeXX35ZK1euVJMmTfSXv/xF7dq108CBA41kt27dWi1bttSuXbtUt25d\n5efnG8ktKSnRHXfcIenEbPgXX3zRSK7pHanIyEjVrVtXjz32mJHxncr2WRPT20OS6tWrp7lz56qs\nrEzr16/Xz3/+84AzbbvjjjvUrVu3Cqd6MzIyAs49fPiw6tWrp7Vr1yoxMdHo2ZLBgwerdu3axhdV\nMT2pz/Z7ve38oCljW0dthw8flnTidILJ2b0XYkLNk08+qa+++kqtWrXSp59+amTCknTiCPZkpk6t\nr1y5UgsXLlR4eLhKS0uVmppqrIxbtmypOXPmqFWrVpJO3FJgYpnGY8eOacuWLWrQoIGys7ON/Y6Y\n3pGyzfY+v9WqAAAKTUlEQVRZExvb47HHHlNiYqL27NmjrKws1axZ0+hqcja8/PLLGjRokP9eXVO/\nbyNHjtSrr77qX9L122+/Vf369Y1kSyfuxzfN9KQ+W+/1Fyo/6B4UsXHjRiu5py7unZ+fH1De008/\n7T399NNegwYN/H9/+umnvRkzZhgasecVFBR4s2bN8kaNGuW988473jfffGMk9+Qnxrz++uvetGnT\njOSmpKRU+NjkwwHuv/9+r3///t7f/vY3z/PMPdAhKyvL69y5s3fbbbd5Xbp08TZv3mwk9+DBg96H\nH37olZWVeZ988knAv28Xiq3Xn43tUf77Vf67YOqBJzYNGjTIav7Bgwe94cOHe9dee633m9/8xvv8\n888Dzpw8ebL3xRdf+J/YZOqpTZ534j2utLTUW7VqlZebm2sk85tvvvFWrFjheZ7nzZgxw9u9e7eR\n3HKmu6Rc0BwZl7N11GZ6ce8LMaFmzJgxatWqldatW6eqVatqzJgxWrBgQcC5J58iS0xMNLb+cLNm\nzTR06FDdeOON+vzzz3XDDTcYyZVOLCr/zDPPaPjw4Tpw4IDCw8386j7wwAM6ePCgqlWrpqysLPXo\n0UNxcXF65JFHApp8FhkZqc8//1xvv/222rRpo8OHDxtbatMmW68/G9ujrKxMGzduVN26dVVSUmJ8\n2UMbKleurAEDBlRYSc7E84I/+ugj/fWvf1VOTo46duyosWPH6vjx4xowYIDeeOONgLI//fRTvf/+\n+zp06JAuv/xyhYWFadWqVQGP2dbdIiNHjvQvetKqVStj80DKXTIPipg4caL/75s2bTL24HvTi3tf\niAk1+fn56tq1q15//XUlJyfL8zwjuSevepObm2vsIeSjRo3SBx98oO3bt6tz5866/fbbjeSWi4yM\n1Jw5czR27FitX7/eSGZycrKGDBmiq666Srt27dLcuXP1u9/9TiNGjAiojG3tSNlm6/VnY3t06tRJ\nEydOVEZGhqZPn270fmtbWrduLcnsgjiStGLFCvXo0UMtWrSokG1i3YNx48Zp0qRJioyM1N13323s\n2rGNSX3lyg8EkpOTjS7EJF1CD4qwddRmenFv2xNqpBMv2PLZrN99952xWzfK71mVTuypP/nkkwHl\nHTt2TGVlZUpPT9esWbN0yy23qKysTH369DF2m1f5JJfw8HA9/vjj/je1QO3du1dXXXWVpBPL/u3Z\ns0f16tUL+Mjb1o6UbbZefza2R69evfzXtMeOHVthtm+wMvV0rVOd/CjXk5Uv2xiIp556Sq+88oqx\nB0WUszGpT5JiY2O1ZMkSNW3aVBs2bPDf8mXKJfOgCFtHbbYXD7dh7NixGj16tH9lr5OPWgKxfft2\nY5PBJOm1117Ts88+qwMHDqhdu3aS/v0ovkDNmzdPDz300I+Ot/xnBSIuLk4zZsxQ06ZNtX79esXF\nxenjjz8OeIKRrR0p22y9/mxsj0WLFumll17yP1kqOjo64FOy+CGfz2f0QREn59qY5Pj444/rP//z\nP/Xuu+8qMTFRU6dONZJbzlaX+Lwg22U/eWmx8tMiJhZhkE7cquF5nr744gs1adJEkZGRRnIvlNLS\nUiO/sEOHDtVDDz1U4VnMJrbFsmXLjC+HWT7Ted26dT84mjJxv+bRo0e1ZMkSbd++Xddcc426du2q\nrKysCqsknY/s7GyNHz9eOTk5uuqqqzRx4kQrt5OZZuv1Z2N73Hvvvfrzn/+s+fPn66677tIHH3yg\n0aNHBzxWVDRmzBhVr15dH330kTp06KDt27cbmbWel5dX4W6RBg0aGJlXsWfPngrvFT6fL+CHW5zK\nRpcEXRmXn+o0LRRvgyjf8y9fLs7Unn+HDh1UVFSkQ4cOqVq1avL5fEYmZOzZs0crV66scLpwyJAh\nAWWeegpcOjFxZ9CgQUZubbpQTO1I2Wbr9XcqE9tjwIABeuGFFzRixAhNnz7d/zHMKi0t1bJly7R1\n61YlJiYqNTXVSPkUFhbq+eef1/79+9WmTRtdc801SkhICDi3W7du/oedfPvtt0pISNCiRYsCzi1n\nq0uC7jR1aWmptmzZYvyobePGjRo3bpzS0tKUmZlpbJk/mxYuXKiXX365wp6/CSNGjNCjjz6q+Ph4\nFRcXG3sKy7Bhw3TzzTfriiuukOd5Riap2DwFbpOtHSnbbL3+bGyP2NhYvfvuuwoLC9OiRYusrG2M\nE2sp9OzZ03iurUmOJ69b/69//Uvjx48POPNktrok6Mp4x44dGjx4sPGjtlC8DaJWrVqqXbu2CgsL\nddNNN+n55583kjt37lwtW7ZMNWrU0L59+/T73//eyCL7MTExGj58uIER/ltqaqpSU1P16quvqmvX\nrkazbbK1I2Wbrdefje3Rq1cvbdq0Senp6Zo8ebKxpxThwrgQkxxjYmK0e/duo5m2usTMavsGjRgx\nQj6fT/Hx8YqIiDCyVJz079sg+vfvHzK3Qdja84+KivLfglW7dm0ji8pLJ9bdffPNN7V9+3Zt3769\nwgMjzlf5Xu7OnTs1a9YszZw5U7Nmzbogp1IDceqO1LZt21wP6azYev3Z2B4ZGRlq3bq1ateurZEj\nRxrZacCFY2uSY2pqqrp3767U1FS1bdvW+NrUtrok6I6MbR21heJtEKb3/P/yl79IkipVqqQ//elP\nSk5O1vr16xUVFWViuNq8eXOF+1JLS0srzM49H+X3NMbGxqpy5cqKjY3VnDlzjN3bZ0uonkK19fqz\nsT0iIyP91xjj4+NDZsY6TjB9t0j57V116tTxf65Bgwb+J2SZYqtLgq6MbR21heJtEBkZGZo9e7Z/\nz3/kyJEaMGDAeedVrVpVPp9P9957r/+abnJysrEFCNq3b++/LijJSMnfdtttkqS3335bs2fPVkJC\ngpKTkzVy5EhjD2a3IVRPodp6/dnYHldccYVmzZqlJk2a6KuvvlKtWrUMjBQXyi9/+csK13fL35vP\nV/k8h/J1A2yx1SVBU8a2j9pC8Rqe6T1/WwsOlFu0aJFeeeUVPfPMM8a3cagdBZnekbLN9uvPxvbI\nyMjQ4sWLtXr1aiUmJobE2gH4N9OT+my/v5Wz1SVBU8a2j9psTYayKdT2/GvVqqVatWpZ2cahti1C\nbefB9uvPxva47LLLgv5yBX5aKB4gSfa6JGjK2PZeTShewwu1PX+b2zjUtkWo7TzYfv2F2vaAfaF4\ngCTZe58LukU/bPnHP/6hTZs2qX379po8ebJuuOGGoD5tGIoKCwu1a9cu1ahRQy+++KJat25tZJWs\nUHT06FEtXrxYO3bsUGJiorp37x5yK76ZxPbAqf7whz/onnvu0XvvvaemTZtq4cKFQT+PR7LXJZdM\nGXfu3Nk/AWj37t0aOXKkFi5c6HpYAHBJCtUDJFtdEnT3GdsSatfwAOBiFqr3idvqkqC5Zmwb16wA\nIHiE6gGSrS65ZE5Tc80KAILH8OHDFR8f7y+13bt3/+RzmYOJrS65ZMoYABA8OECqiDIGAMCxS2YC\nFwAAwYoyBgDAMcoYAADHKGMAAByjjAEAcOz/AV7hgV+Mu4DxAAAAAElFTkSuQmCC\n",
      "text/plain": [
       "<matplotlib.figure.Figure at 0x109b60b90>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "captions, counts=zip(*juliusrdd.flatMap(lambda line: line.split())\n",
    " .map(lambda word: word.strip().lower())\n",
    " .filter(lambda word: word not in stopwords)\n",
    " .map(lambda word: (word, 1))\n",
    " .reduceByKey(lambda a, b: a + b)\n",
    " .takeOrdered(20, lambda x: -x[1])\n",
    ")\n",
    "pos = np.arange(len(counts))\n",
    "plt.bar(pos, counts);\n",
    "plt.xticks(pos+0.4, captions, rotation=90);"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "shakesrdd=sc.textFile(\"./sparklect/shakes/*.txt\", minPartitions=4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'1601',\n",
       " u'AS YOU LIKE IT',\n",
       " u'',\n",
       " u'by William Shakespeare',\n",
       " u'',\n",
       " u'',\n",
       " u'',\n",
       " u'DRAMATIS PERSONAE.',\n",
       " u'',\n",
       " u'']"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "shakesrdd.take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(2001, u'thou'),\n",
       " (1547, u'thy'),\n",
       " (1094, u'good'),\n",
       " (774, u'hath'),\n",
       " (703, u'thee'),\n",
       " (692, u'you,'),\n",
       " (640, u'make'),\n",
       " (599, u\"'tis\"),\n",
       " (597, u'enter'),\n",
       " (567, u'sir,'),\n",
       " (565, u'love'),\n",
       " (556, u'give'),\n",
       " (493, u'me,'),\n",
       " (489, u'o,'),\n",
       " (427, u'man'),\n",
       " (393, u'come,'),\n",
       " (390, u'lord,'),\n",
       " (389, u'sir'),\n",
       " (377, u'why,'),\n",
       " (359, u'you.'),\n",
       " (358, u'ham.'),\n",
       " (351, u'doth'),\n",
       " (348, u'speak'),\n",
       " (338, u'now,'),\n",
       " (327, u'king'),\n",
       " (324, u'him.'),\n",
       " (324, u'hear'),\n",
       " (323, u'time'),\n",
       " (314, u'art'),\n",
       " (312, u'scene')]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(shakesrdd.flatMap(lambda line: line.split())\n",
    " .map(lambda word: word.strip().lower())\n",
    " .filter(lambda word: word not in stopwords)\n",
    " .map(lambda word: (word, 1))\n",
    " .reduceByKey(lambda a, b: a + b)\n",
    " .map(lambda (x,y): (y,x))\n",
    " .sortByKey(0,1)\n",
    " .take(30)\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`SparkContext.wholeTextFile`...\n",
    "\n",
    ">lets you read a directory containing multiple small text files, and returns each of them as (filename, content) pairs. This is in contrast with textFile, which would return one record per line in each file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From http://www.stat.berkeley.edu/scf/paciorek-spark-2014.html:\n",
    "\n",
    ">You want each partition to be able to fit in the memory availalbe on a node, and if you have multi-core nodes, you want that as many partitions as there are cores be able to fit in memory.\n",
    "\n",
    ">For load-balancing you'll want at least as many partitions as total computational cores in your cluster and probably rather more partitions. The Spark documentation suggests 2-4 partitions (which they also seem to call slices) per CPU. Often there are 100-10,000 partitions. Another rule of thumb is that tasks should take at least 100 ms. If less than that, you may want to repartition to have fewer partitions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###DataFrames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Convert Spark DataFrame to Pandas\n",
    "\n",
    "`pandas_df = spark_df.toPandas()`\n",
    "\n",
    "Create a Spark DataFrame from Pandas\n",
    "\n",
    "`spark_df = context.createDataFrame(pandas_df)`\n",
    "\n",
    "Must fit in memory.\n",
    "\n",
    "![](https://ogirardot.files.wordpress.com/2015/05/rdd-vs-dataframe.png?w=640&h=360)\n",
    "\n",
    "VERY IMPORTANT: DataFrames in Spark are like RDD in the sense that they’re an immutable data structure."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Gender</th>\n",
       "      <th>Height</th>\n",
       "      <th>Weight</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Male</td>\n",
       "      <td>73.847017</td>\n",
       "      <td>241.893563</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Male</td>\n",
       "      <td>68.781904</td>\n",
       "      <td>162.310473</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Male</td>\n",
       "      <td>74.110105</td>\n",
       "      <td>212.740856</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Male</td>\n",
       "      <td>71.730978</td>\n",
       "      <td>220.042470</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>Male</td>\n",
       "      <td>69.881796</td>\n",
       "      <td>206.349801</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  Gender     Height      Weight\n",
       "0   Male  73.847017  241.893563\n",
       "1   Male  68.781904  162.310473\n",
       "2   Male  74.110105  212.740856\n",
       "3   Male  71.730978  220.042470\n",
       "4   Male  69.881796  206.349801"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df=pd.read_csv(\"https://dl.dropboxusercontent.com/u/75194/stats/data/01_heights_weights_genders.csv\")\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[Gender: string, Height: double, Weight: double]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlsc=SQLContext(sc)\n",
    "sparkdf = sqlsc.createDataFrame(df)\n",
    "sparkdf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----------------+----------------+\n",
      "|Gender|           Height|          Weight|\n",
      "+------+-----------------+----------------+\n",
      "|  Male|  73.847017017515|241.893563180437|\n",
      "|  Male|68.78190404589029|  162.3104725213|\n",
      "|  Male|74.11010539178491|  212.7408555565|\n",
      "|  Male| 71.7309784033377|220.042470303077|\n",
      "|  Male| 69.8817958611153|206.349800623871|\n",
      "+------+-----------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sparkdf.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.column.Column"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(sparkdf.Gender)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.rdd.PipelinedRDD'>\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[u'Male',\n",
       " u'Male',\n",
       " u'Male',\n",
       " u'Male',\n",
       " u'Male',\n",
       " u'Male',\n",
       " u'Male',\n",
       " u'Male',\n",
       " u'Male',\n",
       " u'Male']"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "temp = sparkdf.map(lambda r: r.Gender)\n",
    "print type(temp)\n",
    "temp.take(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###Machine Learning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[LabeledPoint(1.0, [73.8470170175,241.89356318]),\n",
       " LabeledPoint(1.0, [68.7819040459,162.310472521]),\n",
       " LabeledPoint(1.0, [74.1101053918,212.740855557]),\n",
       " LabeledPoint(1.0, [71.7309784033,220.042470303]),\n",
       " LabeledPoint(1.0, [69.8817958611,206.349800624])]"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.mllib.classification import LogisticRegressionWithLBFGS\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "\n",
    "data=sparkdf.map(lambda row: LabeledPoint(row.Gender=='Male',[row.Height, row.Weight]))\n",
    "data.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(1.0, DenseVector([73.847, 241.8936]))"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data2=sparkdf.map(lambda row: LabeledPoint(row[0]=='Male',row[1:]))\n",
    "data2.take(1)[0].label, data2.take(1)[0].features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[76] at RDD at PythonRDD.scala:43"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train, test = data.randomSplit([0.7,0.3])\n",
    "train.cache()\n",
    "test.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.PipelinedRDD"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "model = LogisticRegressionWithLBFGS.train(train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DenseVector([-0.2554, 0.1052])"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "model.weights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(1.0, 1.0), (1.0, 1.0), (1.0, 1.0), (1.0, 0.0), (1.0, 1.0), (1.0, 1.0), (1.0, 1.0), (1.0, 1.0), (1.0, 1.0), (1.0, 1.0)]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.PipelinedRDD"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "results = test.map(lambda lp: (lp.label, float(model.predict(lp.features))))\n",
    "print results.take(10)\n",
    "type(results)                       "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.92220367278798"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_accuracy=results.filter(lambda (a,p): a==p).count()/float(results.count())\n",
    "test_accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.evaluation import BinaryClassificationMetrics\n",
    "metrics = BinaryClassificationMetrics(results)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.mllib.evaluation.BinaryClassificationMetrics'>\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "0.9221911239964319"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "print type(metrics)\n",
    "metrics.areaUnderROC"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.mllib.classification.LogisticRegressionModel"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(model)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!rm -rf mylogistic.model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "model.save(sc, \"mylogistic.model\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The pipeline API automates a lot of this stuff, allowing us to work directly on dataframes. It is not all supported in Python, as yet. More of it in the HW."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Also see:\n",
    "\n",
    "- http://jordicasanellas.weebly.com/data-science-blog/machine-learning-with-spark\n",
    "- http://spark.apache.org/docs/latest/mllib-guide.html\n",
    "- http://www.techpoweredmath.com/spark-dataframes-mllib-tutorial/\n",
    "- http://spark.apache.org/docs/latest/api/python/\n",
    "- http://spark.apache.org/docs/latest/programming-guide.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`rdd.saveAsTextFile()` saves an RDD as a string."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}