lab1_word_count_text.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "EWrJSD9o8h5C"
   },
   "source": [
    "![Spark Logo](http://spark-mooc.github.io/web-assets/images/ta_Spark-logo-small.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "KW3rh0nE8h5D"
   },
   "source": [
    "<p style=\"font-size:20px;font-weight: bold;color:blue\">\"ESIEE Paris\" </p>\n",
    "<p style=\"font-size:15px;font-weight: bold;color:blue\">BigData analytics with Spark - DSIA_5202A - 2022-2023 -- Instructor: B. Men</p>\n",
    "\n",
    "**Sources**: These labs synthetize and *builds on* labs from several origins: \n",
    "- The series of moocs from Berkeley and Databricks,(Creative Commons licences), namely\n",
    "   - [Introduction to Apache Spark](https://courses.edx.org/courses/course-v1:BerkeleyX+CS105x+1T2016/info)\n",
    "   - [Big data Analysis with Apache Spark](https://courses.edx.org/courses/course-v1:BerkeleyX+CS110x+2T2016/info)\n",
    "   - [Distributed Machine Learning with Apache Spark](https://courses.edx.org/courses/course-v1:BerkeleyX+CS120x+2T2016/info)\n",
    "   - [Introduction to Big Data with Apache Spark](https://courses.edx.org/courses/BerkeleyX/CS100.1x/1T2015/info)\n",
    "   - [Scalable Machine Learning](https://courses.edx.org/courses/BerkeleyX/CS190.1x/1T2015/info)\n",
    "- [Apache Spark & Python (pySpark) tutorials for Big Data Analysis and Machine Learning](https://github.com/jadianes/spark-py-notebooks) (Apache License, Version 2.0)\n",
    "\n",
    "We have kept the labs text in english. This will enable us to reuse them in international sections. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Yjf33yzi8h5G"
   },
   "source": [
    "<p style=\"font-size:35px;font-weight: bold;\"> Lab 1 :  Word Count with Spark</p>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "yxz7pLsA8h5H"
   },
   "source": [
    "This lab will enable us to develop a simple word count application.  The volume of unstructured text in existence is growing dramatically, and Spark is an excellent tool for analyzing this type of data.  In this lab, we will write code that calculates the most common words in the [Complete Works of William Shakespeare](http://www.gutenberg.org/ebooks/100) retrieved from [Project Gutenberg](https://www.gutenberg.org/).  This could also be scaled to find the most common words on the Internet.\n",
    "\n",
    "** During this lab we will cover: **\n",
    "\n",
    "- **Part 1:** Word Count with RDDs  -- You will play with basic RDDs, then build a pair RDD, use it for counting words. Finally you will learn how to clean and prepare the RDD and build the application to count words in a file. \n",
    "\n",
    "- **Part 2:** is devoted to a small tutorial on Spark Dataframes (introduced in Spark 1.3)\n",
    "\n",
    "- **Part 3:** Word Count with dataframes -- You will follow the same steps as in Part 1 to develop a Word Counting application and apply it to a file. \n",
    "\n",
    "\n",
    "**Exercises** will include an explanation of what is expected, followed by code cells where one cell will have one or more `<FILL IN>` sections.  The cell that needs to be modified will have `// TODO: Replace <FILL IN> with appropriate code` on its first line.  Once the `<FILL IN>` sections are updated and the code is run, the test cell can then be run to verify the correctness of your solution.  The last code cell before the next markdown section will contain the tests.\n",
    "\n",
    "Note that, for reference, you can look up the details of the relevant methods in [Spark's Scala API](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/index.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "MpCNZmnK8h5I"
   },
   "source": [
    "##  Prerequisites : Spark Context configuration \n",
    "\n",
    "</p>Execute this cell, this will import spark in the notebook.</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import $ivy.`org.apache.spark::spark-sql:3.3.1`\n",
    "// import $ivy.`sh.almond::almond-spark:_` // Added automatically on importing Spark with Binder\n",
    "\n",
    "import org.apache.spark.sql._"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "val spark = SparkSession.builder\n",
    ".appName(\"lab1_word_count_text\")\n",
    ".master(\"local[*]\")\n",
    ".getOrCreate()\n",
    "val sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This will limit the logs outputed by spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import org.apache.log4j.{Level, Logger}\n",
    "Logger.getLogger(\"org\").setLevel(Level.OFF)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "RhHwvWV68h5Q"
   },
   "source": [
    "# Part 1 : Word Count with RDDs  \n",
    "\n",
    "##  1. Creating a base RDD and pair RDDs \n",
    "\n",
    "We will first explore creating a base RDD with `parallelize` and using pair RDDs to count words.\n",
    "\n",
    "\n",
    "**_(1a) Create a base RDD_**\n",
    "\n",
    "We'll start by generating a base RDD by using a List and the `sc.parallelize` method.  Then we'll print out the type of the base RDD.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "JWcCjDQM8h5S",
    "outputId": "deb1f61d-7cc0-4b1d-e4ca-f2faf71052c1"
   },
   "outputs": [],
   "source": [
    "val wordsList = List(\"cat\", \"elephant\", \"rat\", \"rat\", \"cat\")\n",
    "val wordsRDD = sc.parallelize(wordsList, 4)\n",
    "// Print out the type of wordsRDD\n",
    "println(wordsRDD.getClass)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "N9CacR-68h5X"
   },
   "source": [
    "**_(1b) Pluralize and test_**\n",
    "\n",
    "Let's use a `map()` transformation to add the letter 's' to each string in the base RDD we just created. We'll define a function that returns the word with an 's' at the end of the word.  Please replace `???` with your solution.  If you have trouble, the next cell has the solution.  After you have defined `makePlural` you can run the third cell which contains a test.  If you implementation is correct it will print `1 test passed`.\n",
    "\n",
    "This is the general form that exercises will take, except that no example solution will be provided. \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def assertResult[A](expected : A, error : String)(answer : A) = {\n",
    "    if (expected equals answer) \"1 test passed\"\n",
    "    else error\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "5bxyS-xs8h5X",
    "outputId": "cce47d3c-2ae8-482c-a0db-b5cc6944fe94"
   },
   "outputs": [],
   "source": [
    "//TODO: Replace ??? with appropriate code\n",
    "def makePlural(word : String): String = {\n",
    "    /*Adds an 's' to `word`.\n",
    "\n",
    "    Note:\n",
    "        This is a simple function that only adds an 's'.  No attempt is made to follow proper\n",
    "        pluralization rules.\n",
    "\n",
    "    Args:\n",
    "        word (str): A string.\n",
    "\n",
    "    Returns:\n",
    "        str: A string with 's' added to it.\n",
    "    */\n",
    "    return ???\n",
    "}\n",
    "    \n",
    "\n",
    "print(makePlural(\"cat\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "RC9wobX08h5c",
    "outputId": "9284858f-1997-44f3-fc78-fa33f6bf725b"
   },
   "outputs": [],
   "source": [
    "//One way of completing the function\n",
    "val makePlural = (word : String) => \n",
    "    word + \"s\"\n",
    "\n",
    "print(makePlural(\"cat\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "eesMtpv18h5h",
    "outputId": "cc0580d5-b95c-4423-f7f2-e16ef8325bd8"
   },
   "outputs": [],
   "source": [
    "/*\n",
    "# Load in the testing code and check to see if your answer is correct\n",
    "# Make sure to rerun any cell you change before trying the test again\n",
    "*/\n",
    "\n",
    "//import org.scalatest.Assertions._\n",
    "//import scala.collection.mutable.Stack\n",
    "//TEST Pluralize and test (1b)\n",
    "assertResult(\"rats\", \"incorrect result: makePlural does not add an s\") {\n",
    "      makePlural(\"rat\")\n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "RZvSlKIc8h5l"
   },
   "source": [
    "**_(1c) Apply `makePlural` to the base RDD_**\n",
    "\n",
    "Now pass each item in the base RDD into a [map()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/rdd/RDD.html#map[U](f:T=%3EU)(implicitevidence$3:scala.reflect.ClassTag[U]):org.apache.spark.rdd.RDD[U]) transformation that applies the `makePlural()` function to each element. And then call the [collect()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/rdd/RDD.html#collect():Array[T]) action to see the transformed RDD.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "8e2bkVbV8h5m",
    "outputId": "b74aece7-fb9a-4fc3-f9e8-3edc194d415c"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val pluralRDD = wordsRDD.map(???)\n",
    "pluralRDD.collect().map(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "W7SKJtbb8h5p",
    "outputId": "d5985991-f2aa-40dc-e185-528d20cbb515"
   },
   "outputs": [],
   "source": [
    "//TEST Apply makePlural to the base RDD(1c)\n",
    "assertResult(List(\"cats\", \"elephants\", \"rats\", \"rats\", \"cats\"), \"incorrect values for pluralRDD\") {\n",
    "      pluralRDD.collect.toList\n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "GfX_IE1o8h5u"
   },
   "source": [
    "**_(1d) Pass an `anonymous` function to `map`_**\n",
    "\n",
    "Let's create the same RDD using a `anonymous` function.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "f17bOf9I8h5u",
    "outputId": "5c589cb7-ab3f-41ec-d922-9f5bf05372c3"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val pluralLambdaRDD = wordsRDD.map(???)\n",
    "pluralLambdaRDD.collect().foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "WTcFb4dn8h5z",
    "outputId": "fc432933-38d0-4d71-8e9b-443d04b25a62"
   },
   "outputs": [],
   "source": [
    "//TEST Pass an anonymous function to map (1d)\n",
    "assertResult(List(\"cats\", \"elephants\", \"rats\", \"rats\", \"cats\"), \"incorrect values for pluralLambdaRDD (1d)\") {\n",
    "    pluralLambdaRDD.collect.toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Coy_xwMT8h52"
   },
   "source": [
    "**_(1e) Length of each word_**\n",
    "\n",
    "Now use `map()` and an `anonymous` function to return the number of characters in each word.  We'll `collect` this result directly into a variable.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "7w-AnSRD8h52",
    "outputId": "b3c57a95-d047-480b-ac99-012c17f57453"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val pluralLengths = pluralRDD\n",
    "  ???\n",
    "  .collect\n",
    "pluralLengths.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "AK3Z4SED8h56",
    "outputId": "a97d43e9-b622-43ae-e01c-361ea1bdf173"
   },
   "outputs": [],
   "source": [
    "// TEST Length of each word (1e)\n",
    "assertResult(List(4, 9, 4, 4, 4), \"incorrect values for pluralLengths\") {\n",
    "  pluralLengths.toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "VPKz10jR8h6A"
   },
   "source": [
    "**_(1f) Pair RDDs_**\n",
    "\n",
    "The next step in writing our word counting program is to create a new type of RDD, called a pair RDD. A pair RDD is an RDD where each element is a pair tuple `(k, v)` where `k` is the key and `v` is the value. In this example, we will create a pair consisting of `('<word>', 1)` for each word element in the RDD.\n",
    "\n",
    "We can create the pair RDD using the `map()` transformation with an `anonymous` function to create a new RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "i82jdnpD8h6A",
    "outputId": "c371a81e-1c89-42b4-e5e6-bdda2c47fdc3"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val wordPairs = wordsRDD.???\n",
    "wordPairs.collect.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "6Mc163jN8h6D",
    "outputId": "0a4a99fe-b686-4cff-f905-84c16e018765"
   },
   "outputs": [],
   "source": [
    "assertResult(List((\"cat\", 1), (\"elephant\", 1), (\"rat\", 1), (\"rat\", 1), (\"cat\", 1)), \"incorrect values for wordPairs\") {\n",
    "  wordPairs.collect.toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "uslTEwd_8h6H"
   },
   "source": [
    "## 2. Counting with pair RDDs "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "IWdKbata8h6I"
   },
   "source": [
    "Now, let's count the number of times a particular word appears in the RDD. There are multiple ways to perform the counting, but some are much less efficient than others.\n",
    "\n",
    "A naive approach would be to `collect()` all of the elements and count them in the driver program. While this approach could work for small datasets, we want an approach that will work for any size dataset including terabyte- or petabyte-sized datasets. In addition, performing all of the work in the driver program is slower than performing it in parallel in the workers. For these reasons, we will use data parallel operations."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "dQjPU5Gw8h6J"
   },
   "source": [
    "**_(2a) `groupByKey()` approach_**\n",
    "\n",
    "An approach you might first consider (we'll see shortly that there are better ways) is based on using the [groupByKey()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/rdd/PairRDDFunctions.html#groupByKey():org.apache.spark.rdd.RDD[(K,Iterable[V])]) transformation. As the name implies, the `groupByKey()` transformation groups all the elements of the RDD with the same key into a single list in one of the partitions. There are two problems with using `groupByKey()`:\n",
    "\n",
    "  +  The operation requires a lot of data movement to move all the values into the appropriate partitions.\n",
    "  +  The lists can be very large. Consider a word count of English Wikipedia: the lists for common words (e.g., the, a, etc.) would be huge and could exhaust the available memory in a worker.\n",
    " \n",
    "Use `groupByKey()` to generate a pair RDD of type `('word', Iterable)`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "LtB0-WtI8h6K",
    "outputId": "0fbbdde0-e613-4c61-ccbd-cd6690fe7bf0"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "// Note that groupByKey requires no parameters\n",
    "val wordsGrouped = wordPairs.???\n",
    "val keyValues = wordsGrouped.collect\n",
    "keyValues.foreach(x => println(x._1,x._2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "rlSIbiJD8h6M",
    "outputId": "6e1ce6c0-2d15-4cfd-e6ff-2d32a5fb9cd3"
   },
   "outputs": [],
   "source": [
    "// TEST groupByKey() approach (2a)\n",
    "assertResult(List((\"cat\", List(1, 1)), (\"elephant\", List(1)), (\"rat\", List(1, 1))),\n",
    "\"incorrect value for wordsGrouped\") {\n",
    "  wordsGrouped.mapValues(_.toList).collect.sortBy(_._1).toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "xflhVssG8h6Q"
   },
   "source": [
    "**_(2b) Use `groupByKey()` to obtain the counts_**\n",
    "\n",
    "<p>Using the `groupByKey()` transformation creates an RDD containing 3 elements, each of which is a pair of a word and an Iterable.<br/>\n",
    "Now sum the Iterable using a `map()` transformation.  The result should be a pair RDD consisting of (word, count) pairs.</p>\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Qgd6TUQn8h6Q",
    "outputId": "3c07e7c1-1377-4d4e-b67a-73f93834c654"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val wordCountsGrouped = wordsGrouped.???\n",
    "wordCountsGrouped.collect.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "r_5cCRko8h6T",
    "outputId": "dc32e45e-7c10-4c32-dccf-cb02d871abcd"
   },
   "outputs": [],
   "source": [
    "// TEST Use groupByKey() to obtain the counts (2b)\n",
    "assertResult(List((\"cat\", 2), (\"elephant\", 1), (\"rat\", 2)),\n",
    "\"incorrect value for wordCountsGrouped\") {\n",
    "  wordCountsGrouped.collect.sortBy(_._1).toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Ptkp5wIJ8h6W"
   },
   "source": [
    "**_(2c) Counting using `reduceByKey`_**\n",
    "\n",
    "A better approach is to start from the pair RDD and then use the [reduceByKey()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/rdd/PairRDDFunctions.html#reduceByKey(func:(V,V)=%3EV):org.apache.spark.rdd.RDD[(K,V)]) transformation to create a new pair RDD. The `reduceByKey()` transformation gathers together pairs that have the same key and applies the function provided to two values at a time, iteratively reducing all of the values to a single value. `reduceByKey()` operates by applying the function first within each partition on a per-key basis and then across the partitions, allowing it to scale efficiently to large datasets.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "0I12QiTZ8h6W",
    "outputId": "388efefd-c009-483d-a9fe-8b463a6ee400"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "// Note that reduceByKey takes in a function that accepts two values and returns a single value\n",
    "val wordCounts = wordPairs.???\n",
    "wordCounts.collect.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "7PQ5mj358h6Z",
    "outputId": "43be1422-5175-4b8e-89c4-8cb3961b36b1"
   },
   "outputs": [],
   "source": [
    "// TEST Counting using reduceByKey (2c)\n",
    "assertResult(List((\"cat\", 2), (\"elephant\", 1), (\"rat\", 2)),\n",
    "  \"incorrect value for wordCounts\") {\n",
    "  wordCounts.collect.sortBy(_._1).toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "w7CqmxCp8h6c"
   },
   "source": [
    "**_(2d) All together_**\n",
    "\n",
    "The expert version of the code performs the `map()` to pair RDD, `reduceByKey()` transformation, and `collect` in one statement.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "bj2QnNXD8h6d",
    "outputId": "99a29944-b1fb-44b3-d5e5-13b7d34bee1f"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val wordCountsCollected = wordsRDD\n",
    "                          .???\n",
    "  .collect()\n",
    "wordCountsCollected.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "xjRohIMS8h6g",
    "outputId": "ddd80969-35ab-4d0d-d503-678241299797"
   },
   "outputs": [],
   "source": [
    "// TEST All together (2d)\n",
    "assertResult(List((\"cat\", 2), (\"elephant\", 1), (\"rat\", 2)),\n",
    "\"incorrect value for wordCountsCollected\") {\n",
    "  wordCountsCollected.sortBy(_._1).toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ohsUSZSr8h6j"
   },
   "source": [
    "##  3. Finding unique words and a mean value "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "x3_AG2QR8h6k"
   },
   "source": [
    "**_(3a) Unique words_**\n",
    "\n",
    "Calculate the number of unique words in `wordsRDD`.  You can use other RDDs that you have already created to make this easier.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "4DWA-bzQ8h6k",
    "outputId": "2442ddbf-7d74-49fb-84fc-f10b86697394"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val uniqueWords = ???\n",
    "println(uniqueWords)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "9XMqvRgI8h6n",
    "outputId": "a344ff38-7f83-499f-973d-3dda43875f95"
   },
   "outputs": [],
   "source": [
    "// TEST Unique words (3a)\n",
    "assertResult(3, \"incorrect count of uniqueWords\") {\n",
    "  uniqueWords.toInt\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "aLV9HUeN8h6q"
   },
   "source": [
    "**_(3b) Mean using `reduce`_**\n",
    "\n",
    "<p>Find the mean number of words per unique word in `wordCounts`.<br/>\n",
    "\n",
    "Use a `reduce()` action to sum the counts in `wordCounts` and then divide by the number of unique words.  First `map()` the pair RDD `wordCounts`, which consists of (key, value) pairs, to an RDD of values.</p>\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "cxp2YEYa8h6r",
    "outputId": "58895fa3-02eb-45b5-ac60-023d25d2815d"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "import scala.math.BigDecimal.RoundingMode\n",
    "val totalCount = (wordCounts\n",
    "  .map(???)\n",
    "  .reduce(???))\n",
    "val average = totalCount / ???.toFloat\n",
    "println(totalCount)\n",
    "println(BigDecimal(average).setScale(2, RoundingMode.HALF_UP))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Xv1IjkgT8h6s",
    "outputId": "a39e9656-383e-40a0-e76c-5ac4f2a3d8f6"
   },
   "outputs": [],
   "source": [
    "// TEST Mean using reduce (3b)\n",
    "assertResult( BigDecimal(1.67), \"incorrect value of average\") {\n",
    "  BigDecimal(average).setScale(2, RoundingMode.HALF_UP)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "8rr3tixi8h6v"
   },
   "source": [
    "## 4.  Apply word count to a file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "PgeNV29l8h6x"
   },
   "source": [
    "In this section we will finish developing our word count application.  We'll have to build the `wordCount` function, deal with real world problems like capitalization and punctuation, load in our data source, and compute the word count on the new data.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "MHPFrNTZ8h6y"
   },
   "source": [
    "**_(4a) `wordCount` function_**\n",
    "\n",
    "First, define a function for word counting.  You should reuse the techniques that have been covered in earlier parts of this lab.  This function should take in an RDD that is a list of words like `wordsRDD` and return a pair RDD that has all of the words and their associated counts.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "yok0C93J8h6z",
    "outputId": "09bbf1f9-e7b9-4dad-c71b-77e8b1b5314d"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace <FILL IN> with appropriate code\n",
    "import org.apache.spark.rdd.RDD\n",
    "def wordCount(wordListRDD: RDD[String]) : RDD[(String,Int)] = {\n",
    "  /*Creates a pair RDD with word counts from an RDD of words.\n",
    "  Args:\n",
    "  wordListRDD (RDD of str): An RDD consisting of words.\n",
    "  Returns:\n",
    "  RDD of (str, int): An RDD consisting of (word, count) tuples.\n",
    "  */\n",
    "  ???\n",
    "}\n",
    "wordCount(wordsRDD).collect.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "eBA-bJcK8h61",
    "outputId": "55b9c015-1bda-4128-d216-0034c560dd8b"
   },
   "outputs": [],
   "source": [
    "// TEST wordCount function (4a)\n",
    "assertResult(List((\"cat\", 2), (\"elephant\", 1), (\"rat\", 2)),\n",
    "\"incorrect definition for wordCount function\") {\n",
    "  wordCount(wordsRDD).collect.sortBy(_._1).toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "czWJXvRA8h63"
   },
   "source": [
    "**_(4b) Capitalization and punctuation_**\n",
    "\n",
    "Real world files are more complicated than the data we have been using in this lab. Some of the issues we have to address are:\n",
    "\n",
    "  +  Words should be counted independent of their capitialization (e.g., Spark and spark should be counted as the same word).\n",
    "  +  All punctuation should be removed.\n",
    "  +  Any leading or trailing spaces on a line should be removed.\n",
    " \n",
    "Define the function `removePunctuation` that converts all text to lower case, removes any punctuation, and removes leading and trailing spaces.  Use `replaceAll` to remove any text that is not a letter, number, or space.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "_gx5oDGL8h64",
    "outputId": "8fbc264d-6b56-46a3-b3b4-19ab363c1ae7"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace <FILL IN> with appropriate code\n",
    "val removePunctuation = (text:String) => {\n",
    "  /*\n",
    "Removes punctuation, changes to lower case, and strips leading and trailing spaces.\n",
    "Note:\n",
    "    Only spaces, letters, and numbers should be retained.  Other characters should should be\n",
    "    eliminated (e.g. it's becomes its).  Leading and trailing spaces should be removed after\n",
    "    punctuation is removed.\n",
    "Args:\n",
    "    text (str): A string.\n",
    "Returns:\n",
    "    str: The cleaned up string.\n",
    "   */\n",
    "  val text_no_punctuation = text.replaceAll(\"[^a-zA-Z0-9 ]+\",\"\")\n",
    "  ???\n",
    "}\n",
    "println(removePunctuation(\"Hi, you!\"))\n",
    "println(removePunctuation(\" No under_score!\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "DTJJEOqe8h66",
    "outputId": "43600bd8-96ff-44c8-c166-0f2c266542f2"
   },
   "outputs": [],
   "source": [
    "assertResult(\"the elephants 4 cats\",\n",
    "  \"incorrect definition for removePunctuation function\") {\n",
    "  removePunctuation(\" The Elephant's 4 cats. \")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "XknhAGRu8h7E"
   },
   "source": [
    "**_(4c) Load a text file_**\n",
    "\n",
    "For the next part of this lab, we will use the [Complete Works of William Shakespeare](http://www.gutenberg.org/ebooks/100) from [Project Gutenberg](https://www.gutenberg.org/). To convert a text file into an RDD, we use the `SparkContext.textFile()` method. We also apply the recently defined `removePunctuation()` function using a `map()` transformation to strip out the punctuation and change all text to lowercase.  Since the file is large we use `take(15)`, so that we only print 15 lines.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Qql3fAYr8h7E",
    "outputId": "512344e7-32a5-4ebe-b459-bc8161ae4d6e"
   },
   "outputs": [],
   "source": [
    "// Just run this code\n",
    "val fileName = \"data/shakespeare.txt\"\n",
    "val shakespeareRDD = (sc\n",
    "  .textFile(fileName, 8)\n",
    "  .map(removePunctuation))\n",
    "shakespeareRDD\n",
    "  .zipWithIndex()  // to (line, lineNum)\n",
    "  .map(x => s\"${x._2}:${x._1}\")  // to 'lineNum: line'\n",
    ".take(15).foreach(println)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ozASHMVd8h7I"
   },
   "source": [
    "**_(4d) Words from lines_**\n",
    "\n",
    "Before we can use the `wordcount()` function, we have to address two issues with the format of the RDD:\n",
    "\n",
    "  +  The first issue is that  that we need to split each line by its spaces.\n",
    "  +  The second issue is we need to filter out empty lines.\n",
    " \n",
    "Apply a transformation that will split each element of the RDD by its spaces. For each element of the RDD, you should apply the string [split()](https://docs.oracle.com/javase/8/docs/api/java/lang/String.html#split-java.lang.String-) function. You might think that a `map()` transformation is the way to do this, but think about what the result of the `split()` function will be. Use an other transformation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "FILBtgAh8h7I",
    "outputId": "8dc0818c-be89-454d-8592-33c49dc7bc98"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val shakespeareWordsRDD = shakespeareRDD.???\n",
    "val shakespeareWordCount = shakespeareWordsRDD.count\n",
    "shakespeareWordsRDD.top(5).foreach(println)\n",
    "println(shakespeareWordCount)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "pq0WhF9M8h7L",
    "outputId": "ffe5ffff-6339-4e34-f93a-7f179d75bf55"
   },
   "outputs": [],
   "source": [
    "// TEST Words from lines (4d)\n",
    "// This test allows for leading spaces to be removed either before or after punctuation is removed.\n",
    "assertResult(true, \"incorrect value for shakespeareWordCount\") {\n",
    "  shakespeareWordCount == 927631 || shakespeareWordCount == 928908\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "vuFJZZM_8h7N"
   },
   "source": [
    "**_(4e) Remove empty elements_**\n",
    "\n",
    "The next step is to filter out the empty elements.  Remove all entries where the word is `''`.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "dDG-XPQA8h7N",
    "outputId": "a6d3d0e3-c330-42ab-f900-cce7b1757c9b"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val shakeWordsRDD = shakespeareWordsRDD.???\n",
    "val shakeWordCount = shakeWordsRDD.count\n",
    "println(shakeWordCount)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "3ylcw9V48h7Q",
    "outputId": "62a7275b-dc9c-4adc-e721-a094afb6b1c5"
   },
   "outputs": [],
   "source": [
    "// TEST Remove empty elements (4e)\n",
    "assertResult(882996 , \"incorrect value for shakeWordCount\") {\n",
    "  shakeWordCount.toInt\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "jcRqxmKU8h7R"
   },
   "source": [
    "**_(4f) Count the vocabulary size_**\n",
    "\n",
    "<p> How many different words (vocabulary size) are there in Shakespeare's vocabulary? </p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "lN-iMRrc8h7S",
    "outputId": "ad37f339-3b84-441b-b4bd-83faf98fd6d1"
   },
   "outputs": [],
   "source": [
    "// let us get the complete Shakespare' vocabulary\n",
    "val shakespare_vocab_size = ???\n",
    "println(s\"Shakespeare vocabulary contains $shakespare_vocab_size words\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "DIa-eNdy8h7T",
    "outputId": "78e65e6d-99fd-4d10-e5e4-6a6780bb7003"
   },
   "outputs": [],
   "source": [
    "// TEST Count the vocabulary size (4f)\n",
    "assertResult(28147, \"incorrect value for shakespare_vocab_size\") {\n",
    " shakespare_vocab_size.toInt\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "y8Z3c-KT8h7V"
   },
   "source": [
    "**_(4g) Get the top words_**\n",
    "\n",
    "We now have an RDD that is only words.  Next, let's apply the `wordCount()` function to produce a list of word counts. We can view the top 15 words by using the `takeOrdered()` action; however, since the elements of the RDD are pairs, we need a custom sort function that sorts using the value part of the pair.\n",
    "\n",
    "<p>You'll notice that many of the words are common English words. These are called stopwords. In a later lab, we will see how to eliminate them from the results.<br/>\n",
    "Use the `wordCount()` function and `takeOrdered()` to obtain the fifteen most common words and their counts.</p>\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "-yJHKmX48h7W",
    "outputId": "2f14b164-1fa4-48c0-a891-9fe60eac546b"
   },
   "outputs": [],
   "source": [
    "// TODO Replace ??? with appropriate code\n",
    "val top15WordsAndCounts = ???\n",
    "top15WordsAndCounts.foreach(x => println(s\"${x._1}: ${x._2}\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "skBit9wL8h7Y",
    "outputId": "9a8eed15-21e9-4250-8421-b346bb2f7342"
   },
   "outputs": [],
   "source": [
    "// TEST Count the words (4f)\n",
    "assertResult(\n",
    "  List((\"the\", 27361), (\"and\", 26028), (\"i\", 20681), (\"to\", 19150), (\"of\", 17463),\n",
    "(\"a\", 14593), (\"you\", 13615), (\"my\", 12481), (\"in\", 10956), (\"that\", 10890),\n",
    "(\"is\", 9134), (\"not\", 8497), (\"with\", 7771), (\"me\", 7769), (\"it\", 7678)),\n",
    "\"incorrect value for top15WordsAndCounts\") {\n",
    "  top15WordsAndCounts.toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "tO2W98fy8h7a"
   },
   "source": [
    "# Part 2 : Short tutorial to DataFrames \n",
    "\n",
    "and chaining together transformations and actions\n",
    "\n",
    "## 1. Working with your first DataFrames\n",
    "\n",
    "In Spark, we first create a base [DataFrame](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/index.html#DataFrame=org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]). We can then apply one or more transformations to that base DataFrame. *A DataFrame is immutable, so once it is created, it cannot be changed.* As a result, each transformation creates a new DataFrame. Finally, we can apply one or more actions to the DataFrames.\n",
    "\n",
    "> 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 DataFrames:\n",
    "* Create a collection of 10,000 people \n",
    "* Create a Spark DataFrame 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\n",
    "\n",
    "A DataFrame consists of a series of `Row` objects; each `Row` object has a set of named columns. You can think of a DataFrame as modeling a table, though the data source being processed does not have to be a table.\n",
    "\n",
    "More formally, a DataFrame must have a _schema_, which means it must consist of columns, each of which has a _name_ and a _type_. Some data sources have schemas built into them. Examples include RDBMS databases, Parquet files, and NoSQL databases like Cassandra. Other data sources don't have computer-readable schemas, but you can often apply a schema programmatically."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "QLHIegNZ8h7a"
   },
   "source": [
    "## 2. Create a collection of 10,000 people\n",
    "\n",
    "Normally in this section we would use a third-party Scala library to create a collection of fake person records. Since the import doesn't work well in the Scala version of the notebook, we will load a .csv file containing the records previously and ramdomly generated from the python version of the notebook"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We're going to load a collection of randomly generated people records.\n",
    "In the next section, we'll turn that collection into a DataFrame.\n",
    "We'll use the `spark.implicits._`, because that will give us helpers to create a DataFrame from a List for example.\n",
    "There are other ways to define schemas, though; see\n",
    "the Spark Programming Guide's discussion of [schema inference](http://spark.apache.org/docs/latest/sql-programming-guide.html#inferring-the-schema-using-reflection) for more information. (In here, we can directly create a dataframe from reading the csv file, but for the purpose of this section we will create a dataframe from a RDD)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "ptntAWZk8h7b",
    "outputId": "56c78173-ef4c-48c5-8837-a261b31a2183"
   },
   "outputs": [],
   "source": [
    "import spark.implicits._\n",
    "val dataRDD = sc.textFile(\"data/10000persons.csv\").map(_.split(\";\")).map(x => (x(0),x(1),x(2),x(3),x(4).toInt))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "pLTO_AfZ8h7o"
   },
   "source": [
    "## 3. Distributed data and using a collection to create a DataFrame\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). Even DataFrames are ultimately represented as RDDs, with additional meta-data.\n",
    "\n",
    "<img src=\"http://spark-mooc.github.io/web-assets/images/cs105x/diagram-3b.png\" style=\"width: 900px; float: right; margin: 5px\"/>\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",
    "The figure to the right illustrates how Spark breaks a list of data entries into partitions that are each stored in memory on a worker.\n",
    "\n",
    "\n",
    "To create the DataFrame, we'll use `spark.createDataFrame()`, and we'll pass our array of data in as an argument to that function. Spark will create a new set of input data based on data that is passed in.  A DataFrame requires a _schema_, which is a list of columns, where each column has a name and a type. Our list of data has elements with types (mostly strings, but one integer). We'll supply the rest of the schema and the column names as the second argument to `createDataFrame()`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "yCHt4DUM8h7q"
   },
   "source": [
    "Let's view the help for `createDataFrame()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "B1VlyYDn8h7s"
   },
   "outputs": [],
   "source": [
    "val dataDF = dataRDD.toDF(\"last_name\", \"first_name\", \"ssn\", \"occupation\", \"age\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "O9w33cc-8h7t"
   },
   "source": [
    "Let's take a look at the DataFrame's schema and some of its rows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "WjM8hT8R8h7v",
    "outputId": "8115dec2-ebd5-4a61-accc-8c4861f1abe8"
   },
   "outputs": [],
   "source": [
    "dataDF.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "4MBXaZa_8h7w"
   },
   "source": [
    "How many partitions will the DataFrame be split into?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "8Nfh9AXP8h7w",
    "outputId": "4258bce8-9ff9-40f0-cc02-69bc99ea6f2e"
   },
   "outputs": [],
   "source": [
    "dataDF.rdd.getNumPartitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "496VkJ_S8h7z"
   },
   "source": [
    "## 4. Subtract one from each value using _select_\n",
    "\n",
    "So far, we've created a distributed DataFrame that is splitted 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",
    "One of the most common DataFrame operations is `select()`, and it works more or less like a SQL `SELECT` statement: You can select specific columns from the DataFrame, and you can even use `select()` to create _new_ columns with values that are derived from existing column values. We can use `select()` to create a new column that decrements the value of the existing `age` column.\n",
    "\n",
    "`select()` is a _transformation_. It returns a new DataFrame that captures both the previous DataFrame and the operation to add to the query (`select`, in this case). But it does *not* actually execute anything on the cluster. When transforming DataFrames, we are building up a _query plan_. That query plan will be optimized, implemented (in terms of RDDs), and executed by Spark _only_ when we call an action (see Lazy Evaluation process)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "N2w-GCtc8h7z"
   },
   "outputs": [],
   "source": [
    "// Transform dataDF through a select transformation and rename the newly created '(age -1)' column to 'age'\n",
    "// Because select is a transformation and Spark uses lazy evaluation, no jobs, stages, or tasks will be launched when we run this code.\n",
    "val subDF = dataDF.select(dataDF(\"last_name\"), dataDF(\"first_name\"), dataDF(\"ssn\"), dataDF(\"occupation\"), (dataDF(\"age\") - 1).alias(\"age\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "8AwIX5Ez8h72"
   },
   "source": [
    "##  5.  Use _collect_ or _show_ to view results\n",
    "\n",
    "<img src=\"http://spark-mooc.github.io/web-assets/images/cs105x/diagram-3d.png\" style=\"height:700px;float:right\"/>\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 can call the `collect()` method on our DataFrame.  `collect()` is often used after transformations 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 values returned by the action.  In our example, this means that tasks will now be launched to perform the `createDataFrame`, `select`, and `collect` operations.\n",
    "\n",
    "In the diagram, 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 driver, which creates a list of the values, as shown in the figure below.\n",
    "\n",
    "Now let's run `collect()` on `subDF`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "GTWtcSkv8h72",
    "outputId": "6842ae80-12cc-435d-a75d-1007814fe724"
   },
   "outputs": [],
   "source": [
    "subDF.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "EfxHgs1P8h74",
    "outputId": "96afc59b-3b8b-4db4-d4f2-48bc42389892"
   },
   "outputs": [],
   "source": [
    "// Let's collect the data\n",
    "val results = subDF.take(3)\n",
    "results.foreach(println)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "C_I8b7578h75",
    "outputId": "77f41e29-039e-4e38-e870-509812d61c15"
   },
   "outputs": [],
   "source": [
    "subDF.show() // look at parameters"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "6wmnBBbu8h77"
   },
   "source": [
    "## 6. Use _count_ to get total\n",
    "\n",
    "One of the most basic jobs that we can run is the `count()` job which will count the number of elements in a DataFrame, using the `count()` action. Since `select()` creates a new DataFrame with the same number of elements as the starting DataFrame, we expect that applying `count()` to each DataFrame will return the same result.\n",
    "\n",
    "<img src=\"http://spark-mooc.github.io/web-assets/images/cs105x/diagram-3e.png\" style=\"height:750px;float:right\"/>\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 on the right shows what would happen if we ran `count()` on a small example dataset with just four partitions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "pAqS-_6n8h77",
    "outputId": "2fd4e8c8-5c1e-4340-9a26-a6316411bdaf"
   },
   "outputs": [],
   "source": [
    "println (dataDF.count())\n",
    "println (subDF.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "1QOXavWM8h79"
   },
   "source": [
    "## 7. Apply transformation _filter_ and view results with _collect_\n",
    "\n",
    "Next, we'll create a new DataFrame that only contains the people whose ages are less than 10. To do this, we'll use the `filter()` transformation. (You can also use `where()`, an alias for `filter()`, if you prefer something more SQL-like). The `filter()` method is a transformation operation that creates a new DataFrame from the input DataFrame, keeping only values that match the filter expression.\n",
    "\n",
    "The figure shows how this might work on the small four-partition dataset.\n",
    "\n",
    "<img src=\"http://spark-mooc.github.io/web-assets/images/cs105x/diagram-3f.png\" style=\"height:700px;float:right\"/>\n",
    "\n",
    "\n",
    "- To view the filtered list of elements less than 10, 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 DataFrame to the driver program."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "EZ5Sh7Yy8h7-",
    "outputId": "4e048b6d-8c35-4897-a2bf-132eaabe9d82"
   },
   "outputs": [],
   "source": [
    "val filteredDF = subDF.filter(subDF(\"age\") < 10)\n",
    "filteredDF.show(truncate=false)\n",
    "filteredDF.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "_pqpqzYt8h7_"
   },
   "source": [
    "## 8. Additional DataFrame actions\n",
    "\n",
    "Let's investigate some additional actions:\n",
    "\n",
    "* [first()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#first():T)\n",
    "* [take()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#take(n:Int):Array[T])\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 actions like `first()`, `take()`, and `show()`. Note that for the `first()` and `take()` actions, the elements that are returned depend on how the DataFrame 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 DataFrame. The `first()` action returns the first element of a DataFrame, and is equivalent to `take(1)[0]`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "XbgKm9fE8h7_",
    "outputId": "3c5729cc-77d9-49fd-9985-efc6ca8cf06e"
   },
   "outputs": [],
   "source": [
    "println(s\"first: ${filteredDF.first()}\")\n",
    "println(s\"Four of them: ${filteredDF.take(4).mkString}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "U8aktY4s8h8A"
   },
   "source": [
    "## 9. Additional DataFrame transformations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "DFu2ZOJ08h8B"
   },
   "source": [
    "**_orderBy_**\n",
    "\n",
    "[`orderBy()`](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#orderBy(sortExprs:org.apache.spark.sql.Column*):org.apache.spark.sql.Dataset[T]):org.apache.spark.sql.Dataset[T]) allows you to sort a DataFrame by one or more columns, producing a new DataFrame.\n",
    "\n",
    "For example, let's get the first five oldest people in the original (unfiltered) DataFrame. We can use the `orderBy()` transformation. `orderBy` takes one or more columns, either as _names_ (strings) or as `Column` objects. To get a `Column` object, we use one of two notations on the DataFrame:\n",
    "\n",
    "* Subscript notation: `filteredDF(\"age\")`\n",
    "\n",
    "Both of those syntaxes return a `Column`, which has additional methods like `desc()` (for sorting in descending order) or `asc()` (for sorting in ascending order, which is the default).\n",
    "\n",
    "Here are some examples:\n",
    "\n",
    "```\n",
    "dataDF.orderBy(dataDF(\"age\"))  // sort by age in ascending order; returns a new DataFrame\n",
    "dataDF.orderBy(dataDF(\"last_name\").desc) // sort by last name in descending order\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "lfxYi03a8h8C",
    "outputId": "31c19bb3-f557-4b0d-d0c0-e9280f676c2b"
   },
   "outputs": [],
   "source": [
    "// Get the five oldest people in the list. To do that, sort by age in descending order.\n",
    "dataDF.orderBy(dataDF(\"age\").desc).take(5).foreach(println)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "gPEMqdZT8h8C"
   },
   "source": [
    "**_distinct_ and _dropDuplicates_**\n",
    "\n",
    "[`distinct()`](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#distinct():org.apache.spark.sql.Dataset[T]) filters out duplicate rows, and it considers all columns. Since our data is completely randomly generated (by `fake-factory`), it's extremely unlikely that there are any duplicate rows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "HNWWRBrI8h8C",
    "outputId": "e9d6de1b-e3ca-4918-8f20-496a244f699e"
   },
   "outputs": [],
   "source": [
    "println(dataDF.count())\n",
    "println(dataDF.distinct().count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "vD1jWTyl8h8F"
   },
   "source": [
    "**_drop_**\n",
    "\n",
    "[`drop()`](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#drop(col:org.apache.spark.sql.Column):org.apache.spark.sql.DataFrame) is like the opposite of `select()`: Instead of selecting specific columns from a DataFrame, it drops a specifed column from a DataFrame.\n",
    "\n",
    "Here's a simple use case: Suppose you're reading from a 1,000-column CSV file, and you have to get rid of five of the columns. Instead of selecting 995 of the columns, it's easier just to drop the five you don't want."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "LFPhWOmO8h8F"
   },
   "source": [
    "**_groupBy_**\n",
    "\n",
    "[`groupBy()`](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#groupBy(cols:org.apache.spark.sql.Column*):org.apache.spark.sql.RelationalGroupedDataset) is one of the most powerful transformations. It allows you to perform aggregations on a DataFrame.\n",
    "\n",
    "Unlike other DataFrame transformations, `groupBy()` does _not_ return a DataFrame. Instead, it returns a special [RelationalGroupedDataset](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/RelationalGroupedDataset.html) object that contains various aggregation functions.\n",
    "\n",
    "The most commonly used aggregation function is [count()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/RelationalGroupedDataset.html#count():org.apache.spark.sql.DataFrame),\n",
    "but there are others (like [sum()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/RelationalGroupedDataset.html#sum(colNames:String*):org.apache.spark.sql.DataFrame), [max()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/RelationalGroupedDataset.html#max(colNames:String*):org.apache.spark.sql.DataFrame), and [avg()](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/RelationalGroupedDataset.html#avg(colNames:String*):org.apache.spark.sql.DataFrame).\n",
    "\n",
    "These aggregation functions typically create a new column and return a new DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "C1WLCveK8h8F",
    "outputId": "d12befca-8047-4b9f-93a1-e1aebb2f3480"
   },
   "outputs": [],
   "source": [
    "dataDF.groupBy(\"occupation\").count().show(truncate=false)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "WNc93gtz8h8H"
   },
   "source": [
    "# Part 3 : WordCount with Dataframes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "2-G_mFNK8h8H"
   },
   "source": [
    "**_Create a DataFrame_**\n",
    "\n",
    "We'll start by generating a base DataFrame by using a List of Strings and the `toDF` method.  Then we'll print out the type and schema of the DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "Z_f3arOK8h8I",
    "outputId": "ea0a0268-a0ae-424a-ceba-a681f8b0c961"
   },
   "outputs": [],
   "source": [
    "import spark.implicits._\n",
    "import org.apache.spark.sql._\n",
    "\n",
    "val wordsDF = List(\"cat\",\"elephant\",\"rat\",\"rat\",\"cat\").toDF(\"word\")\n",
    "wordsDF.show()\n",
    "println(wordsDF.getClass)\n",
    "wordsDF.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "4PxEEBZs8h8J"
   },
   "source": [
    "## 1. Counting with Spark SQL and DataFrames \n",
    "\n",
    "Now, let's count the number of times a particular word appears in the 'word' column. There are multiple ways to perform the counting, but some are much less efficient than others.\n",
    "\n",
    "A naive approach would be to call `collect` on all of the elements and count them in the driver program. While this approach could work for small datasets, we want an approach that will work for any size dataset including terabyte- or petabyte-sized datasets. In addition, performing all of the work in the driver program is slower than performing it in parallel in the workers. For these reasons, we will use data parallel operations.\n",
    "\n",
    "**_Using `groupBy` and `count`_**\n",
    "\n",
    "Using DataFrames, we can preform aggregations by grouping the data using the [`groupBy` function](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#groupBy(cols:org.apache.spark.sql.Column*) on the DataFrame.  Using `groupBy` returns a [RelationalGroupedDataset object](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/RelationalGroupedDataset.html) and we can use the functions available for `RelationalGroupedDataset` to aggregate the groups.  For example, we can call `avg` or `count` on a `GroupedData` object to obtain the average of the values in the groups or the number of occurrences in the groups, respectively.\n",
    "\n",
    "To find the counts of words, group by the words and then use the [`count` function](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/RelationalGroupedDataset.html#count():org.apache.spark.sql.DataFrame) to find the number of times that words occur."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Us2mEXMU8h8K"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val wordCountsDF = wordsDF.???\n",
    "wordCountsDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "TX9j_PcR8h8O"
   },
   "outputs": [],
   "source": [
    "// TEST groupBy and count (2a)\n",
    "import org.apache.spark.sql.Row\n",
    "assertResult(List((\"cat\",2), (\"elephant\",1), (\"rat\", 2)), \"incorrect counts for wordCountsDF\") {\n",
    "  wordCountsDF.collect().map({case Row(s : String, c : Long) => (s,c.toInt)}).sortBy(_._1).toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ZGc_OMuE8h8Q"
   },
   "source": [
    "## 2. Finding unique words and a mean value \n",
    "\n",
    "**_Unique words_**\n",
    "\n",
    "Calculate the number of unique words in `wordsDF`.  You can use other DataFrames that you have already created to make this easier.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "ix35g6_M8h8Q"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val uniqueWordsCount = ???\n",
    "println(uniqueWordsCount)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "lGdqUuQs8h8R"
   },
   "outputs": [],
   "source": [
    "// TEST Unique words\n",
    "assertResult(3, \"incorrect count of unique words\") {\n",
    "  uniqueWordsCount.toInt\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "FjGZg0jC8h8S"
   },
   "source": [
    "##  Apply word count to a file "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "1odGo_rS8h8T"
   },
   "source": [
    "In this section we will finish developing our word count application.  We'll have to build the `wordCount` function, deal with real world problems like capitalization and punctuation, load in our data source, and compute the word count on the new data.\n",
    "\n",
    "**_The `wordCount` function_**\n",
    "\n",
    "First, define a function for word counting.  You should reuse the techniques that have been covered in earlier parts of this lab.  This function should take in a DataFrame that is a list of words like `wordsDF` and return a DataFrame that has all of the words and their associated counts."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "RNyLsDi48h8U"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "def wordCountDF(wordListDF : DataFrame) = {\n",
    "  /*\n",
    "Creates a DataFrame with word counts.\n",
    "Args:\n",
    "    wordListDF (DataFrame of str): A DataFrame consisting of one string column c\n",
    "Returns:\n",
    "    DataFrame of (str, int): A DataFrame containing 'word' and 'count' columns.\n",
    "   */\n",
    " ???\n",
    "}\n",
    "wordCountDF(wordsDF).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "uinvC9Bi8h8V"
   },
   "outputs": [],
   "source": [
    "// TEST wordCount function (4a)\n",
    "assertResult(List((\"cat\", 2), (\"elephant\", 1), (\"rat\", 2)), \"incorrect definition for wordCountDF function\") {\n",
    "  wordCountDF(wordsDF).collect().map({case Row(s : String, c : Long) => (s,c.toInt)}).sortBy(_._1).toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "BFTKQH8R8h8W"
   },
   "source": [
    "**_Capitalization and punctuation_**\n",
    "\n",
    "Real world files are more complicated than the data we have been using in this lab. Some of the issues we have to address are:\n",
    "  + Words should be counted independent of their capitialization (e.g., Spark and spark should be counted as the same word).\n",
    "  + All punctuation should be removed.\n",
    "  + Any leading or trailing spaces on a line should be removed.\n",
    "\n",
    "Define the function `removePunctuation` that converts all text to lower case, removes any punctuation, and removes leading and trailing spaces.  \n",
    "\n",
    "To complete this task, you should use Pyspark implemented functions such as :\n",
    "\n",
    "[regexp_replace](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/RelationalGroupedDataset.html#count():org.apache.spark.sql.DataFrame) module to remove any text that is not a letter, number, or space. If you are unfamiliar with regular expressions, refer to the `removePunctuation` function defined previously in this notebook. You may also want to review [this tutorial](https://developers.google.com/edu/python/regular-expressions) from Google.  Also, [this website](https://regex101.com/#python) is  a great resource for debugging your regular expression. \n",
    "\n",
    "You should also use the `trim` and `lower` functions found in [org.apache.spark.sql.functions] in order to lower text and remove leading and trailing spaces. (https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/functions$.html).\n",
    "\n",
    "> Note that you shouldn't use any RDD operations or need to create custom user defined functions (udfs) to accomplish this task"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "HJr7c6kg8h8X",
    "outputId": "93843658-47ae-47f0-b7c2-7855b9ab994d"
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.sql.functions._\n",
    "// TODO: Replace ??? with appropriate code\n",
    "def removePunctuationDF(coll : Column) = {\n",
    "  /*\n",
    "Removes punctuation, changes to lower case, and strips leading and trailing spaces.\n",
    "Note:\n",
    "    Only spaces, letters, and numbers should be retained.  Other characters should should be\n",
    "    eliminated (e.g. it's becomes its).  Leading and trailing spaces should be removed after\n",
    "    punctuation is removed.\n",
    "Args:\n",
    "    column (Column): A Column containing a sentence.\n",
    "Returns:\n",
    "    Column: A Column named 'sentence' with clean-up operations applied.\n",
    "   */\n",
    "  lower(trim(regexp_replace(coll, \"[^a-zA-Z0-9 ]+\",\"\")))\n",
    "}\n",
    "val sentenceDF = List((\"Hi, you!\"),(\" No under_score!\"), (\" *      Remove punctuation then spaces  * \"))\n",
    "  .toDF(\"sentence\")\n",
    "sentenceDF.show(truncate = false)\n",
    "sentenceDF\n",
    "  .select(removePunctuationDF(sentenceDF(\"sentence\")))\n",
    "  .show(truncate = false)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "o0gL6iSp8h8Y"
   },
   "source": [
    "**_Load a text file_**\n",
    "\n",
    "For the next part of this lab, we will use the [Complete Works of William Shakespeare](http://www.gutenberg.org/ebooks/100) from [Project Gutenberg](https://www.gutenberg.org/). To convert a text file into a DataFrame, we use the `spark.read.text()` method. We also apply the recently defined `removePunctuation()` function using a `select()` transformation to strip out the punctuation and change all text to lower case.  Since the file is large we use `show(15)`, so that we only print 15 lines."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "pcdGLZG88h8Z"
   },
   "outputs": [],
   "source": [
    "val shakespeareDF = spark.sqlContext.read.text(fileName).select(removePunctuationDF(col(\"value\")))\n",
    "shakespeareDF.show(15, truncate = false)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "jCMNhPQH8h8a"
   },
   "source": [
    "**_Words from lines_**\n",
    "\n",
    "Before we can use the `wordcount()` function, we have to address two issues with the format of the DataFrame:\n",
    "  + The first issue is that  that we need to split each line by its spaces.\n",
    "  + The second issue is we need to filter out empty lines or words.\n",
    "\n",
    "Apply a transformation that will split each 'sentence' in the DataFrame by its spaces, and then transform from a DataFrame that contains lists of words into a DataFrame with each word in its own row.  To accomplish these two tasks you can use the `split` and `explode` functions found in [org.apache.spark.sql.functions](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/functions$.html).\n",
    "\n",
    "Once you have a DataFrame with one word per row you can apply the [DataFrame operation `where`](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#where(condition:org.apache.spark.sql.Column):org.apache.spark.sql.Dataset[T]) to remove the rows that contain ''.\n",
    "\n",
    "> Note that `shakeWordsDF` should be a DataFrame with one column named `word`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "SOsrvWDK8h8a"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val shakeWordsDF :DataFrame = (shakespeareDF.???)\n",
    "shakeWordsDF.show()\n",
    "val shakeWordsDFCount = shakeWordsDF.count()\n",
    "println(shakeWordsDFCount)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "rGI_SA068h8b"
   },
   "outputs": [],
   "source": [
    "// TEST Remove empty elements (4d)\n",
    "assertResult(882996, \"incorrect value for shakeWordCount\") {\n",
    "  shakeWordsDFCount.toInt\n",
    "}\n",
    "assertResult(List(\"word\"),\"shakeWordsDF should only contain the Column 'word'\") {\n",
    "  shakeWordsDF.columns.toList\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Bt7RCjUP8h8d"
   },
   "source": [
    "**_Count the words_**\n",
    "\n",
    "We now have a DataFrame that is only words.  Next, let's apply the `wordCount()` function to produce a list of word counts. We can view the first 20 words by using the `show()` action; however, we'd like to see the words in descending order of count, so we'll need to apply the [`orderBy` DataFrame method](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#orderBy(sortCol:String,sortCols:String*):org.apache.spark.sql.Dataset[T]) to first sort the DataFrame that is returned from `wordCount()`.\n",
    "\n",
    "You'll notice that many of the words are common English words. These are called stopwords. In a later lab, we will see how to eliminate them from the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "k3RxV74F8h8d"
   },
   "outputs": [],
   "source": [
    "// TODO: Replace ??? with appropriate code\n",
    "val topWordsAndCountsDF = ???\n",
    "topWordsAndCountsDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "zSYJbINI8h8f"
   },
   "outputs": [],
   "source": [
    "// TEST Count the words (4e)\n",
    "assertResult(\n",
    "  List((\"the\", 27361), (\"and\", 26028), (\"i\", 20681), (\"to\", 19150), (\"of\", 17463),\n",
    "    (\"a\", 14593), (\"you\", 13615), (\"my\", 12481), (\"in\", 10956), (\"that\", 10890),\n",
    "    (\"is\", 9134), (\"not\", 8497), (\"with\", 7771), (\"me\", 7769), (\"it\", 7678)),\n",
    "  \"incorrect value for top15WordsAndCountsDF\") {\n",
    "  topWordsAndCountsDF.take(15).map({\n",
    "    case Row(s:String,c:Long) => (s,c.toInt)\n",
    "  }).toList\n",
    "}"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "colab": {
   "provenance": []
  },
  "kernelspec": {
   "display_name": "Scala (2.12)",
   "language": "scala",
   "name": "scala212"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".sc",
   "mimetype": "text/x-scala",
   "name": "scala",
   "nbconvert_exporter": "script",
   "version": "2.12.12"
  },
  "latex_envs": {
   "LaTeX_envs_menu_present": true,
   "autocomplete": true,
   "bibliofile": "biblio.bib",
   "cite_by": "apalike",
   "current_citInitial": 1,
   "eqLabelWithNumbers": true,
   "eqNumInitial": 0,
   "hotkeys": {
    "equation": "ctrl-e"
   },
   "labels_anchors": false,
   "latex_user_defs": false,
   "report_style_numbering": false,
   "user_envs_cfg": false
  },
  "nav_menu": {
   "height": "371px",
   "width": "433px"
  },
  "toc": {
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": "block",
   "toc_window_display": false
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}