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": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = SparkConf().setMaster(\"local\").setAppName(\"Tutorial\")\n",
    "sc = SparkContext(conf = conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.43.84:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.0.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Tutorial</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local appName=Tutorial>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "stringList = [\"Spark is awesome\",\"Spark is cool\"]\n",
    "stringRDD = sc.parallelize(stringList)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[0] at readRDDFromFile at PythonRDD.scala:262"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Spark is awesome', 'Spark is cool']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings = sc.textFile(\"data/ml-1m/ratings.dat\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1::1193::5::978300760',\n",
       " '1::661::3::978302109',\n",
       " '1::914::3::978301968',\n",
       " '1::3408::4::978300275',\n",
       " '1::2355::5::978824291']"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1::1193::5::978300760',\n",
       " '1::661::3::978302109',\n",
       " '1::914::3::978301968',\n",
       " '1::3408::4::978300275',\n",
       " '1::2355::5::978824291']"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['SPARK IS AWESOME', 'SPARK IS COOL']"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "stringRDD_uppercase= stringRDD.map(lambda x: x.upper())\n",
    "stringRDD_uppercase.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['SpArK Is aWeSoMe', 'SpArK Is cOoL']"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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",
    "stringRDD_alternate_uppercase= stringRDD.map(alternate_char_upper)\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": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Spark', 'is', 'awesome', 'Spark', 'is', 'cool']"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "flatMap_Split= stringRDD.flatMap(lambda x: x.split(\" \"))\n",
    "flatMap_Split.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Difference Between Map and FlatMap "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Split using Map transformation:\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[['Spark', 'is', 'awesome'], ['Spark', 'is', 'cool']]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "print(\"Split using Map transformation:\")\n",
    "map_Split= stringRDD.map(lambda x: x.split(\" \"))\n",
    "map_Split.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Split using FlatMap transformation:\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "['Spark', 'is', 'awesome', 'Spark', 'is', 'cool']"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "print(\"Split using FlatMap transformation:\")\n",
    "flatMap_Split.collect()"
   ]
  },
  {
   "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": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Spark is awesome']"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "awesomeLineRDD = stringRDD.filter(lambda x: \"awesome\" in x)\n",
    "awesomeLineRDD.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Spark is awesome', 'Spark is cool']"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1, 2, 3, 4, 5, 1, 6, 7, 8]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd1 = sc.parallelize([1,2,3,4,5])\n",
    "rdd2 = sc.parallelize([1,6,7,8])\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": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['One']"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd1 = sc.parallelize([\"One\", \"Two\", \"Three\"])\n",
    "rdd2 = sc.parallelize([\"two\",\"One\",\"threed\",\"One\"])\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`.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['very', 'simple', 'amazing', 'thing', 'about', 'spark', 'learn']"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "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",
    "stopWords = sc.parallelize([\"the\", \"it\", \"is\", \"to\", \"that\", ''])\n",
    "\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": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['one', 1, 'two', 2, 'three']"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "duplicateValueRDD = sc.parallelize([\"one\", 1,\"two\", 2, \"three\", \"one\", \"two\", 1, 2])\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": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[6]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "numbers = sc.parallelize([1,2,3,4,5,6,7,8,9,10], 2)\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": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('J', ['John', 'James']), ('F', ['Fred']), ('A', ['Anna'])]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize(['John', 'Fred', 'Anna', 'James'])\n",
    "y = x.groupBy(lambda w: w[0])\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": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('B', 5), ('B', 4), ('A', 3), ('A', 2), ('A', 1)]\n",
      "[('B', [5, 4]), ('A', [3, 2, 1])]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([('B',5),('B',4),('A',3),('A',2),('A',1)])\n",
    "y = x.groupByKey()\n",
    "print(x.collect())\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": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[1], [2, 3]]\n",
      "[[1, 42], [5, 42]]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([1,2,3], 2)\n",
    "def f(iterator): yield sum(iterator); yield 42\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": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[1], [2, 3]]\n",
      "[[(0, 1)], [(1, 5)]]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([1,2,3], 2)\n",
    "def f(partitionIndex, iterator): yield (partitionIndex, sum(iterator))\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": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('b', (2, 5)), ('a', (1, 3)), ('a', (1, 4))]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([(\"a\", 1), (\"b\", 2)])\n",
    "y = sc.parallelize([(\"a\", 3), (\"a\", 4), (\"b\", 5)])\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": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[1], [2, 3], [4, 5]]\n",
      "[[1], [2, 3, 4, 5]]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([1, 2, 3, 4, 5], 3)\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": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('J', 'John'), ('F', 'Fred'), ('A', 'Anna'), ('J', 'James')]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize(['John', 'Fred', 'Anna', 'James'])\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": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[('J', 'James')], [('F', 'Fred')], [('A', 'Anna'), ('J', 'John')]]\n",
      "[[('F', 'Fred'), ('A', 'Anna')], [('J', 'James'), ('J', 'John')]]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([('J','James'),('F','Fred'),\n",
    "('A','Anna'),('J','John')], 3)\n",
    "y = x.partitionBy(2, lambda w: 0 if w[0] < 'H' else 1)\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": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(1, 1), (2, 4), (3, 9)]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([1, 2, 3])\n",
    "y = x.map(lambda n:n*n)\n",
    "z = x.zip(y)\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": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[1], [2, 3]]\n",
      "2\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([1,2,3], 2)\n",
    "y = x.getNumPartitions()\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": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[1], [2, 3]]\n",
      "[1, 2, 3]\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([1,2,3], 2)\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": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "numberRDD = sc.parallelize([1,2,3,4,5,6,7,8,9,10], 2)\n",
    "numberRDD.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### First Action\n",
    "\n",
    "*Return the first element in this RDD.*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "numberRDD = sc.parallelize([1,2,3,4,5,6,7,8,9,10], 2)\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": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1, 2, 3, 4]"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "numberRDD = sc.parallelize([1,2,3,4,5,6,7,8,9,10], 2)\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": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 2, 3, 4]\n",
      "10\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([1,2,3,4])\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": [
    "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": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "([1, 2, 3, 4], 10)\n"
     ]
    }
   ],
   "source": [
    "seqOp = lambda data, item: (data[0] + [item], data[1] + item)\n",
    "combOp = lambda d1, d2: (d1[0] + d2[0], d1[1] + d2[1])\n",
    "x = sc.parallelize([1,2,3,4])\n",
    "y = x.aggregate(([], 0), seqOp, combOp)\n",
    "print(y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Map Action\n",
    "\n",
    "*Return the maximum item in the RDD*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[2, 4, 1]\n",
      "4\n"
     ]
    }
   ],
   "source": [
    "x = sc.parallelize([2,4,1])\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": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  }
 ],
 "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.8.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}