PySpark Overview - Part 2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RDD manipulation with Key-Value Pairs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Group By Key\n",
    "\n",
    "This operation allows you to manipulate key-value pair RDDs according to their **key**.\n",
    "\n",
    "This operation places all of the values with the same key into an **iterable**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "kv_rdd = sc.parallelize([(\"Ibrahim\", 21), (\"Juan\", 32), (\"Andrew\", 200), (\"Ibrahim\", 12), (\"Ibrahim\", -45), (\"Ibrahim\", \"Chocolate\")], \n",
    "                        numSlices=6) # 6 paritions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Andrew', <pyspark.resultiterable.ResultIterable at 0x114506908>),\n",
       " ('Juan', <pyspark.resultiterable.ResultIterable at 0x114506048>),\n",
       " ('Ibrahim', <pyspark.resultiterable.ResultIterable at 0x114506978>)]"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kv_rdd.groupByKey().collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have generated a list with the keys as the first entry of a tuple, and an **iterator** object as the second entry.\n",
    "\n",
    "In order to view the values in the iterable, we do the following:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "This entry corresponds to Andrew. The contents of the iterable are: \n",
      "200\n",
      "This entry corresponds to Juan. The contents of the iterable are: \n",
      "32\n",
      "This entry corresponds to Ibrahim. The contents of the iterable are: \n",
      "21\n",
      "12\n",
      "-45\n",
      "Chocolate\n"
     ]
    }
   ],
   "source": [
    "results = kv_rdd.groupByKey().collect()\n",
    "\n",
    "for entry in results:\n",
    "    print(f\"This entry corresponds to {entry[0]}. The contents of the iterable are: \")\n",
    "    for contents in entry[1]:\n",
    "        print(contents)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is a great technique if you need to view what the results are by key, however, should you wish to perform calculations on the iterable, it would be much better to use `reduceByKey` or `aggregateByKey`.\n",
    "\n",
    "Also, in the above code snipped, I used a nested for-loop to show all results. In reality, you would maybe wish to inspect a partiular result. This can be achieved more efficiently through a list comprehension:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[21, 12, -45, 'Chocolate']"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "[x for x in results[-1][1]] #results for ibrahim"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`groupByKey` is using `reduce` behind the scenes - as such, with large RDD's you may see some performance lag. This is because spark is moving the data from the edxecutor that is reading the data to the executor that is collecting the values for a given key. Recall that we partitioned our data into 6.\n",
    "\n",
    "All of this data movement, by way of using `reduce` under the hood, leads to some performance lag. As mentioned in the previous notebook, using reduce is an expensive operation, as such, working to **reduce** the number of `reduce`'s will always lead to a performance improvement."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Reduce By Key\n",
    "\n",
    "An excellent operation for calculating aggregates over your entire RDD.\n",
    "\n",
    "Recall that if you just wanted to calculate a single aggregate metric, you could ust use the simple [`reduce`](https://spark.apache.org/docs/1.1.1/api/python/pyspark.rdd.RDD-class.html#reduce) command.\n",
    "\n",
    "However, it is more common, within the context of big data, to aggregate over a subset of your data - e.g. total users per IP address, total revenue per demographic. This is where `reduceByKey` shines!\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from random import randint, seed\n",
    "\n",
    "def generate_random_ips(amount):\n",
    "    lst = []\n",
    "    for _ in range(amount):\n",
    "        lst.append(\".\".join(str(randint(0, 255)) for _ in range(4)))\n",
    "    return lst"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['24.232.205.214',\n",
       " '84.155.214.192',\n",
       " '183.99.209.84',\n",
       " '3.46.212.177',\n",
       " '234.255.0.244']"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "seed(318)\n",
    "ips = generate_random_ips(5)\n",
    "ips"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('183.99.209.84', 1),\n",
       " ('24.232.205.214', 1),\n",
       " ('84.155.214.192', 1),\n",
       " ('3.46.212.177', 1),\n",
       " ('234.255.0.244', 1)]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from operator import add\n",
    "\n",
    "ip_rdd = sc.parallelize(ips)\n",
    "ip_rdd.map(lambda x:(x,1)).reduceByKey(add).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First and foremost, if you are not familiar with the [`operator`](https://docs.python.org/3.6/library/operator.html) library - I highly suggest you read the documentation. It is incredibly useful! Those of you with exposure to functional programming will particularly appreciate it.\n",
    "\n",
    "I first generated 5 random IP address and made them into an RDD. By using a `lambda` function, we changed each IP address to be a tuple containing the IP and the value 1 - (ip address, 1). We have now created a key value pair!\n",
    "\n",
    "As such, we can employ `reduceByKey`.\n",
    "\n",
    "`reduceByKey` requires a **function** to be passed as an argument. This is the function that will be used to do the aggregation. I passed the `add` function which will simply add up all the 1's in each tuple and gives us a result. Furthermore, the function that is passed must be [**associative**](https://en.wikipedia.org/wiki/Associative_property) and [**commutative**](https://en.wikipedia.org/wiki/Commutative_property). This is because the data is split up behind the scenes and sent to difference executors - there is no guarnatee about the order in which the data will be processed/returned.\n",
    "\n",
    "Note: If you want to just return a list of key-value pairs in a list, rather than doing an aggregation, it is better to use `groupByKey`!\n",
    "\n",
    "From the above, we see that there was only 1 of each IP address - that is to say, they are all unique. Below is an example with duplicate IP's.\n",
    "\n",
    "A more interesting use case comes from text analysis - after you tokenize words, you may which to calculate word counts in a corpus. `reduceByKey` is an excellent method for this task."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('183.99.209.84', 1), ('24.232.205.214', 3), ('234.255.0.244', 1)]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ips = ['24.232.205.214',\n",
    " '24.232.205.214',\n",
    " '183.99.209.84',\n",
    " '24.232.205.214',\n",
    " '234.255.0.244']\n",
    "\n",
    "ip_rdd = sc.parallelize(ips)\n",
    "ip_rdd.map(lambda x:(x,1)).reduceByKey(add).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Aggregate By Key\n",
    "\n",
    "This method is very similar to `reduceByKey`, however, it is **significantly** more flexible in terms of it's use case. This increased flexibility comes at a cost: increased complexity! In fact, the main difference between the two is taht `aggregatebykey` allows you to specify an initial value for aggregation.\n",
    "\n",
    "Again, for those of you with exposure to functional programming - specifically Haskell will breeze through this. You should think of `aggregateByKey` as a higher order function e.g. `foldl` or `foldr` only instead of operating on lists, you are operating on RDD's. However, take this description with a grain of salt as spark also has a `foldByKey` method which is closer to `foldl` and `foldr` \n",
    "\n",
    "`aggregateByKey` takes 3 parameters:\n",
    "\n",
    "- The Zero Value: This is the starting value that you wish to pass to your first function!\n",
    "\n",
    "- The sequencing Function: This is the fuction that is applied to each argument **per** partition.\n",
    "\n",
    "- The Combining Function: This is the function that enables you to combine the results of the sequences functions per partition together."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"Ibrahim\",1),(\"Juan\",2),(\"Ibrahim\",2),(\"Juan\",3)], numSlices=4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.getNumPartitions() # 4 partitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From the above, we can see that each tuple is it's own partition."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[('Ibrahim', 1)], [('Juan', 2)], [('Ibrahim', 2)], [('Juan', 3)]]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.glom().collect() # this is what our partitions look like in list form"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Ibrahim', 5), ('Juan', 7)]"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.aggregateByKey(1,lambda acc,val: acc+val, lambda acc1, acc2: acc1+acc2).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's walk through how we get the above result:\n",
    "\n",
    "The Zero Value (1) is applied to accumulator on **each** partition. Thus, each tuple, pn its own partition is passed to the sequence function.\n",
    "\n",
    "it is important to remember, since we are aggregating by key, the val referenced in the lambda functions refers to the numeric element in the tuples!\n",
    "\n",
    "lambda 1,1: 1+1 = 2. At this point, the accumulator value on the first partition is 2\n",
    "\n",
    "Value of acc on 1st parition: 2 -> lambda 1,1 = 1+1 = 2\n",
    "value of acc on 2nd partition: 3 -> lambda 1,2 = 1+2 = 3\n",
    "value of acc on 3rd partition: 3 - > lambda 1,2 = 1+2 = 3\n",
    "Value of acc on 4th partition: 4 -> lambda 1,3 = 1+3 = 4\n",
    "\n",
    "These values are now passed onto the combining function that combines the accumulator values on a **key** basis. Thus, for the \"Ibrahim\" key, we have lambda 2,3 = 2 + 3 = 5 (equivalent to adding the results of the 1st and 3rd partition).\n",
    "\n",
    "For the \"Juan\" key, we have lambda 3,4 = 3 + 4 = 7\n",
    "\n",
    "It is important to realize here that since we supply an initial accumulator value, and since the accumulator value is applied to each partition in order to kickstart the sequencing function - that if we change the number of partitions, then we will get a different result!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd = sc.parallelize([(\"Ibrahim\",1),(\"Juan\",2),(\"Ibrahim\",2),(\"Juan\",3)], numSlices=1)\n",
    "rdd.getNumPartitions() # 4 partitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[('Ibrahim', 1), ('Juan', 2), ('Ibrahim', 2), ('Juan', 3)]]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.glom().collect() # this is what our partitions look like in list form"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Ibrahim', 4), ('Juan', 6)]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.aggregateByKey(1,lambda acc,val: acc+val, lambda acc1, acc2: acc1+acc2).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The reason we get the above result is now that the accumulator value is applied to a single partion and so get's passed on amongst the respective keys:\n",
    "\n",
    "\n",
    "Ibrahim Key:\n",
    "\n",
    "Sequencing Functions: \n",
    "\n",
    "lambda 1,1 = 1 + 1 = 2\n",
    "lambda 2, 2 = 2 + 2 = 4\n",
    "\n",
    "Combine function:\n",
    "lambda 4,0 = 4 + 0 = 4 -- notice how acc2 here is 0.\n",
    "\n",
    "Juan key:\n",
    "\n",
    "Sequencing Functions: \n",
    "\n",
    "lambda 1,2 = 1 + 2 = 3\n",
    "lambda 3,3 = 3 + 3 = 6\n",
    "\n",
    "Combine function:\n",
    "lambda 6,0 = 6 + 0 = 6 -- notice how acc2 here is 0.\n",
    "\n",
    "Since we have only 1 partition, there will be only one accumulator per key. As such, acc2 = 0\n",
    "\n",
    "You can find another great example of `aggregateByKey` [here](http://www.learnbymarketing.com/618/pyspark-rdd-basics-examples/)\n",
    "\n",
    "Here are some graphics that may help visualize waht is happening:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "4 paritions:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![4 paritions](imgs/aggbykey_1.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1 Partition:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![1 Partition](imgs/aggbykey_2.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Sort By Key\n",
    "\n",
    "Allows you to sort an RDD of key-value pairs by the key. It also allows you to set the sort order, number of partiions to output and a preprocessing function called a keyfunc.\n",
    "\n",
    "The pre-processing function will not change the the data that is present in the result, rather, the way the sortbykey functions \"views\" the data will be transformed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\",2), (\"d\",17), (\"A\",3), (\"C\",1)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('A', 3), ('C', 1), ('a', 2), ('d', 17)]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.sortByKey().collect() # default sorting"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('d', 17), ('a', 2), ('C', 1), ('A', 3)]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.sortByKey(ascending=False).collect() # reverse order"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[('A', 3), ('C', 1)], [('a', 2)], [('d', 17)]]"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.sortByKey(numPartitions=3).glom().collect() # 3 paritions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[('A', 3), ('C', 1), ('a', 2), ('d', 17)]]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.sortByKey(numPartitions=1).glom().collect() # 1 paritions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, let's say we wantd to normalize our sorting - that is to say, ignore the fact that we have capital letters with regards to our sorting. This is where they keyfunc comes into play."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('a', 2), ('A', 3), ('C', 1), ('d', 17)]"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd.sortByKey(keyfunc=lambda x: x.lower()).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Each key was passed through the lambda function and then sorted on the basis of that results. As such, we can see from the above that we have a normalized sorting of our key-value pairs."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Join\n",
    "\n",
    "This function will allow you to combine RDD's. This works by taking 2 key-value RDD's and combining all values across the RDD's by key!\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd1 = sc.parallelize([(1,\"a\"), (2,\"a\")])\n",
    "rdd2 = sc.parallelize([(2,\"b\"), (3,\"b\")])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(2, ('a', 'b'))]"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd1.join(rdd2).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Recall that join only works by operating on keys - thus we only get output where keys have matched. However, what if we had the same key present in one RDD and joined to another RDD?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd3 = sc.parallelize([(2,\"b\"), (3,\"b\"), (2,\"c\")])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(2, ('a', 'b')), (2, ('a', 'c'))]"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd1.join(rdd3).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Interesting, we now get pairs of every combination from rdd1 and rdd3. When having identical keys, you may wish to keep them all in the same RDD, or you could use the CoGroup function to bring it all together in one result.\n",
    "\n",
    "Also, notice how you are **losing** data, that is, any key that was only present in 1 RDD is lost.\n",
    "\n",
    "If you wish to retain data, then you can use the following:\n",
    "\n",
    "- Left Outer Join: Ensures all data from the original RDD makes it to the end result, if the second RDD doesnt have a matching key, then spark will insert a none value.\n",
    "\n",
    "- Right Outer Join: Same as above, but ensures data from the second RDD makes it to the end result.\n",
    "\n",
    "- Full Outer Join: Preserves all values in both RDD's.\n",
    "\n",
    "It is important to mention here that when using these joins, you must be prepared to deal with the none values. Furthemore, all of these join functions allow you to specify the number of paritions. As such, when using joins, you could potentially be moving a large amount of data in the cluster. This is almost always a performance bottleneck.\n",
    "\n",
    "You should think carefully about if you really need a join or whether you can filter your data before joining."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, ('a', None)), (2, ('a', 'b'))]"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd1.leftOuterJoin(rdd2).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### CoGroup\n",
    "\n",
    "Also allows you to combine two RDD's. But is slightly different from Join!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "rdd1 = sc.parallelize([(1,\"a\"), (2,\"a\")])\n",
    "rdd2 = sc.parallelize([(2,\"b\"), (3,\"d\"), (2,\"c\")])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(2, ('a', 'b')), (2, ('a', 'c'))]"
      ]
     },
     "execution_count": 45,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd1.join(rdd2).collect() # just as we saw before"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1,\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x10d956a90>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x10d956b38>)),\n",
       " (2,\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x10d956f60>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x10d956e48>)),\n",
       " (3,\n",
       "  (<pyspark.resultiterable.ResultIterable at 0x10d956b00>,\n",
       "   <pyspark.resultiterable.ResultIterable at 0x10d956dd8>))]"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd1.cogroup(rdd2).collect() # we get result iterables."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's see what is inside those iterables by casting them to lists!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, [['a'], []]), (2, [['a'], ['b', 'c']]), (3, [[], ['d']])]"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd1.cogroup(rdd2).mapValues(lambda x: [list(x[0]), list(x[1])]).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From the above, we can see that co-group makes a **group** for each key! It essentially provides a list for values that were seen for a particular key on both of the RDD's"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Input and Output\n",
    "\n",
    "#### Whole Text Files.\n",
    "\n",
    "This functions allows you to read in an entire directly or text files and returns them as pairs.\n",
    "\n",
    "Each pair is of the form: (filename, content)\n",
    "\n",
    "This function is similar to the previous function we used: `textFile`. However, `textFile` returns an RDD with an element for each line in the file.\n",
    "\n",
    "Whole text file returns an RDD, however, 1 elements in a 1 **entire** file of the form (filename, content)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['This is some test data.',\n",
       " 'That takes more than one line.',\n",
       " 'This is a second test file.',\n",
       " 'Is it also',\n",
       " 'a multiline file.']"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.textFile(\"samples\").collect() #samples is a folder"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So the above loaded every file in the sample folder and get each line in each file an element position on the RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('file:/Users/ibrahimgabr/spv/samples/sample_1.txt',\n",
       "  'This is some test data.\\nThat takes more than one line.\\n'),\n",
       " ('file:/Users/ibrahimgabr/spv/samples/sample_2.txt',\n",
       "  'This is a second test file.\\nIs it also\\na multiline file.\\n')]"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.wholeTextFiles(\"samples\").collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "So when do you use which? It depends on what you are trying to do!\n",
    "\n",
    "Use wholeTextFiles if you are, for example, generating total word counts per file. In this case, it would be impoortant to keep each file seperate!\n",
    "\n",
    "However, if you were generating a term frequency table, that needs tp take into account the entire corpus of text, then textFile is the way to go. \n",
    "\n",
    "In essence, you need to decide if the level of fidelity should be every line in a file, or simply the file itself.\n",
    "\n",
    "Also, recall that **parallelism can never be higher than the number of elements in your RDD**. As such, if you used wholetext files to load up two 5Gb textfiles, then even if you were on a 100 node cluster, only 2 nodes would be busy working while the other 98 would sit idle. As such, whenever you can, using textFile is the way to go as it allows you to exploit the power of parallelism."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Pickle Files\n",
    "\n",
    "This is a way to store and load up python objects in spark.\n",
    "\n",
    "Let's say you have finished training an ML model, you could pickle it and then load it back up later.\n",
    "\n",
    "While I dont have a model on hand, let's run through a simple exmaple of making a pickle file and loading a pickle file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.parallelize([\"ML Model\", \"Logisitic Regression\", {\"alpha\": 0.05}]).saveAsPickleFile(\"model\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['ML Model', 'Logisitic Regression', {'alpha': 0.05}]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.pickleFile(\"model\").collect() # reloaded the pickle file returned as RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Conclusion\n",
    "\n",
    "That's all I have for today. Plenty of information for you to play with and get your hands dirty!! I really hope you enjoy the notebook!\n",
    "\n",
    "The next notebook will cover topics about performance boosting, cluster configuration in addition to exposing you to some advanced spark features like MLib and Spark Streaming.\n",
    "\n",
    "As always, feel free to reach out: igabr@uchicago.edu or [@Gabr\\_Ibrahim](https://twitter.com/Gabr_Ibrahim)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "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.6.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}