02_Low_Level_APIs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Spark Image](https://upload.wikimedia.org/wikipedia/commons/thumb/f/f3/Apache_Spark_logo.svg/1200px-Apache_Spark_logo.svg.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Low-Level Unstructured APIs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook we will discuss the oldest fundamental concept in spark called *RDDs (Resilient distributed\n",
    "datasets)*.<br> \n",
    "To truly understand how Spark works, `you must understand the essence of RDDs`. They provide an extremely solid foundation that other abstractions are built upon. Starting with Spark 2.0, Spark users will have fewer needs for directly interacting with RDD, but having a strong mental model of how RDD works is essential. `In a nutshell, Spark revolves around the concept of RDDs`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction to RDDs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "An RDD in Spark is simply an immutable distributed collection of objects. Each is split into multiple partitions, which may be computed on different nodes of the cluster.<br>\n",
    "RDDs are `immutable`, `fault-tolerant`, `parallel data structures` that let users explicitly persist intermediate results `in memory`, control their partitioning to optimize data placement, and `manipulate` them using a rich set of `operators`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Immutable\n",
    "\n",
    "RDDs are designed to be immutable, which means you `can’t` specifically `modify a particular row` in the dataset represented by that RDD. You can call one of the available RDD operations to manipulate the rows in the RDD into the way you want, but that operation will `return a new RDD`. The `basic RDD will stay unchanged`, and the new RDD\n",
    "will contain the data in the way that you want. *Spark leverages Immutability to efficiently provide the fault tolerance capability.* "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fault Tolerant\n",
    "\n",
    "The ability to process multiple datasets in parallel usually requires a cluster of machines to host and execute the computational logic. If one or more machices dies due to unexpected circumstances then whats happens to the data in those machines?.  Spark automatically takes care of handling the failure on behalf of its users by rebuilding the failed portion using the lineage information."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Parallel Data Structures\n",
    "\n",
    "Suppose you have huge amount of data and you need process each and every row of the datset. One solution will be to iterate over each row and process it one by one. But that would be very slow. So instead we will divide the huge chuck of Data in smaller chunks of Data. Each chunk contains a collection of rows, and all the chunks are being processed in parallel. This is where the phrase parallel data structures comes from."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## In-Memory Computing\n",
    "\n",
    "The idea of speeding up the computation of large datasets that reside on disks in a parallelized manner using a cluster of machines was introduced by a MapReduce paper from Google. RDD pushes the speed boundary by introducing a novel idea, which is the ability to do distributed in-memory computation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RDD Operations\n",
    "\n",
    "RDDs provide a rich set of commonly needed data processing operations. They include the ability to perform data transformation, filtering, grouping, joining, aggregation, sorting, and counting.<br>\n",
    "Each row in a dataset is represented as a Java object, and the structure of this Java object is opaque to Spark. The user of RDD has complete control over how to manipulate this Java object. This flexibility comes with a lot of responsibilities, meaning some of the commonly needed operations such as the computing average will have to be handcrafted. Higher-level abstractions such as the Spark SQL component will provide this functionality out of the box.<br>\n",
    "\n",
    "***The RDD operations are classified into two types: `transformations` and `actions`***\n",
    "\n",
    "| Type | Evaluation | Returned Value |\n",
    "|--|--|--|\n",
    "| Transformation | Lazy | Another RDD |\n",
    "| Action | Eager | Some result or write result to disk |\n",
    "\n",
    "Transformation operations are lazily evaluated, meaning Spark will delay the evaluations of the invoked operations until an action is taken. In other words, the transformation operations merely record the specified transformation logic and will apply them at a later point. On the other hand, invoking an action operation will trigger the evaluation of all the transformations that preceded it, and it will either return some result to the driver or write data to a storage system, such as HDFS or the local file system."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initialising Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The programming language Python is used for the implementation in this course - for this we use 'pyspark. (PySpark documentation https://spark.apache.org/docs/latest/api/python/)\n",
    "PySpark is an interface for Apache Spark in Python. It not only allows you to write Spark applications using Python APIs, but also provides the PySpark shell for interactively analyzing your data in a distributed environment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# ipmort libraries from pyspark \n",
    "from pyspark import SparkConf, SparkContext\n",
    "\n",
    "# set values for Spark configuration\n",
    "conf = SparkConf().setMaster(\"local\").setAppName(\"Tutorial\")\n",
    "\n",
    "# get (if already running) or create a Spark Context\n",
    "sc = SparkContext.getOrCreate(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check (try) if Spark context variable (sc) exists and print information about the Spark context\n",
    "try:\n",
    "    sc\n",
    "except NameError:\n",
    "    print(\"Spark context does not context exist. Please create Spark context first (run cell above).\")\n",
    "else:\n",
    "    configurations = sc.getConf().getAll()\n",
    "    for item in configurations: print(item)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# print link to Spark UI, Version, Master and AppName\n",
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating RDDs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**There are two ways to create RDDs:**\n",
    "\n",
    "**`The first way to create an RDD is to parallelize an python object, meaning converting it to a distributed dataset that can be operated in parallel.`**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a list of strings\n",
    "stringList = [\"Spark is awesome\",\"Spark is cool\"]\n",
    "# covert list of strings into a Spark RDD\n",
    "stringRDD = sc.parallelize(stringList)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# output RDD information\n",
    "stringRDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*One thing to notice is that you are not able to see the output, because of Spark's Lazy evaluation utill you call an action on that RDD.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# retrieve all the elements of the RDD/DataFrame/Dataset (from all nodes)\n",
    "stringRDD.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*.collect() is an `action` as it name suggests it collects all the rows from each of the partitions in an RDD and brings them over to the driver program.*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**`The second way to create an RDD is to read a dataset from a storage system, which can be a local computer file system, HDFS, Cassandra, Amazon S3, and so on.`**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# read text file inro RDD\n",
    "ratings = sc.textFile(\"data/ml-1m/ratings.dat\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# retrieve all the elements of the RDD/DataFrame/Dataset (from all nodes) and output first 5 rows\n",
    "ratings.collect()[:5]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this particular example we had 1M rows calling .collect() of it didn't take lot of time but If your RDD contains 100 billion rows, then it is not a good idea to invoke the collect action because the driver program most likely doesn’t have sufficient memory to hold all those rows. As a result, the driver will most likely run into an out-of-memory error and your Spark application or shell will die. This action is typically used once the RDD is filtered down to a smaller size that can fit the memory size of the driver program. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# take the first 5 elements of the RDD\n",
    "ratings.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformations\n",
    "\n",
    "Transformations are operations on RDDs that return a new RDD. Transformed RDDs are computed lazily, only when you\n",
    "use them in an action.\n",
    "\n",
    "Following Table describes commonly used transformations.\n",
    "\n",
    "<table>\n",
    "<tbody><tr><th style=\"width:25%\">Transformation</th><th>Meaning</th></tr>\n",
    "<tr>\n",
    "  <td> <b>map</b>(<i>func</i>) </td>\n",
    "  <td> Return a new distributed dataset formed by passing each element of the source through a function <i>func</i>. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>filter</b>(<i>func</i>) </td>\n",
    "  <td> Return a new dataset formed by selecting those elements of the source on which <i>func</i> returns true. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>flatMap</b>(<i>func</i>) </td>\n",
    "  <td> Similar to map, but each input item can be mapped to 0 or more output items (so <i>func</i> should return a Seq rather than a single item). </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>mapPartitions</b>(<i>func</i>) <a name=\"MapPartLink\"></a> </td>\n",
    "  <td> Similar to map, but runs separately on each partition (block) of the RDD, so <i>func</i> must be of type\n",
    "    Iterator&lt;T&gt; =&gt; Iterator&lt;U&gt; when running on an RDD of type T. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>mapPartitionsWithIndex</b>(<i>func</i>) </td>\n",
    "  <td> Similar to mapPartitions, but also provides <i>func</i> with an integer value representing the index of\n",
    "  the partition, so <i>func</i> must be of type (Int, Iterator&lt;T&gt;) =&gt; Iterator&lt;U&gt; when running on an RDD of type T.\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>sample</b>(<i>withReplacement</i>, <i>fraction</i>, <i>seed</i>) </td>\n",
    "  <td> Sample a fraction <i>fraction</i> of the data, with or without replacement, using a given random number generator seed. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>union</b>(<i>otherDataset</i>) </td>\n",
    "  <td> Return a new dataset that contains the union of the elements in the source dataset and the argument. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>intersection</b>(<i>otherDataset</i>) </td>\n",
    "  <td> Return a new RDD that contains the intersection of elements in the source dataset and the argument. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>distinct</b>([<i>numPartitions</i>])) </td>\n",
    "  <td> Return a new dataset that contains the distinct elements of the source dataset.</td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>groupByKey</b>([<i>numPartitions</i>]) <a name=\"GroupByLink\"></a> </td>\n",
    "  <td> When called on a dataset of (K, V) pairs, returns a dataset of (K, Iterable&lt;V&gt;) pairs. <br>\n",
    "    <b>Note:</b> If you are grouping in order to perform an aggregation (such as a sum or\n",
    "      average) over each key, using <code>reduceByKey</code> or <code>aggregateByKey</code> will yield much better\n",
    "      performance.\n",
    "    <br>\n",
    "    <b>Note:</b> By default, the level of parallelism in the output depends on the number of partitions of the parent RDD.\n",
    "      You can pass an optional <code>numPartitions</code> argument to set a different number of tasks.\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>reduceByKey</b>(<i>func</i>, [<i>numPartitions</i>]) <a name=\"ReduceByLink\"></a> </td>\n",
    "  <td> When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function <i>func</i>, which must be of type (V,V) =&gt; V. Like in <code>groupByKey</code>, the number of reduce tasks is configurable through an optional second argument. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>aggregateByKey</b>(<i>zeroValue</i>)(<i>seqOp</i>, <i>combOp</i>, [<i>numPartitions</i>]) <a name=\"AggregateByLink\"></a> </td>\n",
    "  <td> When called on a dataset of (K, V) pairs, returns a dataset of (K, U) pairs where the values for each key are aggregated using the given combine functions and a neutral \"zero\" value. Allows an aggregated value type that is different than the input value type, while avoiding unnecessary allocations. Like in <code>groupByKey</code>, the number of reduce tasks is configurable through an optional second argument. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>sortByKey</b>([<i>ascending</i>], [<i>numPartitions</i>]) <a name=\"SortByLink\"></a> </td>\n",
    "  <td> When called on a dataset of (K, V) pairs where K implements Ordered, returns a dataset of (K, V) pairs sorted by keys in ascending or descending order, as specified in the boolean <code>ascending</code> argument.</td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>join</b>(<i>otherDataset</i>, [<i>numPartitions</i>]) <a name=\"JoinLink\"></a> </td>\n",
    "  <td> When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (V, W)) pairs with all pairs of elements for each key.\n",
    "    Outer joins are supported through <code>leftOuterJoin</code>, <code>rightOuterJoin</code>, and <code>fullOuterJoin</code>.\n",
    "  </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>cogroup</b>(<i>otherDataset</i>, [<i>numPartitions</i>]) <a name=\"CogroupLink\"></a> </td>\n",
    "  <td> When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (Iterable&lt;V&gt;, Iterable&lt;W&gt;)) tuples. This operation is also called <code>groupWith</code>. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>cartesian</b>(<i>otherDataset</i>) </td>\n",
    "  <td> When called on datasets of types T and U, returns a dataset of (T, U) pairs (all pairs of elements). </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>pipe</b>(<i>command</i>, <i>[envVars]</i>) </td>\n",
    "  <td> Pipe each partition of the RDD through a shell command, e.g. a Perl or bash script. RDD elements are written to the\n",
    "    process's stdin and lines output to its stdout are returned as an RDD of strings. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>coalesce</b>(<i>numPartitions</i>) <a name=\"CoalesceLink\"></a> </td>\n",
    "  <td> Decrease the number of partitions in the RDD to numPartitions. Useful for running operations more efficiently\n",
    "    after filtering down a large dataset. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>repartition</b>(<i>numPartitions</i>) </td>\n",
    "  <td> Reshuffle the data in the RDD randomly to create either more or fewer partitions and balance it across them.\n",
    "    This always shuffles all data over the network. <a name=\"RepartitionLink\"></a></td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>repartitionAndSortWithinPartitions</b>(<i>partitioner</i>) <a name=\"Repartition2Link\"></a></td>\n",
    "  <td> Repartition the RDD according to the given partitioner and, within each resulting partition,\n",
    "  sort records by their keys. This is more efficient than calling <code>repartition</code> and then sorting within\n",
    "  each partition because it can push the sorting down into the shuffle machinery. </td>\n",
    "</tr>\n",
    "</tbody></table>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformation Examples"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Map transformation\n",
    "\n",
    "*Return a new RDD by applying a function to each element of this RDD*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# use the already created RDD and convert all letter to uppercase \n",
    "# using the map transformation and the upper function\n",
    "stringRDD_uppercase= stringRDD.map(lambda x: x.upper())\n",
    "# retrieve all the elements of the RDD/DataFrame/Dataset (from all nodes)\n",
    "stringRDD_uppercase.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# implement the function 'alternate_char_upper' which converts every other letter to upper case\n",
    "def alternate_char_upper(text):\n",
    "    new_text= []\n",
    "    for i, character in enumerate(text):\n",
    "        if i % 2 == 0:\n",
    "            new_text.append(character.upper())\n",
    "        else:\n",
    "            new_text.append(character)\n",
    "    return ''.join(new_text)\n",
    "\n",
    "# use the already created RDD and use the map transformation with the 'alternate_char_upper' function\n",
    "stringRDD_alternate_uppercase= stringRDD.map(alternate_char_upper)\n",
    "# retrieve all the elements of the RDD/DataFrame/Dataset (from all nodes)\n",
    "stringRDD_alternate_uppercase.collect()  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Flat Map Transfermation\n",
    "\n",
    "*Return a new RDD by first applying a function to all elements of this RDD, and then flattening the results*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# use the already created RDD and split the string at every ' ' character\n",
    "# using the flatMap transformation and the split function\n",
    "flatMap_Split= stringRDD.flatMap(lambda x: x.split(\" \"))\n",
    "# retrieve all the elements of the RDD/DataFrame/Dataset (from all nodes)\n",
    "flatMap_Split.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Difference Between Map and FlatMap "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Split using Map transformation:\")\n",
    "# use the already created RDD (stringRDD) and split the string at every ' ' character\n",
    "# using the map transformation and the split function\n",
    "map_Split= stringRDD.map(lambda x: x.split(\" \"))\n",
    "# retrieve all the elements of the RDD/DataFrame/Dataset (from all nodes)\n",
    "map_Split.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "print(\"Split using FlatMap transformation:\")\n",
    "# the FlatMap tranformation on the RDD (stringRDD) is already defined (see two code cells above)\n",
    "# so we just have to retrieve all the elements of the RDD/DataFrame/Dataset (from all nodes)\n",
    "flatMap_Split.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since the source RDD contains two strings, the map transformation returns two separate objects (each with separate strings). The flatMap transformation returns only one object with all separated strings from both input objects (strings)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Filter Transformation\n",
    "\n",
    "*Return a new RDD containing only the elements that satisfy a predicate*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# filter all objects from RDD containing 'awesome'\n",
    "awesomeLineRDD = stringRDD.filter(lambda x: \"awesome\" in x)\n",
    "awesomeLineRDD.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# filter all objects from RDD containing 'spark'\n",
    "sparkLineRDD = stringRDD.filter(lambda x: \"spark\" in x.lower())\n",
    "sparkLineRDD.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Union Transformation\n",
    "\n",
    "*Return a new RDD containing all items from two original RDDs. Duplicates are not culled.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create two new RDDs\n",
    "rdd1 = sc.parallelize([1,2,3,4,5])\n",
    "rdd2 = sc.parallelize([1,6,7,8])\n",
    "# create a third RDD with 'union' transformation on RDD1 and RDD2\n",
    "rdd3 = rdd1.union(rdd2)\n",
    "rdd3.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Intersection Transformation\n",
    "\n",
    "*Return the intersection of this RDD and another one. The output will not contain any duplicate elements, even if the input RDDs did.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create two new RDDs\n",
    "rdd1 = sc.parallelize([\"One\", \"Two\", \"Three\"])\n",
    "rdd2 = sc.parallelize([\"two\",\"One\",\"threed\",\"One\"])\n",
    "# create a third RDD with 'intersection' transformation on RDD1 and RDD2\n",
    "rdd3 = rdd1.intersection(rdd2)\n",
    "rdd3.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Substract Trsnformation\n",
    "\n",
    "*Return each value in `self` that is not contained in `other`.* </br>\n",
    "(return a new DataFrame containing rows in this DataFrame but not in another DataFrame.)</br>\n",
    "This is equivalent to EXCEPT DISTINCT in SQL."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a new RDD 'words', use transformation flatMap and map ... all in one line\n",
    "words = sc.parallelize([\"The amazing thing about spark \\\n",
    "                is that it is very simple to learn\"]).flatMap(lambda x: x.split(\" \")).map(lambda c: c.lower())\n",
    "\n",
    "# create a new TDD 'stopWords'\n",
    "stopWords = sc.parallelize([\"the\", \"it\", \"is\", \"to\", \"that\", ''])\n",
    "\n",
    "# use substract transformation on words RDD. \n",
    "realWords = words.subtract(stopWords)\n",
    "realWords.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Distinct Transformation\n",
    "\n",
    "*Return a new RDD containing distinct items from the original RDD (omitting all duplicates)*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new RDD 'duplicateValueRDD'\n",
    "duplicateValueRDD = sc.parallelize([\"one\", 1,\"two\", 2, \"three\", \"one\", \"two\", 1, 2])\n",
    "# use distinct transformation on RDD and collect action - in one line\n",
    "duplicateValueRDD.distinct().collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###  Sample Transformation\n",
    "\n",
    "*Return a new RDD containing a statistical sample of the original RDD*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a new RDD 'numbers'. The second parameter of the parallelize transformation is optional integer value\n",
    "# and defines the number of partitions the data would be parallelized to.\n",
    "numbers = sc.parallelize([1,2,3,4,5,6,7,8,9,10], 2)\n",
    "\n",
    "# The transformation 'sample' returns a sampled subset of the numbers RDD.\n",
    "# The first parameter (here True) defines 'withReplacement'. The same element can be produced more than \n",
    "# once as the result of sample.\n",
    "# The second parameter (here 0.3) defines the fraction of rows to generate. Note that it doesn’t guarantee \n",
    "# to provide the exact number of the fraction of records.\n",
    "numbers.sample(True, 0.3).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### GroupBy Transformation\n",
    "\n",
    "*Group the data in the original RDD. Create pairs where the key is the output of a user function, and the value is all items for which the function yields this key.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a new RDD 'x'\n",
    "x = sc.parallelize(['John', 'Fred', 'Anna', 'James'])\n",
    "# groupBy all elements by the first letter of each element (which will be used as keys)\n",
    "y = x.groupBy(lambda w: w[0])\n",
    "# 'loop' through all element of the 'y' RDD and print the objects\n",
    "print([(k, list(v)) for (k, v) in y.collect()])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## GroupByKey Transformation\n",
    "\n",
    "*Group the values for each key in the original RDD. Create a new pair where the original key corresponds to this collected group of values.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new 'x' RDD with key,value pairs\n",
    "x = sc.parallelize([('B',5),('B',4),('A',3),('A',2),('A',1)])\n",
    "# create RDD 'y' using the groupBy transformation on the keys of RDD 'x'\n",
    "y = x.groupByKey()\n",
    "# print objects of RDD 'x'\n",
    "print(x.collect())\n",
    "# 'loop' through all element of the 'y' RDD and print the objects\n",
    "print(list((j[0], list(j[1])) for j in y.collect()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## MapPartitions Transformation\n",
    "\n",
    "*Return a new RDD by applying a function to each partition of this RDD*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new RDD with two partitions\n",
    "x = sc.parallelize([1,2,3], 2)\n",
    "# define the function 'f' - it is an iterable. The function sums all values of a partition\n",
    "# and returns the sum and the number 42 as an object\n",
    "def f(iterator): yield sum(iterator); yield 42\n",
    "# use the transformation 'mapPartitions' with the function 'f'\n",
    "y = x.mapPartitions(f)\n",
    "# glom() flattens elements on the same partition\n",
    "print(x.glom().collect())\n",
    "print(y.glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### MapPartitionWithIndex Transformation\n",
    "\n",
    "*Return a new RDD by applying a function to each partition of this RDD, while tracking the index of the original partition.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new RDD with two partitions\n",
    "x = sc.parallelize([1,2,3], 2)\n",
    "# define the function 'f' - it is an iterable. The function sums all values of a partition\n",
    "# and returns the index of the origina partition and the sum as an object\n",
    "def f(partitionIndex, iterator): yield (partitionIndex, sum(iterator))\n",
    "# use the transformation 'mapPartitionsWithIndex' with the function 'f'\n",
    "y = x.mapPartitionsWithIndex(f)\n",
    "# glom() flattens elements on the same partition\n",
    "print(x.glom().collect())\n",
    "print(y.glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Join Transformation\n",
    "\n",
    "*Return a new RDD containing all pairs of elements having the same key in the original RDDs*\n",
    "\n",
    "`union(otherRDD, numPartitions=None)`\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create RDD 'x'\n",
    "x = sc.parallelize([(\"a\", 1), (\"b\", 2)])\n",
    "# create RDD 'y'\n",
    "y = sc.parallelize([(\"a\", 3), (\"a\", 4), (\"b\", 5)])\n",
    "# create RDD 'x' as a join result on keys from RDD 'x' and 'y'\n",
    "z = x.join(y)\n",
    "print(z.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Coalesce Transformation\n",
    "\n",
    "*Return a new RDD which is reduced to a smaller number of partitions*\n",
    "\n",
    "`coalesce(numPartitions, shuffle=False)`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a RDD with three partition\n",
    "x = sc.parallelize([1, 2, 3, 4, 5], 3)\n",
    "# reduce the number of partitions to two by using the coalesce transformation\n",
    "y = x.coalesce(2)\n",
    "print(x.glom().collect())\n",
    "print(y.glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### KeyBy Transformation\n",
    "\n",
    "*Create a Pair RDD, forming one pair for each item in the original RDD. The pair’s key is calculated from the value via a user-supplied function.*\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a new RDD 'x'\n",
    "x = sc.parallelize(['John', 'Fred', 'Anna', 'James'])\n",
    "# use the first letter of each element as the key for the element\n",
    "y = x.keyBy(lambda w: w[0])\n",
    "print(y.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### PartitionBy Transformation\n",
    "\n",
    "*Return a new RDD with the specified number of partitions, placing original items into the partition returned by a user supplied function*\n",
    "\n",
    "`partitionBy(numPartitions, partitioner=portable_hash)`\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a RDD with three partition\n",
    "x = sc.parallelize([('J','James'),('F','Fred'),('A','Anna'),('J','John')], 3)\n",
    "# creta a new RDD 'y' with only two partitions and place each item in partition 0\n",
    "# if the first letter of the item is < 'H'. The item will be placed in partition 1 otherwise. \n",
    "y = x.partitionBy(2, lambda w: 0 if w[0] < 'H' else 1)\n",
    "# glom() flattens elements on the same partition\n",
    "print(x.glom().collect())\n",
    "print(y.glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Zip Transformation\n",
    "\n",
    "*Return a new RDD containing pairs whose key is the item in the original RDD, and whose\n",
    "value is that item’s corresponding element (same partition, same index) in a second RDD*\n",
    "\n",
    "`zip(otherRDD)`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create RDD 'x'\n",
    "x = sc.parallelize([1, 2, 3])\n",
    "# create RDD 'y' using the transformation map on RDD 'x'\n",
    "y = x.map(lambda n:n*n)\n",
    "# create RDD 'z' using the transformation zip on RDDs 'x' and 'y'\n",
    "z = x.zip(y)\n",
    "print(x.collect())\n",
    "print(y.collect())\n",
    "print(z.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Actions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<table class=\"table\">\n",
    "<tbody><tr><th>Action</th><th>Meaning</th></tr>\n",
    "<tr>\n",
    "  <td> <b>reduce</b>(<i>func</i>) </td>\n",
    "  <td> Aggregate the elements of the dataset using a function <i>func</i> (which takes two arguments and returns one). The function should be commutative and associative so that it can be computed correctly in parallel. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>collect</b>() </td>\n",
    "  <td> Return all the elements of the dataset as an array at the driver program. This is usually useful after a filter or other operation that returns a sufficiently small subset of the data. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>count</b>() </td>\n",
    "  <td> Return the number of elements in the dataset. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>first</b>() </td>\n",
    "  <td> Return the first element of the dataset (similar to take(1)). </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>take</b>(<i>n</i>) </td>\n",
    "  <td> Return an array with the first <i>n</i> elements of the dataset. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>takeSample</b>(<i>withReplacement</i>, <i>num</i>, [<i>seed</i>]) </td>\n",
    "  <td> Return an array with a random sample of <i>num</i> elements of the dataset, with or without replacement, optionally pre-specifying a random number generator seed.</td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>takeOrdered</b>(<i>n</i>, <i>[ordering]</i>) </td>\n",
    "  <td> Return the first <i>n</i> elements of the RDD using either their natural order or a custom comparator. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>saveAsTextFile</b>(<i>path</i>) </td>\n",
    "  <td> Write the elements of the dataset as a text file (or set of text files) in a given directory in the local filesystem, HDFS or any other Hadoop-supported file system. Spark will call toString on each element to convert it to a line of text in the file. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>saveAsSequenceFile</b>(<i>path</i>) <br> (Java and Scala) </td>\n",
    "  <td> Write the elements of the dataset as a Hadoop SequenceFile in a given path in the local filesystem, HDFS or any other Hadoop-supported file system. This is available on RDDs of key-value pairs that implement Hadoop's Writable interface. In Scala, it is also\n",
    "   available on types that are implicitly convertible to Writable (Spark includes conversions for basic types like Int, Double, String, etc). </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>saveAsObjectFile</b>(<i>path</i>) <br> (Java and Scala) </td>\n",
    "  <td> Write the elements of the dataset in a simple format using Java serialization, which can then be loaded using\n",
    "    <code>SparkContext.objectFile()</code>. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>countByKey</b>() <a name=\"CountByLink\"></a> </td>\n",
    "  <td> Only available on RDDs of type (K, V). Returns a hashmap of (K, Int) pairs with the count of each key. </td>\n",
    "</tr>\n",
    "<tr>\n",
    "  <td> <b>foreach</b>(<i>func</i>) </td>\n",
    "  <td> Run a function <i>func</i> on each element of the dataset. This is usually done for side effects such as updating an Accumulator or interacting with external storage systems.\n",
    "  <br><b>Note</b>: modifying variables other than Accumulators outside of the <code>foreach()</code> may result in undefined behavior. See Understanding closures for more details.</td>\n",
    "</tr>\n",
    "</tbody></table>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### GetNumpartitions Action\n",
    "\n",
    "*Return the number of partitions in RDD*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create RDD 'x' with two partitions\n",
    "x = sc.parallelize([1,2,3], 2)\n",
    "# get the number of partitions of RDD 'x' - the return value is from type integer\n",
    "y = x.getNumPartitions()\n",
    "# glom() flattens elements on the same partition\n",
    "print(x.glom().collect())\n",
    "print(y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Collect Action\n",
    "\n",
    "*Return all items in the RDD to the driver in a single list*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create RDD 'x' with two partitions\n",
    "x = sc.parallelize([1,2,3], 2)\n",
    "# create list 'y' (no RDD - y is from type 'list')\n",
    "y = x.collect()\n",
    "print(x.glom().collect())\n",
    "print(y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Count Action\n",
    "\n",
    "*Return the number of elements in this RDD.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new RDD 'numberRDD' with two partitions\n",
    "numberRDD = sc.parallelize([1,2,3,4,5,6,7,8,9,10], 2)\n",
    "# the action count returns the number of element in the dataset - independent of the number of partitions \n",
    "numberRDD.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### First Action\n",
    "\n",
    "*Return the first element in this RDD.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new RDD 'numberRDD' with two partitions\n",
    "numberRDD = sc.parallelize([1,2,3,4,5,6,7,8,9,10], 2)\n",
    "# return the first element - the order of the elements within the RDD is not effected by the partitioning\n",
    "numberRDD.first()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Take Action\n",
    "\n",
    "*Take the first num elements of the RDD.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new RDD 'numberRDD' with two partitions\n",
    "numberRDD = sc.parallelize([1,2,3,4,5,6,7,8,9,10], 2)\n",
    "# return the first FOUR element - the order of the elements within the RDD is not effected by the partitioning\n",
    "numberRDD.take(4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Reduce Action\n",
    "\n",
    "*Aggregate all the elements of the RDD by applying a user function pairwise to elements and partial results, and returns a result to the driver*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new RDD 'x'\n",
    "x = sc.parallelize([1,2,3,4])\n",
    "# apply function pairwise (a,b) to elements and return the sum - return type is integer\n",
    "y = x.reduce(lambda a,b: a+b)\n",
    "print(x.collect())\n",
    "print(y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregate Action"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since RDD’s are partitioned, the aggregate takes full advantage of it by first aggregating elements in each partition and then aggregating results of all partition to get the final result.\n",
    "\n",
    "Aggregate all the elements of the RDD by:\n",
    "- applying a user function to combine elements with user-supplied objects,\n",
    "- then combining those user-defined results via a second user function,\n",
    "- and finally returning a result to the driver."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The seqOp operator is used to accumulate the results of each partition and stores the running \n",
    "# accumulated result to data.\n",
    "seqOp = lambda data, item: (data[0] + [item], data[1] + item)\n",
    "# The combOp is used to combine the results of all partitions\n",
    "combOp = lambda d1, d2: (d1[0] + d2[0], d1[1] + d2[1])\n",
    "# create new RDD 'x'\n",
    "x = sc.parallelize([1,2,3,4])\n",
    "# aggregate all elements of the RDD\n",
    "y = x.aggregate(([], 0), seqOp, combOp)\n",
    "print(y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Max Action\n",
    "\n",
    "*Return the maximum item in the RDD*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create new RDD 'x'\n",
    "x = sc.parallelize([2,4,1])\n",
    "# return the maximum value from the dataset\n",
    "y = x.max()\n",
    "print(x.collect())\n",
    "print(y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stop The Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# stop the underlying SparkContext.\n",
    "try:\n",
    "    sc\n",
    "except NameError:\n",
    "    print(\"Spark context does not context exist - nothing to stop.\")\n",
    "else:\n",
    "    sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "*This Notebook was an introduction to the Apache Spark concept of RDDs (Resilient distributed\n",
    "datasets) and how to use transformations and actions on RDDs.*\n",
    "\n",
    "**Next UP: [Data Wrangling using RDDs](./03_Data_Wrangling_using_RDDs.ipynb)**"
   ]
  }
 ],
 "metadata": {
  "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.9.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}