Jupyter/M04-DataManipulation/M04F-Spark.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IRGVykDGkqcH"
      },
      "source": [
        "![Cloud-First](https://github.com/tulip-lab/sit742/blob/develop/Jupyter/image/CloudFirst.png?raw=1)\n",
        "\n",
        "\n",
        "# SIT742: Modern Data Science\n",
        "**(Module: Big Data Manipulation)**\n",
        "\n",
        "---\n",
        "- Materials in this module include resources collected from various open-source online repositories.\n",
        "- You are free to use, change and distribute this package.\n",
        "- If you found any issue/bug for this document, please submit an issue at [tulip-lab/sit742](https://github.com/tulip-lab/sit742/issues)\n",
        "\n",
        "\n",
        "Prepared by **SIT742 Teaching Team**\n",
        "\n",
        "---\n",
        "\n",
        "\n",
        "\n",
        "## Session 4F: A Touch of Apache Spark\n",
        "---\n",
        "\n",
        "This lab session will teach you how to use [Apache Spark](http://spark.apache.org/), a framework for large-scale data processing, within a notebook.\n",
        "\n",
        "You may need to do the following before running this notebook in Cloud platform such as Google Colab, DataBricks, or IBM Cloud:\n",
        "- Create one cluster to run your tasks and the following lab sessions\n",
        "- Install the necessary packages such as\n",
        " * matplotlib\n",
        " * test_helper\n",
        " * pyzmq\n",
        " * pandas\n",
        " * hashlib\n",
        "- Import all related lab sessions into Workspace\n",
        "- Import all needed data into Tables or S3, and note down the address of each file.\n",
        "\n",
        "\n",
        "### Content\n",
        "\n",
        "Part A: Apache Spark\n",
        "\n",
        "* [PySpark](#pyspark)\n",
        "\n",
        "* [RDD Transformations and Actions](#rdd)\n",
        "\n",
        "* [Lambda functions](#lambda)\n",
        "\n",
        "* [Additional RDD actions](#actions)\n",
        "\n",
        "* [Additional RDD transformations](#transfoormations)\n",
        "\n",
        "* [Caching RDDs and storage options](#cache)\n",
        "\n",
        "* [Debugging Spark applications and lazy evaluation](#debug)\n",
        "\n",
        "Part B: Test Spark functionality\n",
        "\n",
        "* [Check Spark Functionality](#spark)\n",
        "\n",
        "* [Check class Testing Functionality](#test)\n",
        "\n",
        "* [Check Plotting](#plot)\n",
        "---\n",
        "\n",
        "\n",
        "## Introduction ##\n",
        "\n",
        "Many traditional frameworks were designed to be run on a single computer. However, many datasets today are too large to be stored on a single computer, and even when a dataset can be stored on one computer (such as the datasets in this tutorial), the dataset can often be processed much more quickly using multiple computers.  Spark has efficient implementations of a number of transformations and actions that can be composed together to perform data processing and analysis.  Spark excels at distributing these operations across a cluster while abstracting away many of the underlying implementation details. Spark has been designed with a focus on scalability and efficiency. With Spark you can begin developing your solution on your laptop, using a small dataset, and then use that same code to process terabytes or even petabytes across a distributed cluster.\n",
        "\n",
        "The following transformations will be covered:\n",
        "*  `map()`, `mapPartitions()`, `mapPartitionsWithIndex()`, `filter()`, `flatMap()`, `sortByKey()`, `reduceByKey()`, `groupByKey()`, `Join()`, `leftOuterJoin()`, `rightOuterJoin()`, `fullOuterJoin()`\n",
        "\n",
        "The following actions will be covered:\n",
        "*  `first()`, `take()`, `takeSample()`, `takeOrdered()`, `collect()`, `count()`, `countByValue()`, `reduce()`, `fold()`, `aggregate()`, `top()`\n",
        "\n",
        "Also covered:\n",
        "*  `cache()`, `unpersist()`, `id()`, `setName()`\n",
        "\n",
        "Note that, for reference, you can look up the details of these methods in [Spark's Python API](https://spark.apache.org/docs/latest/api/python/reference/pyspark.html)\n",
        "\n",
        "After that you will test that the virtual machine (VM) is functioning properly.\n",
        "\n",
        "You will not need to solve any problems to complete this lab.  You can run a cell by pressing \"shift-enter\", which will compute the current cell and advance to the next cell, or by clicking in a cell and pressing \"control-enter\", which will compute the current cell and remain in that cell.  "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jeyia0Bdkqcf"
      },
      "source": [
        "## Part A: Apache Spark ##\n",
        "\n",
        "\n",
        "###  **1. Spark Context**\n",
        "\n",
        "In Spark, communication occurs between a driver and executors.  The driver has Spark jobs that it needs to run and these jobs are split into tasks that are submitted to the executors for completion.  The results from these tasks are delivered back to the driver.\n",
        "\n",
        "In section 1, we saw that normal python code can be executed via cells. When using Databricks Cloud this code gets executed in the Spark driver's Java Virtual Machine (JVM) and not in an executor's JVM, and when using an IPython notebook it is executed within the kernel associated with the notebook. Since no Spark functionality is actually being used, no tasks are launched on the executors.\n",
        "\n",
        "In order to use Spark functionality and its API we will need to use a `SparkContext`.  When running Spark, you start a new Spark application by creating a [SparkContext](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkContext.html).  When the `SparkContext` is created, it asks the master for some cores to use to do work.  The master sets these cores aside just for you; they won't be used for other applications. When using Databricks Cloud or the virtual machine provisioned for this class, the `SparkContext` is created for you automatically as `sc`.\n",
        "\n",
        "###  **2. [pySpark API](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.html) running in the browser**\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "j6SVZY_Qkqcg"
      },
      "source": [
        "#### **(2a) Example Cluster**\n",
        "\n",
        "The diagram below shows an example cluster, where the cores allocated for an application are outlined in purple.\n",
        "![executors](http://spark-mooc.github.io/web-assets/images/executors.png)\n",
        "\n",
        "At a high level, every Spark application consists of a driver program that launches various parallel operations on executor Java Virtual Machines (JVMs) running either in a cluster or locally on the same machine. When running a cluster, such as ...In Databricks Cloud, \"Databricks Shell\" is the driver program.  When running locally, \"PySparkShell\" is the driver program. In all cases, this driver program contains the main loop for the program and creates distributed datasets on the cluster, then applies operations (transformations & actions) to those datasets.\n",
        "\n",
        "Driver programs access Spark through a SparkContext object, which represents a connection to a computing cluster. A Spark context object (`sc`) is the main entry point for Spark functionality. A Spark context can be used to create Resilient Distributed Datasets (RDDs) on a cluster.\n",
        "\n",
        "**Note**: It is important for you to change the path/URL of the latest Spark package, so that it is working with latest Spark release."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "uy4uE0yJkqcm"
      },
      "outputs": [],
      "source": [
        "#update local version of the package catalog\n",
        "!apt-get update\n",
        "# install Java8\n",
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n",
        "# download spark 3.3.3\n",
        "!wget -q https://archive.apache.org/dist/spark/spark-3.3.3/spark-3.3.3-bin-hadoop3.tgz\n",
        "# unzip it\n",
        "!tar xf spark-3.3.3-bin-hadoop3.tgz\n",
        "# install findspark\n",
        "!pip install -q findspark\n",
        "\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark-3.3.3-bin-hadoop3\"\n",
        "\n",
        "import findspark\n",
        "findspark.init()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "1oWoAf7Hkqco"
      },
      "outputs": [],
      "source": [
        "from pyspark import SparkContext\n",
        "sc = SparkContext.getOrCreate()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "2S9OTCPfkqcq"
      },
      "source": [
        "##### ** Predefined Variables**\n",
        "\n",
        "\n",
        "Notebooks already have some of the most useful Apache Spark variables that you’re going to need:\n",
        "\n",
        "- sc\n",
        "- sqlContext\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vaUDDqAYkqct"
      },
      "outputs": [],
      "source": [
        "# A SQLContext is also already created for you.\n",
        "# Do not create another or unspecified behavior may occur.\n",
        "# As you can see below, the sqlContext provided is a HiveContext.\n",
        "#For the difference between the HiveContext and sqlContext, please refer the URL\n",
        "#https://stackoverflow.com/questions/33666545/what-is-the-difference-between-apache-spark-sqlcontext-vs-hivecontext\n",
        "\n",
        "from pyspark.sql import SQLContext\n",
        "sqlContext = SQLContext(sc)\n",
        "sqlContext"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "hRf03Eumkqcu"
      },
      "outputs": [],
      "source": [
        "# A Spark Context is already created for you.\n",
        "# Do not create another or unspecified behavior may occur.\n",
        "#The sc command will show SparkContext Infomation including the Version, Master name and APP\n",
        "sc"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "X4yfJp9xkqcy"
      },
      "outputs": [],
      "source": [
        "# Run this to check Spark Context sc is already loded\n",
        "# Display the type of the Spark Context sc\n",
        "print(\"The sc type is \",type(sc),\"\\nThe sqlContext type is \",type(sqlContext))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "oJk5EKYUkqc0"
      },
      "source": [
        "#### **(2b) `SparkContext` attributes**\n",
        "\n",
        "You can use Python's [dir()](https://docs.python.org/2/library/functions.html?highlight=dir#dir) function to get a list of all the attributes (including methods) accessible through the `sc` object."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ghF8iyoakqc1"
      },
      "outputs": [],
      "source": [
        "# List sc's attributes\n",
        "dir(sc)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7Z7saDfBkqc4"
      },
      "source": [
        "#### **(2c) Getting help**\n",
        "\n",
        "Alternatively, you can use Python's [help()](https://docs.python.org/2/library/functions.html?highlight=help#help) function to get an easier to read list of all the attributes, including examples, that the `sc` object has."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "aIKPJtaEkqc4"
      },
      "outputs": [],
      "source": [
        "# Use help to obtain more detailed information\n",
        "#Please note tha xrange() is replaced by the range() in Python 3\n",
        "help(sc)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5HkAjqO-kqc6"
      },
      "outputs": [],
      "source": [
        "# After reading the help we've decided we want to use sc.version to see what version of Spark we are running\n",
        "sc.version"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "jLzSxcxOkqc-"
      },
      "outputs": [],
      "source": [
        "# Help can be used on any Python object\n",
        "help(map)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "S7gc94wqkqc_"
      },
      "source": [
        "<a id = \"rdd\"></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XgdU_0ltkqc_"
      },
      "source": [
        "### **3. Using RDDs and chaining together transformations and actions**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "aFXb1VlkkqdA"
      },
      "source": [
        "#### **Working with your first RDD**\n",
        "\n",
        "In Spark, we first create a base [Resilient Distributed Dataset](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.html) (RDD). We can then apply one or more transformations to that base RDD. *An RDD is immutable, so once it is created, it cannot be changed.* As a result, each transformation creates a new RDD. Finally, we can apply one or more actions to the RDDs.  Note that Spark uses lazy evaluation, so transformations are not actually executed until an action occurs.\n",
        "\n",
        "We will perform several exercises to obtain a better understanding of RDDs:\n",
        "*   Create a Python collection of 10,000 integers\n",
        "*   Create a Spark base RDD from that collection\n",
        "*   Subtract one from each value using `map`\n",
        "*   Perform action `collect` to view results\n",
        "*   Perform action `count` to view counts\n",
        "*   Apply transformation `filter` and view results with `collect`\n",
        "*   Learn about lambda functions\n",
        "*   Explore how lazy evaluation works and the debugging challenges that it introduces"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wnTwglM8kqdA"
      },
      "source": [
        "#### **(3a) Create a Python collection of integers in the range of 1 .. 10000**\n",
        "\n",
        "In python 3, the xrang() is removed. We will use the [range()](https://docs.python.org/3.8/library/functions.html?highlight=xrange#func-range) function to create a  constructor to represent an immutable sequence of numbers.  `"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6j4GLVx-kqdA"
      },
      "outputs": [],
      "source": [
        "data = range(1, 10001)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "IKlohPf9kqdB"
      },
      "outputs": [],
      "source": [
        "# Data is just a normal Python list\n",
        "# Obtain data's first element\n",
        "data[0]"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "iN7IQeLfkqdD"
      },
      "outputs": [],
      "source": [
        "# We can check the size of the list using the len() function\n",
        "len(data)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "HZUiumDGkqdG"
      },
      "source": [
        "#### **(3b) Distributed data and using a collection to create an RDD**\n",
        "\n",
        "In Spark, datasets are represented as a list of entries, where the list is broken up into many different partitions that are each stored on a different machine.  Each partition holds a unique subset of the entries in the list.  Spark calls datasets that it stores \"Resilient Distributed Datasets\" (RDDs).\n",
        "\n",
        "One of the defining features of Spark, compared to other data analytics frameworks (e.g., Hadoop), is that it stores data in memory rather than on disk.  This allows Spark applications to run much more quickly, because they are not slowed down by needing to read data from disk.\n",
        "\n",
        "The figure below illustrates how Spark breaks a list of data entries into partitions that are each stored in memory on a worker.\n",
        "![partitions](http://spark-mooc.github.io/web-assets/images/partitions.png)\n",
        "\n",
        "To create the RDD, we use `sc.parallelize()`, which tells Spark to create a new set of input data based on data that is passed in.  In this example, we will provide an `range`.  The second argument to the [sc.parallelize()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkContext.parallelize.html) method tells Spark how many partitions to break the data into when it stores the data in memory (we'll talk more about this later in this tutorial). Note that for better performance when using `parallelize`, `range()` is recommended if the input represents a range. This is the reason why we used `range()` in 3a.\n",
        "\n",
        "There are many different types of RDDs.  The base class for RDDs is [pyspark.RDD](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.html) and other RDDs subclass `pyspark.RDD`.  Since the other RDD types inherit from `pyspark.RDD` they have the same APIs and are functionally identical.  We'll see that `sc.parallelize()` generates a `pyspark.rdd.PipelinedRDD` when its input is an `range`, and a `pyspark.RDD` when its input is a `range`.\n",
        "\n",
        "After we generate RDDs, we can view them in the \"Storage\" tab of the web UI.  You'll notice that new datasets are not listed until Spark needs to return a result due to an action being executed.  This feature of Spark is called \"lazy evaluation\".  This allows Spark to avoid performing unnecessary calculations."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "3KOCtxqEkqdH"
      },
      "outputs": [],
      "source": [
        "# Parallelize data including 10000 using 8 partitions\n",
        "# This operation is a transformation of data into an RDD\n",
        "# Spark uses lazy evaluation, so no Spark jobs are run at this point\n",
        "rangeRDD = sc.parallelize(data, 8)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "mwSSlh1jkqdK"
      },
      "outputs": [],
      "source": [
        "# Let's view help on parallelize\n",
        "help(sc.parallelize)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "2Xc9la1KkqdN"
      },
      "outputs": [],
      "source": [
        "# Let's see what type sc.parallelize() returned\n",
        "print ('type of rangeRDD: {0}'.format(type(rangeRDD)))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "WIEf0icEkqdO"
      },
      "outputs": [],
      "source": [
        "# Each RDD gets a unique ID\n",
        "print ('rangeRDD id: {0}'.format(rangeRDD.id()))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Xebne50PkqdP"
      },
      "outputs": [],
      "source": [
        "# We can name each newly created RDD using the setName() method\n",
        "rangeRDD.setName('My first RDD')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "KBh1L9N9kqdR"
      },
      "outputs": [],
      "source": [
        "# Let's view the lineage (the set of transformations) of the RDD using toDebugString()\n",
        "print (rangeRDD.toDebugString())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ToKITYtSkqdS"
      },
      "outputs": [],
      "source": [
        "# Let's use help to see what methods we can call on this RDD\n",
        "help(rangeRDD)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "SbPczx7AkqdT"
      },
      "outputs": [],
      "source": [
        "\n",
        "# Let's see how many partitions the RDD will be split into by using the getNumPartitions()rangeRDD.getNumPartitions()\n",
        "print(rangeRDD.getNumPartitions())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jB70y-RFkqdX"
      },
      "source": [
        "#### **(3c): Subtract one from each value using `map`**\n",
        "\n",
        "So far, we've created a distributed dataset that is split into many partitions, where each partition is stored on a single machine in our cluster.  Let's look at what happens when we do a basic operation on the dataset.  Many useful data analysis operations can be specified as \"do something to each item in the dataset\".  These data-parallel operations are convenient because each item in the dataset can be processed individually: the operation on one entry doesn't effect the operations on any of the other entries.  Therefore, Spark can parallelize the operation.\n",
        "\n",
        "`map(f)`, the most common Spark transformation, is one such example: it applies a function `f` to each item in the dataset, and outputs the resulting dataset.  When you run `map()` on a dataset, a single *stage* of tasks is launched.  A *stage* is a group of tasks that all perform the same computation, but on different input data.  One task is launched for each partitition, as shown in the example below.  A task is a unit of execution that runs on a single machine. When we run `map(f)` within a partition, a new *task* applies `f` to all of the entries in a particular partition, and outputs a new partition. In this example figure, the dataset is broken into four partitions, so four `map()` tasks are launched.\n",
        "![tasks](http://spark-mooc.github.io/web-assets/images/tasks.png)\n",
        "\n",
        "The figure below shows how this would work on the smaller data set from the earlier figures.  Note that one task is launched for each partition.\n",
        "![foo](http://spark-mooc.github.io/web-assets/images/map.png)\n",
        "\n",
        "When applying the `map()` transformation, each item in the parent RDD will map to one element in the new RDD. So, if the parent RDD has twenty elements, the new RDD will also have twenty items.\n",
        "\n",
        "Now we will use `map()` to subtract one from each value in the base RDD we just created. First, we define a Python function called `sub()` that will subtract one from the input integer. Second, we will pass each item in the base RDD into a `map()` transformation that applies the `sub()` function to each element. And finally, we print out the RDD transformation hierarchy using `toDebugString()`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "17VX2BNrkqdX"
      },
      "outputs": [],
      "source": [
        "# Create sub function to subtract 1\n",
        "def sub(value):\n",
        "    \"\"\"\"Subtracts one from `value`.\n",
        "\n",
        "    Args:\n",
        "       value (int): A number.\n",
        "\n",
        "    Returns:\n",
        "        int: `value` minus one.\n",
        "    \"\"\"\n",
        "    return (value - 1)\n",
        "\n",
        "# Transform rangeRDD through map transformation using sub function\n",
        "# Because map is a transformation and Spark uses lazy evaluation, no jobs, stages,\n",
        "# or tasks will be launched when we run this code.\n",
        "subRDD = rangeRDD.map(sub)\n",
        "\n",
        "# Let's see the RDD transformation hierarchy\n",
        "print (subRDD.toDebugString())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "HnbYgr2VkqdY"
      },
      "source": [
        "#### ** (3d) Perform action `collect` to view results **\n",
        "\n",
        "To see a list of elements decremented by one, we need to create a new list on the driver from the the data distributed in the executor nodes.  To do this we call the `collect()` method on our RDD.  `collect()` is often used after a filter or other operation to ensure that we are only returning a *small* amount of data to the driver.  This is done because the data returned to the driver must fit into the driver's available memory.  If not, the driver will crash.\n",
        "\n",
        "The `collect()` method is the first action operation that we have encountered.  Action operations cause Spark to perform the (lazy) transformation operations that are required to compute the RDD returned by the action.  In our example, this means that tasks will now be launched to perform the `parallelize`, `map`, and `collect` operations.\n",
        "\n",
        "In this example, the dataset is broken into four partitions, so four `collect()` tasks are launched. Each task collects the entries in its partition and sends the result to the SparkContext, which creates a list of the values, as shown in the figure below.\n",
        "![collect](http://spark-mooc.github.io/web-assets/images/collect.png)\n",
        "\n",
        "The above figures showed what would happen if we ran `collect()` on a small example dataset with just four partitions.\n",
        "\n",
        "Now let's run `collect()` on `subRDD`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "451ocW9zkqdZ"
      },
      "outputs": [],
      "source": [
        "# Let's collect the data\n",
        "print (subRDD.collect())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qNu116KXkqda"
      },
      "source": [
        "#### ** (3d) Perform action `count` to view counts **\n",
        "\n",
        "One of the most basic jobs that we can run is the `count()` job which will count the number of elements in an RDD using the `count()` action. Since `map()` creates a new RDD with the same number of elements as the starting RDD, we expect that applying `count()` to each RDD will return the same result.\n",
        "\n",
        "Note that because `count()` is an action operation, if we had not already performed an action with `collect()`, then Spark would now perform the transformation operations when we executed `count()`.\n",
        "\n",
        "Each task counts the entries in its partition and sends the result to your SparkContext, which adds up all of the counts. The figure below shows what would happen if we ran `count()` on a small example dataset with just four partitions.\n",
        "![count](http://spark-mooc.github.io/web-assets/images/count.png)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "x8FdMKdikqdb"
      },
      "outputs": [],
      "source": [
        "print (rangeRDD.count())\n",
        "print (subRDD.count())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yTgc_wXskqdc"
      },
      "source": [
        "#### ** (3e) Apply transformation `filter` and view results with `collect` **\n",
        "\n",
        "Next, we'll create a new RDD that only contains the values less than ten by using the `filter(f)` data-parallel operation. The `filter(f)` method is a transformation operation that creates a new RDD from the input RDD by applying filter function `f` to each item in the parent RDD and only passing those elements where the filter function returns `True`. Elements that do not return `True` will be dropped. Like `map()`, filter can be applied individually to each entry in the dataset, so is easily parallelized using Spark.\n",
        "\n",
        "The figure below shows how this would work on the small four-partition dataset.\n",
        "![filter](http://spark-mooc.github.io/web-assets/images/filter.png)\n",
        "\n",
        "To filter this dataset, we'll define a function called `ten()`, which returns `True` if the input is less than 10 and `False` otherwise.  This function will be passed to the `filter()` transformation as the filter function `f`.\n",
        "\n",
        "To view the filtered list of elements less than ten, we need to create a new list on the driver from the distributed data on the executor nodes.  We use the `collect()` method to return a list that contains all of the elements in this filtered RDD to the driver program."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Z9MCpX4Bkqdc"
      },
      "outputs": [],
      "source": [
        "# Define a function to filter a single value\n",
        "def ten(value):\n",
        "    \"\"\"Return whether value is below ten.\n",
        "\n",
        "    Args:\n",
        "        value (int): A number.\n",
        "\n",
        "    Returns:\n",
        "        bool: Whether `value` is less than ten.\n",
        "    \"\"\"\n",
        "    if (value < 10):\n",
        "        return True\n",
        "    else:\n",
        "        return False\n",
        "# The ten function could also be written concisely as: def ten(value): return value < 10\n",
        "\n",
        "# Pass the function ten to the filter transformation\n",
        "# Filter is a transformation so no tasks are run\n",
        "filteredRDD = subRDD.filter(ten)\n",
        "\n",
        "# View the results using collect()\n",
        "# Collect is an action and triggers the filter transformation to run\n",
        "print (filteredRDD.collect())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "WfDLwGB_kqdd"
      },
      "outputs": [],
      "source": [
        "def number(value):\n",
        "    if (value > 10):\n",
        "        return True\n",
        "    else:\n",
        "        return False"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "K3Z832Cdkqdf"
      },
      "outputs": [],
      "source": [
        "filterlessRDD = subRDD.filter(number)\n",
        "print (filteredRDD)\n",
        "print (filterlessRDD.collect())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "VBFP_YFxkqdg"
      },
      "source": [
        "<a id = \"lambda\"></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Pr9B-llbkqdh"
      },
      "source": [
        "### ** 4. Lambda Functions **"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "a0Ur3ZRqkqdh"
      },
      "source": [
        "#### ** (4a) Using Python `lambda()` functions **\n",
        "\n",
        "Python supports the use of small one-line anonymous functions that are not bound to a name at runtime. Borrowed from LISP, these `lambda` functions can be used wherever function objects are required. They are syntactically restricted to a single expression. Remember that `lambda` functions are a matter of style and using them is never required - semantically, they are just syntactic sugar for a normal function definition. You can always define a separate normal function instead, but using a `lambda()` function is an equivalent and more compact form of coding. Ideally you should consider using `lambda` functions where you want to encapsulate non-reusable code without littering your code with one-line functions.\n",
        "\n",
        "Here, instead of defining a separate function for the `filter()` transformation, we will use an inline `lambda()` function."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CeiRjlGlkqdh"
      },
      "outputs": [],
      "source": [
        "lambdaRDD = subRDD.filter(lambda x: x < 10)\n",
        "lambdaRDD.collect()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "xnuid8rYkqdj"
      },
      "outputs": [],
      "source": [
        "# Let's collect the even values less than 10\n",
        "evenRDD = lambdaRDD.filter(lambda x: x % 2 == 0)\n",
        "evenRDD.collect()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZQ9jeKIVkqdk"
      },
      "source": [
        "<a id = \"actions\"></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WewGXghJkqdk"
      },
      "source": [
        "### ** 5. Additional RDD actions **"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "z_hTRUOukqdk"
      },
      "source": [
        "#### ** (5a) Other common actions **\n",
        "\n",
        "Let's investigate the additional actions: [first()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.first.html), [take()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.take.html), [top()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.top.html), [takeOrdered()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.takeOrdered.html ), and [reduce()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.reduce.html)\n",
        "\n",
        "One useful thing to do when we have a new dataset is to look at the first few entries to obtain a rough idea of what information is available.  In Spark, we can do that using the `first()`, `take()`, `top()`, and `takeOrdered()` actions. Note that for the `first()` and `take()` actions, the elements that are returned depend on how the RDD is *partitioned*.\n",
        "\n",
        "Instead of using the `collect()` action, we can use the `take(n)` action to return the first n elements of the RDD. The `first()` action returns the first element of an RDD, and is equivalent to `take(1)`.\n",
        "\n",
        "The `takeOrdered()` action returns the first n elements of the RDD, using either their natural order or a custom comparator. The key advantage of using `takeOrdered()` instead of `first()` or `take()` is that `takeOrdered()` returns a deterministic result, while the other two actions may return differing results, depending on the number of partions or execution environment. `takeOrdered()` returns the list sorted in *ascending order*.  The `top()` action is similar to `takeOrdered()` except that it returns the list in *descending order.*\n",
        "\n",
        "The `reduce()` action reduces the elements of a RDD to a single value by applying a function that takes two parameters and returns a single value.  The function should be commutative and associative, as `reduce()` is applied at the partition level and then again to aggregate results from partitions.  If these rules don't hold, the results from `reduce()` will be inconsistent.  Reducing locally at partitions makes `reduce()` very efficient."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "c3vIB4Qhkqdl"
      },
      "outputs": [],
      "source": [
        "# Let's get the first element\n",
        "print (filteredRDD.first())\n",
        "# The first 4\n",
        "print (filteredRDD.take(4))\n",
        "# Note that it is ok to take more elements than the RDD has\n",
        "print (filteredRDD.take(12))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "8OFsY7jSkqdn"
      },
      "outputs": [],
      "source": [
        "# Retrieve the three smallest elements\n",
        "print (filteredRDD.takeOrdered(3))\n",
        "# Retrieve the five largest elements\n",
        "print (filteredRDD.top(5))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ZnYVEqDlkqdr"
      },
      "outputs": [],
      "source": [
        "# Pass a lambda function to takeOrdered to reverse the order\n",
        "filteredRDD.takeOrdered(4, lambda s: -s)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Co8B-qUOkqdv"
      },
      "outputs": [],
      "source": [
        "# Obtain Python's add function\n",
        "from operator import add\n",
        "# Efficiently sum the RDD using reduce\n",
        "print (filteredRDD.reduce(add))\n",
        "# Sum using reduce with a lambda function\n",
        "print (filteredRDD.reduce(lambda a, b: a + b))\n",
        "# Note that subtraction is not both associative and commutative\n",
        "print (filteredRDD.reduce(lambda a, b: a - b))\n",
        "print (filteredRDD.repartition(4).reduce(lambda a, b: a - b))\n",
        "# While addition is\n",
        "print (filteredRDD.repartition(4).reduce(lambda a, b: a + b))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_-Gw1dP5kqdy"
      },
      "source": [
        "#### ** (5b) Advanced actions **\n",
        "\n",
        "Here are two additional actions that are useful for retrieving information from an RDD: [takeSample()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.takeSample.html) and [countByValue()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.countByValue.html)\n",
        "\n",
        "The `takeSample()` action returns an array with a random sample of elements from the dataset.  It takes in a `withReplacement` argument, which specifies whether it is okay to randomly pick the same item multiple times from the parent RDD (so when `withReplacement=True`, you can get the same item back multiple times). It also takes an optional `seed` parameter that allows you to specify a seed value for the random number generator, so that reproducible results can be obtained.\n",
        "\n",
        "The `countByValue()` action returns the count of each unique value in the RDD as a dictionary that maps values to counts."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "e9vDx0BTkqdz"
      },
      "outputs": [],
      "source": [
        "# takeSample reusing elements\n",
        "print (filteredRDD.takeSample(withReplacement=True, num=6))\n",
        "# takeSample without reuse\n",
        "print (filteredRDD.takeSample(withReplacement=False, num=6))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "HEPSBDChkqd0"
      },
      "outputs": [],
      "source": [
        "# Set seed for predictability\n",
        "print (filteredRDD.takeSample(withReplacement=False, num=6, seed=500))\n",
        "# Try reruning this cell and the cell above -- the results from this cell will remain constant\n",
        "# Use ctrl-enter to run without moving to the next cell"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CRxNPyDjkqd2"
      },
      "outputs": [],
      "source": [
        "# Create new base RDD to show countByValue\n",
        "repetitiveRDD = sc.parallelize([1, 2, 3, 1, 2, 3, 1, 2, 1, 2, 3, 3, 3, 4, 5, 4, 6])\n",
        "print (repetitiveRDD.countByValue())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "QNAPDD6Hkqd4"
      },
      "source": [
        "<a id = \"transformations\"></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iO0KG15Ukqd5"
      },
      "source": [
        "### ** 6. Additional RDD transformations **"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "nclbVYTWkqd5"
      },
      "source": [
        "#### ** (6a) `flatMap` **\n",
        "\n",
        "When performing a `map()` transformation using a function, sometimes the function will return more (or less) than one element. We would like the newly created RDD to consist of the elements outputted by the function. Simply applying a `map()` transformation would yield a new RDD made up of iterators.  Each iterator could have zero or more elements.  Instead, we often want an RDD consisting of the values contained in those iterators.  The solution is to use a [flatMap()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.flatMap.html) transformation, `flatMap()` is similar to `map()`, except that with `flatMap()` each input item can be mapped to zero or more output elements.\n",
        "\n",
        "To demonstrate `flatMap()`, we will first emit a word along with its plural, and then a range that grows in length with each subsequent operation."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "53nFbxVHkqd6"
      },
      "outputs": [],
      "source": [
        "# Let's create a new base RDD to work from\n",
        "wordsList = ['cat', 'elephant', 'rat', 'rat', 'cat']\n",
        "wordsRDD = sc.parallelize(wordsList, 4)\n",
        "\n",
        "# Use map\n",
        "singularAndPluralWordsRDDMap = wordsRDD.map(lambda x: (x, x + 's'))\n",
        "# Use flatMap\n",
        "singularAndPluralWordsRDD = wordsRDD.flatMap(lambda x: (x, x + 's'))\n",
        "\n",
        "# View the results\n",
        "print (singularAndPluralWordsRDDMap.collect())\n",
        "print (singularAndPluralWordsRDD.collect())\n",
        "# View the number of elements in the RDD\n",
        "print (singularAndPluralWordsRDDMap.count())\n",
        "print (singularAndPluralWordsRDD.count())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "0sDiCNMskqd8"
      },
      "outputs": [],
      "source": [
        "simpleRDD = sc.parallelize([2, 3, 4])\n",
        "print (simpleRDD.map(lambda x: range(1, x)).collect())\n",
        "print (simpleRDD.flatMap(lambda x: range(1, x)).collect())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_nrc2TQRkqd9"
      },
      "source": [
        "#### ** (6b) `groupByKey` and `reduceByKey`  **\n",
        "\n",
        "Let's investigate the additional transformations: [groupByKey()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.groupByKey.html) and [reduceByKey()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.reduceByKey.html).\n",
        "\n",
        "Both of these transformations operate on pair RDDs.  A pair RDD is an RDD where each element is a pair tuple (key, value).  For example, `sc.parallelize([('a', 1), ('a', 2), ('b', 1)])` would create a pair RDD where the keys are 'a', 'a', 'b' and the values are 1, 2, 1.\n",
        "\n",
        "The `reduceByKey()` transformation gathers together pairs that have the same key and applies a function to two associated values at a time. `reduceByKey()` operates by applying the function first within each partition on a per-key basis and then across the partitions.\n",
        "\n",
        "While both the `groupByKey()` and `reduceByKey()` transformations can often be used to solve the same problem and will produce the same answer, the `reduceByKey()` transformation works much better for large distributed datasets. This is because Spark knows it can combine output with a common key on each partition *before* shuffling (redistributing) the data across nodes.  Only use `groupByKey()` if the operation would not benefit from reducing the data before the shuffle occurs.\n",
        "\n",
        "Look at the diagram below to understand how `reduceByKey` works.  Notice how pairs on the same machine with the same key are combined (by using the lamdba function passed into reduceByKey) before the data is shuffled. Then the lamdba function is called again to reduce all the values from each partition to produce one final result.\n",
        "![reduceByKey() figure](http://spark-mooc.github.io/web-assets/images/reduce_by.png)\n",
        "\n",
        "On the other hand, when using the `groupByKey()` transformation - all the key-value pairs are shuffled around, causing a lot of unnecessary data to being transferred over the network.\n",
        "\n",
        "To determine which machine to shuffle a pair to, Spark calls a partitioning function on the key of the pair. Spark spills data to disk when there is more data shuffled onto a single executor machine than can fit in memory. However, it flushes out the data to disk one key at a time, so if a single key has more key-value pairs than can fit in memory an out of memory exception occurs. This will be more gracefully handled in a later release of Spark so that the job can still proceed, but should still be avoided.  When Spark needs to spill to disk, performance is severely impacted.\n",
        "![groupByKey() figure](http://spark-mooc.github.io/web-assets/images/group_by.png)\n",
        "\n",
        "As your dataset grows, the difference in the amount of data that needs to be shuffled, between the `reduceByKey()` and `groupByKey()` transformations, becomes increasingly exaggerated.\n",
        "\n",
        "Here are more transformations to prefer over `groupByKey()`:\n",
        "  +   [combineByKey()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.combineByKey.html) can be used when you are combining elements but your return type differs from your input value type.\n",
        "  +   [foldByKey()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.foldByKey.html ) merges the values for each key using an associative function and a neutral \"zero value\".\n",
        "\n",
        "Now let's go through a simple `groupByKey()` and `reduceByKey()` example."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "wYdM5RNokqd9"
      },
      "outputs": [],
      "source": [
        "pairRDD = sc.parallelize([('a', 1), ('a', 2), ('b', 1)])\n",
        "# mapValues only used to improve format for printing\n",
        "print(pairRDD.groupByKey().mapValues(lambda x: list(x)).collect())\n",
        "\n",
        "# Using mapValues, which is recommended when they key doesn't change\n",
        "print(pairRDD.groupByKey().mapValues(lambda x: sum(x)).collect())\n",
        "\n",
        "# reduceByKey is more efficient / scalable\n",
        "print(pairRDD.reduceByKey(add).collect())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Uk5_BAcXkqd-"
      },
      "source": [
        "#### ** (6c)  `mapPartitions()` and  `mapPartitionsWithIndex()` ** [Optional]\n",
        "\n",
        "Let's investigate the advanced transformations: [mapPartitions()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.mapPartitions.html) and [mapPartitionsWithIndex()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.mapPartitionsWithIndex.html)\n",
        "\n",
        "The `mapPartitions()` transformation uses a function that takes in an iterator (to the items in that specific partition) and returns an iterator.  The function is applied on a partition by partition basis.\n",
        "\n",
        "The `mapPartitionsWithIndex()` transformation uses a function that takes in a partition index (think of this like the partition number) and an iterator (to the items in that specific partition). For every partition (index, iterator) pair, the function returns a tuple of the same partition index number and an iterator of the transformed items in that partition."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "RbEDFH3Skqd_"
      },
      "outputs": [],
      "source": [
        "# mapPartitions takes a function that takes an iterator and returns an iterator\n",
        "print (wordsRDD.collect())\n",
        "itemsRDD = wordsRDD.mapPartitions(lambda iterator: [','.join(iterator)])\n",
        "print (itemsRDD.collect())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "o7ZYnjtmkqeA"
      },
      "outputs": [],
      "source": [
        "itemsByPartRDD = wordsRDD.mapPartitionsWithIndex(lambda index, iterator: [(index, list(iterator))])\n",
        "# We can see that three of the (partitions) workers have one element and the fourth worker has two\n",
        "# elements, although things may not bode well for the rat...\n",
        "print (itemsByPartRDD.collect())\n",
        "# Rerun without returning a list (acts more like flatMap)\n",
        "itemsByPartRDD = wordsRDD.mapPartitionsWithIndex(lambda index, iterator: (index, list(iterator)))\n",
        "print (itemsByPartRDD.collect())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xLis7tjc68iM"
      },
      "source": [
        "#### ** (6d) Advanced transformations ** [Optional]\n",
        "\n",
        "Let's investigate the additional transformations: [sortByKey()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.sortByKey.html), [join()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.join.html), [leftOuterJoin()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.leftOuterJoin.html), [rightOuterJoin()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.rightOuterJoin.html) and [fullOuterJoin()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.fullOuterJoin.html),\n",
        "\n",
        "The `sortByKey()` returns a new dataset `(k,v)` pairs sorted by keys in ascending order.\n",
        "\n",
        "\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "l3qRiJ0YAwsz"
      },
      "outputs": [],
      "source": [
        "rdd = sc.parallrdd2 = sc.parallelize([(1,'a'), (2,'c'), (1,'b')])\n",
        "rdd.sortByKey().collect()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "OLTW3HyJAvnS"
      },
      "source": [
        "The `join()` is an inner join that returns RDD of all pairs of elements with matching keys in X and Y. Each pair is  `(k, (v1,v2)) ` tuple, where  `(k,v1) ` is in X, and  `(k,v2) ` is in Y.\n",
        "\n",
        "Meanwhile, `leftOuterJoin()`, `rightOuterJoin()` and `fullOuterJoin()` perform outer join.    \n",
        "\n",
        "\n",
        "*   `leftOuterJoin()`: For each element (k, v) in X, resulting RDD will either contain all pairs (k,(v, w)) for w in Y, Or pair (k,(v, None) if no elements in Y have key k.\n",
        "*   `rightOuterJoin()`: For each element (k, w) in Y, resulting RDD will either contain all pairs (k,(v, w)) for v in X, Or pair (k,(None, w) if no elements in X have key k.\n",
        "*   `fullOuterJoin()`: For each element (k,v) in X, resulting RDD will either contain all pairs (k,(v,w)) for w in Y, or (k,(v,None) if no match in Y have k; and For each element (k,w) in Y, resulting RDD with either contain all pairs (k,(v,w)) for v in X, or (k,(None,w) if no match in X have k.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "2OuklE5GAyx5"
      },
      "outputs": [],
      "source": [
        "# join()\n",
        "x = sc.parallelize([(\"a\", 1), (\"b\", 4)])\n",
        "y = sc.parallelize([(\"a\", 2), (\"a\", 3)])\n",
        "sorted(x.join(y).collect())\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "VqDEVtb0A4oN"
      },
      "outputs": [],
      "source": [
        "# leftOuterJoin()\n",
        "x = sc.parallelize([(\"a\", 1), (\"b\", 4)])\n",
        "y = sc.parallelize([(\"a\", 2)])\n",
        "sorted(x.leftOuterJoin(y).collect())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cztVQtiBBP14"
      },
      "outputs": [],
      "source": [
        "# rightOuterJoin()\n",
        "x = sc.parallelize([(\"a\", 1), (\"b\", 4)])\n",
        "y = sc.parallelize([(\"a\", 2), (\"c\", 5)])\n",
        "sorted(x.rightOuterJoin(y).collect())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Wcydy8suBdID"
      },
      "outputs": [],
      "source": [
        "# fullOuterJoin()\n",
        "x = sc.parallelize([(\"a\", 1), (\"b\", 4)])\n",
        "y = sc.parallelize([(\"a\", 2), (\"c\", 5)])\n",
        "sorted(x.fullOuterJoin(y).collect())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "bc3TnZWXkqeC"
      },
      "source": [
        "### ** 7. Caching RDDs and storage options **"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1_GChTlmkqeD"
      },
      "source": [
        "#### ** (7a) Caching RDDs **\n",
        "\n",
        "For efficiency Spark keeps your RDDs in memory. By keeping the contents in memory, Spark can quickly access the data. However, memory is limited, so if you try to keep too many RDDs in memory, Spark will automatically delete RDDs from memory to make space for new RDDs. If you later refer to one of the RDDs, Spark will automatically recreate the RDD for you, but that takes time.\n",
        "\n",
        "So, if you plan to use an RDD more than once, then you should tell Spark to cache that RDD. You can use the `cache()` operation to keep the RDD in memory. However, if you cache too many RDDs and Spark runs out of memory, it will delete the least recently used (LRU) RDD first. Again, the RDD will be automatically recreated when accessed.\n",
        "\n",
        "You can check if an RDD is cached by using the `is_cached` attribute, and you can see your cached RDD in the \"Storage\" section of the Spark web UI. If you click on the RDD's name, you can see more information about where the RDD is stored."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "eZkQOi70kqeD"
      },
      "outputs": [],
      "source": [
        "# Name the RDD\n",
        "filteredRDD.setName('My Filtered RDD')\n",
        "# Cache the RDD\n",
        "filteredRDD.cache()\n",
        "# Is it cached\n",
        "print (filteredRDD.is_cached)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-_ntHxR9kqeE"
      },
      "source": [
        "#### ** (7b) Unpersist and storage options **\n",
        "\n",
        "Spark automatically manages the RDDs cached in memory and will save them to disk if it runs out of memory. For efficiency, once you are finished using an RDD, you can optionally tell Spark to stop caching it in memory by using the RDD's `unpersist()` method to inform Spark that you no longer need the RDD in memory.\n",
        "\n",
        "You can see the set of transformations that were applied to create an RDD by using the `toDebugString()` method, which will provide storage information, and you can directly query the current storage information for an RDD using the `getStorageLevel()` operation.\n",
        "\n",
        "** Advanced: ** Spark provides many more options for managing how RDDs are stored in memory or even saved to disk. You can explore the API for RDD's [persist()](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.persist.html) operation using Python's [help()](https://docs.python.org/2/library/functions.html?highlight=help#help) command.  The `persist()` operation, optionally, takes a pySpark [StorageLevel](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.StorageLevel.html) object."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ZUZri_1tkqeE"
      },
      "outputs": [],
      "source": [
        "# Note that toDebugString also provides storage information\n",
        "print (filteredRDD.toDebugString())"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "niXDAy08kqeF"
      },
      "outputs": [],
      "source": [
        "# If we are done with the RDD we can unpersist it so that its memory can be reclaimed\n",
        "filteredRDD.unpersist()\n",
        "# Storage level for a non cached RDD\n",
        "print (filteredRDD.getStorageLevel())\n",
        "filteredRDD.cache()\n",
        "# Storage level for a cached RDD\n",
        "print (filteredRDD.getStorageLevel())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wraAtMoUkqeH"
      },
      "source": [
        "<a id = \"debug\"></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hV-8L7m8kqeH"
      },
      "source": [
        "### ** 8. Debugging Spark applications and lazy evaluation **"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "nY-vgofskqeH"
      },
      "source": [
        "#### ** How Python is Executed in Spark **\n",
        "\n",
        "Internally, Spark executes using a Java Virtual Machine (JVM). pySpark runs Python code in a JVM using [Py4J](http://py4j.sourceforge.net). Py4J enables Python programs running in a Python interpreter to dynamically access Java objects in a Java Virtual Machine. Methods are called as if the Java objects resided in the Python interpreter and Java collections can be accessed through standard Python collection methods. Py4J also enables Java programs to call back Python objects.\n",
        "\n",
        "Because pySpark uses Py4J, coding errors often result in a complicated, confusing stack trace that can be difficult to understand. In the following section, we'll explore how to understand stack traces."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "KruG0wMlkqeH"
      },
      "source": [
        "#### ** (8a) Challenges with lazy evaluation using transformations and actions **\n",
        "\n",
        "Spark's use of lazy evaluation can make debugging more difficult because code is not always executed immediately. To see an example of how this can happen, let's first define a broken filter function.\n",
        "\n",
        "Next we perform a `filter()` operation using the broken filtering function.  No error will occur at this point due to Spark's use of lazy evaluation.\n",
        "\n",
        "The `filter()` method will not be executed *until* an action operation is invoked on the RDD.  We will perform an action by using the `collect()` method to return a list that contains all of the elements in this RDD."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "V_rhu1OykqeI"
      },
      "outputs": [],
      "source": [
        "def brokenTen(value):\n",
        "    \"\"\"Incorrect implementation of the ten function.\n",
        "\n",
        "    Note:\n",
        "        The `if` statement checks an undefined variable `val` instead of `value`.\n",
        "\n",
        "    Args:\n",
        "        value (int): A number.\n",
        "\n",
        "    Returns:\n",
        "        bool: Whether `value` is less than ten.\n",
        "\n",
        "    Raises:\n",
        "        NameError: The function references `val`, which is not available in the local or global\n",
        "            namespace, so a `NameError` is raised.\n",
        "    \"\"\"\n",
        "    if (val < 10):\n",
        "        return True\n",
        "    else:\n",
        "        return False\n",
        "\n",
        "brokenRDD = subRDD.filter(brokenTen)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "0lRT_RsskqeI"
      },
      "outputs": [],
      "source": [
        "# Now we'll see the error\n",
        "brokenRDD.collect()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "roJWmD2akqeL"
      },
      "source": [
        "#### ** (8b) Finding the bug **\n",
        "\n",
        "When the `filter()` method is executed, Spark evaluates the RDD by executing the `parallelize()` and `filter()` methods. Since our `filter()` method has an error in the filtering function `brokenTen()`, an error occurs.\n",
        "\n",
        "Scroll through the output \"Py4JJavaError     Traceback (most recent call last)\" part of the cell and first you will see that the line that generated the error is the `collect()` method line. There is *nothing wrong with this line*. However, it is an action and that caused other methods to be executed. Continue scrolling through the Traceback and you will see the following error line:\n",
        "    NameError: global name 'val' is not defined\n",
        "\n",
        "Looking at this error line, we can see that we used the wrong variable name in our filtering function `brokenTen()`."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "LEXQZDlvkqeL"
      },
      "source": [
        "#### ** (8c) Moving toward expert style **\n",
        "\n",
        "As you are learning Spark, I recommend that you write your code in the form:\n",
        "    RDD.transformation1()\n",
        "    RDD.action1()\n",
        "    RDD.transformation2()\n",
        "    RDD.action2()\n",
        "\n",
        "Using this style will make debugging your code much easier as it makes errors easier to localize - errors in your transformations will occur when the next action is executed.\n",
        "\n",
        "Once you become more experienced with Spark, you can write your code with the form:\n",
        "    RDD.transformation1().transformation2().action()\n",
        "\n",
        "We can also use `lambda()` functions instead of separately defined functions when their use improves readability and conciseness."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "INbHH35zkqeL"
      },
      "outputs": [],
      "source": [
        "# Cleaner code through lambda use\n",
        "subRDD.filter(lambda x: x < 10).collect()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "KxWN9Im9kqeM"
      },
      "outputs": [],
      "source": [
        "# Even better by moving our chain of operators into a single line.\n",
        "sc.parallelize(data).map(lambda y: y - 1).filter(lambda x: x < 10).collect()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "E7MmxA_vkqeN"
      },
      "source": [
        "#### ** (8d) Readability and code style **\n",
        "\n",
        "To make the expert coding style more readable, enclose the statement in parentheses and put each method, transformation, or action on a separate line."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CxZvr8qZkqeN"
      },
      "outputs": [],
      "source": [
        "# Final version\n",
        "(sc\n",
        " .parallelize(data)\n",
        " .map(lambda y: y - 1)\n",
        " .filter(lambda x: x < 10)\n",
        " .collect())"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "W3_UZZLOkqeO"
      },
      "source": [
        "## **Part B: Test Spark functionality**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "AedI98PLkqeO"
      },
      "source": [
        "<a id = \"spark\"></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "W_7-1Pv0kqeP"
      },
      "source": [
        "### 9. Spark Functionality\n",
        "\n",
        "#### **(9a) Parallelize, filter, and reduce**"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6c5q1_JhkqeQ"
      },
      "outputs": [],
      "source": [
        "# Check that Spark is working\n",
        "largeRange = sc.parallelize(range(100000))\n",
        "reduceTest = largeRange.reduce(lambda a, b: a + b)\n",
        "filterReduceTest = largeRange.filter(lambda x: x % 7 == 0).sum()\n",
        "\n",
        "print (reduceTest)\n",
        "print (filterReduceTest)\n",
        "\n",
        "# If the Spark jobs don't work properly these will raise an AssertionError\n",
        "assert reduceTest == 4999950000\n",
        "assert filterReduceTest == 714264285"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "RPdpMANQkqeR"
      },
      "source": [
        "#### **(9b) Loading a text file**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xPg5Z5fGkqeR"
      },
      "source": [
        "You can upload your file to Amazon S3 or other online storage so that the Spark cluster can access it.\n",
        "\n",
        "In DataBricks, you can check this URL for information:\n",
        "\n",
        "- https://docs.databricks.com/user-guide/advanced/filestore.html\n",
        "\n",
        "You can import the data through Tables, Create Table, Data Import functions. Please remember to note down the prompt address, such as:\n",
        "\n",
        "- /FileStore/tables/dnsmv70v1491290162600/shakespeare.txt\n",
        "\n",
        "If the file uploading is properly done, you can access the file via Web browser using this URL:\n",
        "\n",
        "- https://community.cloud.databricks.com/files/my-stuff/my-file.txt?o=######\n",
        "\n",
        "Replace ###### with your community edition ID. Such as\n",
        "\n",
        "- https://community.cloud.databricks.com/files/tables/dnsmv70v1491290162600/shakespeare.txt?o=48280648682457\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "wgyshgbFkqeR"
      },
      "outputs": [],
      "source": [
        "# You can check the content of a specified folder on DataBricks' DBFS.\n",
        "# display(dbutils.fs.ls(\"/FileStore/tables\"))"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "rAiJa5y2kqeU"
      },
      "outputs": [],
      "source": [
        "!pip install wget\n",
        "import wget\n",
        "\n",
        "link_to_data = 'https://github.com/tuliplab/mds/raw/master/Jupyter/data/shakespeare.txt'\n",
        "DataSet = wget.download(link_to_data)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "x2tHCQWbkqeV"
      },
      "outputs": [],
      "source": [
        "# Check loading data with sc.textFile\n",
        "#import os.path\n",
        "#baseDir = os.path.join('data')\n",
        "#inputPath = os.path.join('cs100', 'lab1', 'shakespeare.txt')\n",
        "#fileName = os.path.join(baseDir, inputPath)\n",
        "\n",
        "# Replace this address by your own filestore address\n",
        "rawData = sc.textFile('shakespeare.txt')\n",
        "shakespeareCount = rawData.count()\n",
        "\n",
        "print(shakespeareCount)\n",
        "\n",
        "# If the text file didn't load properly an AssertionError will be raised\n",
        "assert shakespeareCount == 122395"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_yIw1gQ4kqeV"
      },
      "source": [
        "<a id = \"test\"></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "j-uKRAVykqeZ"
      },
      "source": [
        "### **10. Check plotting**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yg8C3nEekqeZ"
      },
      "source": [
        "#### **(10a) Our first plot**\n",
        "\n",
        "After executing the code cell below, you should see a plot with 50 blue circles.  The circles should start at the bottom left and end at the top right."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "yxHbMFX0kqeZ",
        "collapsed": true
      },
      "outputs": [],
      "source": [
        "# Check matplotlib plotting\n",
        "import matplotlib.pyplot as plt\n",
        "import matplotlib.cm as cm\n",
        "from math import log\n",
        "\n",
        "# %matplotlib inline\n",
        "\n",
        "# function for generating plot layout\n",
        "def preparePlot(xticks, yticks, figsize=(10.5, 6), hideLabels=False, gridColor='#999999', gridWidth=1.0):\n",
        "    plt.close()\n",
        "    fig, ax = plt.subplots(figsize=figsize, facecolor='white', edgecolor='white')\n",
        "    ax.axes.tick_params(labelcolor='#999999', labelsize='10')\n",
        "    for axis, ticks in [(ax.get_xaxis(), xticks), (ax.get_yaxis(), yticks)]:\n",
        "        axis.set_ticks_position('none')\n",
        "        axis.set_ticks(ticks)\n",
        "        axis.label.set_color('#999999')\n",
        "        if hideLabels: axis.set_ticklabels([])\n",
        "    plt.grid(color=gridColor, linewidth=gridWidth, linestyle='-')\n",
        "    map(lambda position: ax.spines[position].set_visible(False), ['bottom', 'top', 'left', 'right'])\n",
        "    return fig, ax\n",
        "\n",
        "# generate layout and plot data\n",
        "x = range(1, 50)\n",
        "y = [log(x1 ** 2) for x1 in x]\n",
        "fig, ax = preparePlot(range(5, 60, 10), range(0, 12, 1))\n",
        "plt.scatter(x, y, s=14**2, c='#d6ebf2', edgecolors='#8cbfd0', alpha=0.75)\n",
        "ax.set_xlabel(r'$range(1, 50)$'), ax.set_ylabel(r'$\\log_e(x^2)$')\n",
        "\n",
        "display(fig)\n",
        "pass"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [
        "aFXb1VlkkqdA",
        "nY-vgofskqeH",
        "roJWmD2akqeL"
      ],
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "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.7.3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}