2017-09-14-PySpark-Workshop.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "<img src=\"images/sads-logo.jpeg\" align=\"left\" width=\"22%\">\n",
    "<img src=\"images/spark-logo-hd.png\" align=\"right\" width=\"28%\">\n",
    "<img src=\"images/python-logo-notext.png\" align=\"right\" width=\"19%\">\n",
    "\n",
    "<h1 align='center'>Intro to PySpark Workshop</h1>\n",
    "<h3 align='center'>Meghann Agarwal<br/>September 14, 2017</h3>\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Bottlenecks and Blockers\n",
    "#### CPU Bound\n",
    "    - Optimize algorithms\n",
    "    - Optimize compilation\n",
    "    - Parallelize code\n",
    "#### Storage Bound\n",
    "    - Store data elsewhere and transfer to compute machine when needed\n",
    "    - Spread data across multiple machines and run computations on all machines\n",
    "#### Network I/O Bound\n",
    "    - Decrease network transfer time\n",
    "        - Compress data for faster transfer\n",
    "        - Move computation to the data\n",
    "    - Consider multithreading\n",
    "        - Unblock code by allowing data to transfer in a separate thread\n",
    "#### Disk I/O  Bound\n",
    "    - Cache data in RAM for faster access"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# MapReduce"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## What is MapReduce?\n",
    "*Material in the slides on MapReduce taken from Chapter 24: MapReduce in \"Data Science from Scratch\" by J. Grus*\n",
    "\n",
    "Basic version of `MapReduce` algorithm:\n",
    "1. Use a `mapper` function to turn each item into zero or more key-value\n",
    "pairs.\n",
    "2. Collect together all the pairs with identical keys.\n",
    "3. Use a `reducer` function on each collection of grouped values to produce output values for the corresponding key.\n",
    "\n",
    "MapReduce paper: <a href=\"https://research.google.com/archive/mapreduce.html\" target=\"_blank\">https://research.google.com/archive/mapreduce.html</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "## Why MapReduce?\n",
    "\n",
    ">\"The primary benefit of MapReduce is that it allows us to distribute computations by moving the processing to the data.\"  \n",
    "-- J. Grus, *Data Science from Scratch*  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "\"The key contributions of the MapReduce framework are not the actual map and reduce functions (which, for example, resemble the 1995 Message Passing Interface standard's reduce and scatter operations), but the scalability and fault-tolerance achieved for a variety of applications by optimizing the execution engine.\"  \n",
    "-- Wikipedia"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Example: Word Count\n",
    "\"Data Science from Scratch\", p. 289\n",
    "\n",
    "Code: <a href=\"https://github.com/joelgrus/data-science-from-scratch/blob/master/code/mapreduce.py#L1-L59\" target=\"_blank\">https://github.com/joelgrus/data-science-from-scratch/blob/master/code/mapreduce.py#L1-L59</a>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "# Load a document\n",
    "documents = open('./data/The Jungle Book, by Rudyard Kipling.htm')\n",
    "doctext = documents.readlines()\n",
    "documents.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "import re\n",
    "\n",
    "def tokenize(message):\n",
    "    \"\"\"returns a list of distinct words contained in `message`\"\"\"\n",
    "    message = message.lower()                       # convert to lowercase\n",
    "    all_words = re.findall(\"[a-z0-9']+\", message)   # extract the words\n",
    "    return set(all_words)                           # remove duplicates"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "def wc_mapper(document):\n",
    "    \"\"\"for each word in document, emit (word,1)\"\"\"\n",
    "    for word in tokenize(document):\n",
    "        yield (word, 1)\n",
    "        \n",
    "def wc_reducer(word, counts):\n",
    "    \"\"\"sum up the counts for a word\"\"\"\n",
    "    yield (word, sum(counts))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "from collections import defaultdict\n",
    "\n",
    "def map_reduce(inputs, mapper, reducer):\n",
    "    \"\"\"runs MapReduce on the inputs using mapper and reducer\"\"\"\n",
    "    collector = defaultdict(list)\n",
    "    \n",
    "    for input in inputs:\n",
    "        for key, value in mapper(input):\n",
    "            collector[key].append(value)\n",
    "            \n",
    "    return [output\n",
    "           for key, values in collector.items()\n",
    "           for output in reducer(key, values)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "word_counts = map_reduce(doctext, wc_mapper, wc_reducer)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('the', 2534)\n",
      "('and', 1832)\n",
      "('p', 1831)\n",
      "('of', 1185)\n",
      "('to', 1123)\n",
      "('a', 1059)\n",
      "('rdquo', 1027)\n",
      "('he', 914)\n",
      "('ldquo', 819)\n",
      "('in', 704)\n",
      "('that', 650)\n",
      "('his', 577)\n",
      "('rsquo', 548)\n",
      "('i', 534)\n",
      "('was', 487)\n",
      "('for', 446)\n",
      "('said', 430)\n",
      "('is', 417)\n",
      "('with', 410)\n",
      "('it', 378)\n"
     ]
    }
   ],
   "source": [
    "word_counts.sort(key=lambda tup: tup[1], reverse=True)\n",
    "\n",
    "# Print the first `num_lines_to_print` word counts\n",
    "num_lines_to_print = 20\n",
    "[print(x) for i, x in enumerate(word_counts) if i < num_lines_to_print];"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Example: Analyzing Status Updates\n",
    "\"Data Science from Scratch\", p. 293\n",
    "\n",
    "Code: <a href=\"https://github.com/joelgrus/data-science-from-scratch/blob/master/code/mapreduce.py#L61-L112\" target=\"_blank\">https://github.com/joelgrus/data-science-from-scratch/blob/master/code/mapreduce.py#L61-L112</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "### Example: Matrix Multiplier\n",
    "\"Data Science from Scratch\", p. 294\n",
    "\n",
    "Code: <a href=\"https://github.com/joelgrus/data-science-from-scratch/blob/master/code/mapreduce.py#L115-L187\" target=\"_blank\">https://github.com/joelgrus/data-science-from-scratch/blob/master/code/mapreduce.py#L115-L187</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Hadoop"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## What is Hadoop?\n",
    "Hadoop is a system (framework, infrastructure) consisting of the Hadoop Distributed File System (HDFS), an API for writing a MapReduce program, etc.\n",
    "\n",
    "- **Hadoop Common** – contains libraries and utilities needed by other Hadoop modules;\n",
    "- **Hadoop Distributed File System (HDFS)** – a distributed file-system that stores data on commodity machines, providing very high aggregate bandwidth across the cluster;\n",
    "- **Hadoop YARN** – a platform responsible for managing computing resources in clusters and using them for scheduling users' applications; and\n",
    "- **Hadoop MapReduce** – an implementation of the MapReduce programming model for large-scale data processing.\n",
    "\n",
    "Hadoop Cluster = Hadoop Distributed File System (HDFS) + Yet Another Resource Negotiator (YARN)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "## Why Hadoop?\n",
    "History of Hadoop: <a href=\"https://medium.com/@markobonaci/the-history-of-hadoop-68984a11704\" target=\"_blank\">https://medium.com/@markobonaci/the-history-of-hadoop-68984a11704</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "- [MapReduce](https://en.wikipedia.org/wiki/MapReduce) --> [Hadoop](http://hadoop.apache.org/): heavily used in business computing\n",
    "- [Message Passing Interface (MPI)](https://en.wikipedia.org/wiki/Message_Passing_Interface) --> [MVAPICH](http://mvapich.cse.ohio-state.edu/): heavily used in scientific computing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "- [Spark](http://spark.apache.org/): complement to Hadoop, faster for iterative applications, rich set of APIs in Scala, Python, and Java, and an interactive shell"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# PySpark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Installation\n",
    "\n",
    "### Databricks Community Edition\n",
    "1. Use Google Chrome browser (Firefox should also work, but not Internet Explorer, Safari, etc.)\n",
    "2. Sign up for a free Community Edition account here: https://databricks.com/try-databricks"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Local on Mac OSX\n",
    "1. Download http://apache.spinellicreations.com/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz\n",
    "2. Extract the Spark package and create `SPARK_HOME`\n",
    "```\n",
    "tar -xvzf spark-2.2.0-bin-hadoop2.7.tgz\n",
    "sudo mv spark-2.2.0-bin-hadoop2.7 /opt/spark\n",
    "export SPARK_HOME=/opt/spark\n",
    "export PATH=$SPARK_HOME/bin:$PATH\n",
    "```\n",
    "3. Run the included Pi Estimator example by executing the following command:\n",
    "```\n",
    "$SPARK_HOME/bin/run-example SparkPi 10\n",
    "```\n",
    "Expect to see something like:\n",
    "```\n",
    "Pi is roughly 3.140576\n",
    "```\n",
    "\n",
    "### Amazon Web Services (AWS)\n",
    "1. Deploy Spark on Elastic Cloud Compute (EC2) cluster\n",
    "2. Deploy Spark on Elastic MapReduce (EMR) cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## What is Spark?\n",
    "Spark is a fast and expressive cluster computing system for doing Big Data computation. It's good for:\n",
    "- iterative tasks\n",
    "- doing big batch processing\n",
    "- interactive data exploration\n",
    "\n",
    "It's compatible with Hadoop-supported file systems and data formats (HDFS, S3, SequenceFile, ...), so if you've been using Hadoop you can use it with your existing data and deploy it on your existing clusters.\n",
    "\n",
    "It achieves fault tolerance through *lineage*: if you lose a partition (chunk) of data you can reconstruct it through a set of *transformations* that act on data stored in memory. This is in contrast to distributed shared memory systems where you have to write to disk and roll back."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Why Spark?\n",
    "https://spark.apache.org/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "<img src=\"images/speed.png\" align=\"left\" width=\"100%\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "<img src=\"images/ease-of-use.png\" align=\"left\" width=\"100%\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<img src=\"images/generality.png\" align=\"left\" width=\"100%\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "<img src=\"images/runs-everywhere.png\" align=\"left\" width=\"100%\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "\n",
    ">\"Although current frameworks provide numerous abstractions for accessing a cluster’s computational resources, they <font color='red'> lack abstractions for leveraging distributed memory</font>. This makes them inefficient for an important class of emerging applications: those that reuse intermediate results across multiple computations. <font color='blue'>Data reuse is common in many iterative machine learning and graph algorithms</font>, including PageRank, K-means clustering, and logistic regression. <font color='blue'>Another compelling use case is interactive data mining</font>, where a user runs multiple ad-hoc queries on the same subset of the data. Unfortunately, in most current frameworks, the only way to reuse data between computations (e.g., between two MapReduce jobs) is to <font color='red'>write it to an external stable storage system</font>, e.g., a distributed file system. This incurs substantial overheads due to data replication, disk I/O, and serialization, which can dominate application execution times.\"  \n",
    "-- Zaharia et al., \"Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing,\" *In NSDI '12*, April 2012\n",
    "\n",
    "https://cs.stanford.edu/~matei/papers/2012/nsdi_spark.pdf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Spark Architecture\n",
    "- Spark Driver and Workers\n",
    "- Spark applications run as independent sets of processes on a cluster, coordinated by the SparkContext object in your main program (called the driver program).\n",
    "\n",
    "<img src=\"images/cluster-overview.png\" align=\"center\" width=\"75%\">\n",
    "\n",
    "<a href=\"https://spark.apache.org/docs/latest/cluster-overview.html\" target=\"_blank\">https://spark.apache.org/docs/latest/cluster-overview.html</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "- Standalone\n",
    "- Hadoop YARN\n",
    "- Apache Mesos\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Spark Programming Concepts\n",
    "- ****SparkContext****: main entry point to Spark functionality\n",
    "- ****SparkSession****: main entry point for DataFrame and SQL functionality\n",
    "- ****Resilient Distributed Datasets (RDDs)****:\n",
    "    - Immutable, distributed collections of objects\n",
    "    - Can be cached in memory for fast reuse\n",
    "- ****Operations on RDDs****:\n",
    "    - *Transformations*: define a new RDD (map, join, ...)\n",
    "    - *Actions*: return or output a result (count, save, ...)\n",
    "- ****Two ways to create RDDs****:\n",
    "    1. By parallelizing an existing collection in your driver program:  \n",
    "        ```\n",
    "        data = [1, 2, 3, 4, 5]  \n",
    "        dist_data = sc.parallelize(data)\n",
    "        ```\n",
    "    2. Or by referencing a dataset in an external storage system, such as a shared filesystem, HDFS, HBase, or any data source offering a Hadoop InputFormat:  \n",
    "        ```\n",
    "        dist_file = sc.textFile(\"data.txt\")\n",
    "        ```\n",
    "<h4 align='right'>http://spark.apache.org/docs/latest/programming-guide.html</h4>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Spark Data Interfaces\n",
    "\n",
    "- [RDD API](https://spark.apache.org/docs/latest/rdd-programming-guide.html#resilient-distributed-datasets-rdds)\n",
    "- [DataFrame API](https://spark.apache.org/docs/latest/sql-programming-guide.html#dataframes)\n",
    "- [Machine Learning API](https://spark.apache.org/docs/latest/mllib-guide.html)\n",
    "\n",
    "\n",
    "****RDD (Resilient Distributed Dataset)****\n",
    "-   The basic abstraction in Spark. Essentially it is an interface to a sequence of data objects that consist of one or more types that are located across a variety of machines in a cluster. RDD's can be created in a variety of ways and are the \"lowest level\" API available to the user.\n",
    "\n",
    "****DataFrame****\n",
    "-   The DataFrame is collection of distributed `Row` types. These provide a flexible interface and are similar in concept to the DataFrames you may be familiar with in python (pandas) as well as in the R language.\n",
    "\n",
    "*\"Introduction to Apache Spark on Databricks\" notebook*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## What is PySpark?\n",
    "- The Python API for Spark\n",
    "- Run interactive jobs in the shell\n",
    "- Supports numpy, pandas and other Python libraries"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "## Why use PySpark?\n",
    "- If you already know Python\n",
    "- Can use Spark in tandem with your favorite Python libraries\n",
    "- If you don't need Python libraries, maybe just write code in Scala"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### PySpark's Core Classes\n",
    "- ****pyspark.SparkContext****  \n",
    "Main entry point for Spark functionality.\n",
    "- ****pyspark.RDD****  \n",
    "A Resilient Distributed Dataset (RDD), the basic abstraction in Spark.\n",
    "- ****pyspark.streaming.StreamingContext****  \n",
    "Main entry point for Spark Streaming functionality.\n",
    "- ****pyspark.streaming.DStream****  \n",
    "A Discretized Stream (DStream), the basic abstraction in Spark Streaming.\n",
    "- ****pyspark.sql.SQLContext****  \n",
    "Main entry point for DataFrame and SQL functionality.\n",
    "- ****pyspark.sql.DataFrame****  \n",
    "A distributed collection of data grouped into named columns.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Transformations\n",
    "- Transform one RDD to another, **new** RDD (immutable)\n",
    "\n",
    "| Transformation | Description | Type |\n",
    "| :------:  | :-----------: | :-----: |\n",
    "| `map(func)`     | Apply a function over each element | Narrow |\n",
    "| `flatMap(func)` | Map then flatten output | Narrow |\n",
    "| `filter(func)`  | Keep only elements where function is `True` | Narrow |\n",
    "| `sample(withReplacement, fraction, seed)` | Return a sampled subset of this RDD | Narrow |\n",
    "| `groupByKey(k, v)` | Group the values for each key in the RDD into a single sequence | Wide |\n",
    "| `reduceByKey(func)` | Merge the values for each key using an associative reduce function | Wide |\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<img src=\"images/narrow_wide_transformations.png\" align=\"center\" width=\"90%\">\n",
    "\n",
    "<h4 align='right'>https://dzone.com/articles/big-data-processing-spark</h4>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Actions\n",
    "- Return or output a result\n",
    "\n",
    "| Action | Description | Try it Out\\*|\n",
    "| :------:  | :-----------:| :---: |\n",
    "| `collect()`     | Return a list that contains all of the elements in this RDD | `sc.parallelize([0, 2, 3, 4, 6], 5).glom().collect()` |\n",
    "| `count()`  | Return the number of elements | `sc.parallelize([2, 3, 4]).count()` |\n",
    "| `saveAsTextFile(path)` | Save as a text file, using string representations of elements | `sc.parallelize(['foo', '-', 'bar', '!']).saveAsTextFile(\"/FileStore/foo-bar.txt\")])`|\n",
    "| `first()`    | Return the first element | `sc.parallelize([2, 3, 4]).first()` |\n",
    "| `take(num)`    | Take the first num elements | `sc.parallelize([2, 3, 4, 5, 6]).take(2)` |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### \\* Try it Out:\n",
    "1. Go to your databricks Workspace and create a new directory within your Users directory called \"2017-09-14-sads-pyspark\" \n",
    "2. Create a notebook called \"0-Introduction\"  within this directory\n",
    "3. Type or copy/paste lines of code into separate cells and run them (you will be prompted to launch a cluster) \n",
    "\n",
    "Here are some cheat sheets you might find helpful:\n",
    "- <a href=\"https://s3.amazonaws.com/assets.datacamp.com/blog_assets/PySpark_Cheat_Sheet_Python.pdf\" target=\"_blank\">PySpark_Cheat_Sheet_Python.pdf</a>\n",
    "- <a href=\"https://s3.amazonaws.com/assets.datacamp.com/blog_assets/PySpark_SQL_Cheat_Sheet_Python.pdf\" target=\"_blank\">PySpark_SQL_Cheat_Sheet_Python.pdf</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "When using Databricks the `SparkContext` is created for you automatically as `sc`.\n",
    "\n",
    "In the Databricks Community Edition there are no Worker Nodes - the Driver Program (Master) executes the entire code."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### More examples to try:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\", 1), (\"b\", 1), (\"a\", 1)])\n",
    "sorted(rdd.groupByKey().mapValues(len).collect())\n",
    "sorted(rdd.groupByKey().mapValues(list).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from operator import add\n",
    "\n",
    "rdd = sc.parallelize([(\"a\", 1), (\"b\", 1), (\"a\", 1)])\n",
    "sorted(rdd.reduceByKey(add).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "Complete documentation of available *transformations* and *actions*: <a href=\"http://spark.apache.org/docs/latest/api/python/pyspark.html\" target=\"_blank\">http://spark.apache.org/docs/latest/api/python/pyspark.html</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Log Mining Code (Scala)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "-"
    }
   },
   "outputs": [],
   "source": [
    "val lines = spark.textFile(\"hdfs://...\")\n",
    "val errors = lines.filter(_.startsWith(\"ERROR\"))\n",
    "val messages = errors.map(_.split('\\t')(2))\n",
    "\n",
    "messages.filter(_.contains(\"foo\")).count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "The computation is expressed declaratively and nothing actually takes place until calling `count` at the end."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Exercises"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Exercise 1: Word Count\n",
    "Create a few transformations to build a dataset of (String, Int) pairs called counts and then save it to a file. Or, upload \"The Jungle Book, by Rudyard Kipling.htm\".\n",
    "\n",
    "1. Create a notebook in \"2017-09-14-sads-pyspark\" called \"1-WordCount\"\n",
    "2. Try to implement the following Word Count example:\n",
    "\n",
    "http://spark.apache.org/examples.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "text_file = sc.textFile(\"hdfs://...\")\n",
    "counts = text_file.flatMap(lambda line: line.split(\" \")) \\\n",
    "             .map(lambda word: (word, 1)) \\\n",
    "             .reduceByKey(lambda a, b: a + b)\n",
    "counts.saveAsTextFile(\"hdfs://...\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## Exercise 2: Clickstream\n",
    "\n",
    "***Switch over to Databricks*** Import notebook \"2-Clickstream\" in to \"2017-09-14-sads-pyspark\"\n",
    "\n",
    "https://databricks-prod-cloudfront.cloud.databricks.com/public/4027ec902e239c93eaaa8714f173bcfc/8253322025282036/502654908621835/2929138047951629/latest.html\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "***Data Source: February 2015 English Wikipedia Clickstream dataset***\n",
    "- https://meta.wikimedia.org/wiki/Research:Wikipedia_clickstream\n",
    "- http://datahub.io/dataset/wikipedia-clickstream/resource/be85cc68-d1e6-4134-804a-fd36b94dbb82. \n",
    "\n",
    ">\"The data contains counts of (referer, resource) pairs extracted from the request logs of English Wikipedia. When a client requests a resource by following a link or performing a search, the URI of the webpage that linked to the resource is included with the request in an HTTP header called the \"referer\". This data captures 22 million (referer, resource) pairs from a total of 3.2 billion requests collected during the month of February 2015.\"\n",
    "\n",
    "The data is approximately 1.2GB and it is hosted in the following Databricks file: `/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed`\n",
    "\n",
    "*Notebook translated from the Databricks \"Quick Start DataFrames\" tutorial in Scala, which was based on a lab developed by [Sameer Farooqui](https://www.linkedin.com/in/blueplastic).*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Exercise 3: Logistic Regression\n",
    "\n",
    "1. Create a notebook in \"2016-09-14-sads-pyspark\" called \"3-LogisticRegression\"\n",
    "2. Try to implement one of the following Logistic Regression examples:\n",
    "    - http://spark.apache.org/examples.html (Prediction with Logistic Regression)\n",
    "    - https://github.com/apache/spark/blob/master/examples/src/main/python/mllib/logistic_regression.py\n",
    "    - https://github.com/apache/spark/blob/master/examples/src/main/python/logistic_regression.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "# Every record of this DataFrame contains the label and\n",
    "# features represented by a vector.\n",
    "df = sqlContext.createDataFrame(data, [\"label\", \"features\"])\n",
    "\n",
    "# Set parameters for the algorithm.\n",
    "# Here, we limit the number of iterations to 10.\n",
    "lr = LogisticRegression(maxIter=10)\n",
    "\n",
    "# Fit the model to the data.\n",
    "model = lr.fit(df)\n",
    "\n",
    "# Given a dataset, predict each point's label, and show the results.\n",
    "model.transform(df).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Resources and References\n",
    "- ****MOOCs****:\n",
    "    - \"Introduction to Apache Spark\": https://www.edx.org/course/introduction-apache-spark-uc-berkeleyx-cs105x\n",
    "    - \"Hadoop Platform and Application Framework\" (week 5 covers Spark): https://www.coursera.org/learn/hadoop/home/week/5\n",
    "- ****Spark/PySpark Docs****:\n",
    "    - https://spark.apache.org/docs/latest/\n",
    "    - http://spark.apache.org/research.html\n",
    "    - http://spark.apache.org/examples.html\n",
    "- ****Other****:\n",
    "    - Spark 2.0 Webinar (2016): http://go.databricks.com/apache-spark-2.0-presented-by-databricks-co-founder-reynold-xin\n",
    "    - PySpark Talk (J. Rosen, 2013): https://www.youtube.com/watch?v=xc7Lc8RA8wE\n",
    "    - \"Apache Spark in 24 Hours\": https://www.amazon.com/Apache-Spark-Hours-Teach-Yourself/dp/0672338513"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "# Thanks for Coming!\n",
    "\n",
    "Ways to get in touch:\n",
    "- <agarwal.meghann@gmail.com>\n",
    "- https://www.linkedin.com/in/meghann-agarwal-7a044b4/"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "celltoolbar": "Slideshow",
  "kernelspec": {
   "display_name": "Python [conda root]",
   "language": "python",
   "name": "conda-root-py"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}