{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RDD Transformations and Actions\n",
    "\n",
    "In this lecture we will begin to delve deeper into using Spark and Python. Please view the video lecture for a full explanation.\n",
    "\n",
    "## Important Terms\n",
    "\n",
    "Let's quickly go over some important terms:\n",
    "\n",
    "Term                   |Definition\n",
    "----                   |-------\n",
    "RDD                    |Resilient Distributed Dataset\n",
    "Transformation         |Spark operation that produces an RDD\n",
    "Action                 |Spark operation that produces a local object\n",
    "Spark Job              |Sequence of transformations on data with a final action"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating an RDD\n",
    "\n",
    "There are two common ways to create an RDD:\n",
    "\n",
    "Method                      |Result\n",
    "----------                               |-------\n",
    "`sc.parallelize(array)`                  |Create RDD of elements of array (or list)\n",
    "`sc.textFile(path/to/file)`                      |Create RDD of lines from file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RDD Transformations\n",
    "\n",
    "We can use transformations to create a set of instructions we want to preform on the RDD (before we call an action and actually execute them).\n",
    "\n",
    "Transformation Example                          |Result\n",
    "----------                               |-------\n",
    "`filter(lambda x: x % 2 == 0)`           |Discard non-even elements\n",
    "`map(lambda x: x * 2)`                   |Multiply each RDD element by `2`\n",
    "`map(lambda x: x.split())`               |Split each string into words\n",
    "`flatMap(lambda x: x.split())`           |Split each string into words and flatten sequence\n",
    "`sample(withReplacement=True,0.25)`      |Create sample of 25% of elements with replacement\n",
    "`union(rdd)`                             |Append `rdd` to existing RDD\n",
    "`distinct()`                             |Remove duplicates in RDD\n",
    "`sortBy(lambda x: x, ascending=False)`   |Sort elements in descending order"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RDD Actions\n",
    "\n",
    "Once you have your 'recipe' of transformations ready, what you will do next is execute them by calling an action. Here are some common actions:\n",
    "\n",
    "Action                             |Result\n",
    "----------                             |-------\n",
    "`collect()`                            |Convert RDD to in-memory list \n",
    "`take(3)`                              |First 3 elements of RDD \n",
    "`top(3)`                               |Top 3 elements of RDD\n",
    "`takeSample(withReplacement=True,3)`   |Create sample of 3 elements with replacement\n",
    "`sum()`                                |Find element sum (assumes numeric elements)\n",
    "`mean()`                               |Find element mean (assumes numeric elements)\n",
    "`stdev()`                              |Find element deviation (assumes numeric elements)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "____\n",
    "# Examples\n",
    "\n",
    "Now the best way to show all of this is by going through examples! We'll first review a bit by creating and working with a simple text file, then we will move on to more realistic data, such as customers and sales data.\n",
    "\n",
    "### Creating an RDD from a text file:\n",
    "\n",
    "** Creating the textfile **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing example2.txt\n"
     ]
    }
   ],
   "source": [
    "%%writefile example2.txt\n",
    "first \n",
    "second line\n",
    "the third line\n",
    "then a fourth line"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's perform some transformations and actions on this text file:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc = SparkContext()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[1] at textFile at NativeMethodAccessorImpl.java:-2"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Show RDD\n",
    "sc.textFile('example2.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Save a reference to this RDD\n",
    "text_rdd = sc.textFile('example2.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['first'],\n",
       " ['second', 'line'],\n",
       " ['the', 'third', 'line'],\n",
       " ['then', 'a', 'fourth', 'line']]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Map a function (or lambda expression) to each line\n",
    "# Then collect the results.\n",
    "text_rdd.map(lambda line: line.split()).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Map vs flatMap"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['first',\n",
       " 'second',\n",
       " 'line',\n",
       " 'the',\n",
       " 'third',\n",
       " 'line',\n",
       " 'then',\n",
       " 'a',\n",
       " 'fourth',\n",
       " 'line']"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Collect everything as a single flat map\n",
    "text_rdd.flatMap(lambda line: line.split()).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RDDs and Key Value Pairs\n",
    "\n",
    "Now that we've worked with RDDs and how to aggregate values with them, we can begin to look into working with Key Value Pairs. In order to do this, let's create some fake data as a new text file.\n",
    "\n",
    "This data represents some services sold to customers for some SAAS business."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing services.txt\n"
     ]
    }
   ],
   "source": [
    "%%writefile services.txt\n",
    "#EventId    Timestamp    Customer   State    ServiceID    Amount\n",
    "201       10/13/2017      100       NY       131          100.00\n",
    "204       10/18/2017      700       TX       129          450.00\n",
    "202       10/15/2017      203       CA       121          200.00\n",
    "206       10/19/2017      202       CA       131          500.00\n",
    "203       10/17/2017      101       NY       173          750.00\n",
    "205       10/19/2017      202       TX       121          200.00"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "services = sc.textFile('services.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['#EventId    Timestamp    Customer   State    ServiceID    Amount',\n",
       " '201       10/13/2017      100       NY       131          100.00']"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "services.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[10] at RDD at PythonRDD.scala:43"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "services.map(lambda x: x.split())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['#EventId', 'Timestamp', 'Customer', 'State', 'ServiceID', 'Amount'],\n",
       " ['201', '10/13/2017', '100', 'NY', '131', '100.00'],\n",
       " ['204', '10/18/2017', '700', 'TX', '129', '450.00']]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "services.map(lambda x: x.split()).take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's remove that first hash-tag!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['EventId    Timestamp    Customer   State    ServiceID    Amount',\n",
       " '201       10/13/2017      100       NY       131          100.00',\n",
       " '204       10/18/2017      700       TX       129          450.00',\n",
       " '202       10/15/2017      203       CA       121          200.00',\n",
       " '206       10/19/2017      202       CA       131          500.00',\n",
       " '203       10/17/2017      101       NY       173          750.00',\n",
       " '205       10/19/2017      202       TX       121          200.00']"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "services.map(lambda x: x[1:] if x[0]=='#' else x).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['EventId', 'Timestamp', 'Customer', 'State', 'ServiceID', 'Amount'],\n",
       " ['201', '10/13/2017', '100', 'NY', '131', '100.00'],\n",
       " ['204', '10/18/2017', '700', 'TX', '129', '450.00'],\n",
       " ['202', '10/15/2017', '203', 'CA', '121', '200.00'],\n",
       " ['206', '10/19/2017', '202', 'CA', '131', '500.00'],\n",
       " ['203', '10/17/2017', '101', 'NY', '173', '750.00'],\n",
       " ['205', '10/19/2017', '202', 'TX', '121', '200.00']]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "services.map(lambda x: x[1:] if x[0]=='#' else x).map(lambda x: x.split()).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using Key Value Pairs for Operations\n",
    "\n",
    "Let us now begin to use methods that combine lambda expressions that use a ByKey argument. These ByKey methods will assume that your data is in a Key,Value form. \n",
    "\n",
    "\n",
    "For example let's find out the total sales per state: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# From Previous\n",
    "cleanServ = services.map(lambda x: x[1:] if x[0]=='#' else x).map(lambda x: x.split())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['EventId', 'Timestamp', 'Customer', 'State', 'ServiceID', 'Amount'],\n",
       " ['201', '10/13/2017', '100', 'NY', '131', '100.00'],\n",
       " ['204', '10/18/2017', '700', 'TX', '129', '450.00'],\n",
       " ['202', '10/15/2017', '203', 'CA', '121', '200.00'],\n",
       " ['206', '10/19/2017', '202', 'CA', '131', '500.00'],\n",
       " ['203', '10/17/2017', '101', 'NY', '173', '750.00'],\n",
       " ['205', '10/19/2017', '202', 'TX', '121', '200.00']]"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cleanServ.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('State', 'Amount'),\n",
       " ('NY', '100.00'),\n",
       " ('TX', '450.00'),\n",
       " ('CA', '200.00'),\n",
       " ('CA', '500.00'),\n",
       " ('NY', '750.00'),\n",
       " ('TX', '200.00')]"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Let's start by practicing grabbing fields\n",
    "cleanServ.map(lambda lst: (lst[3],lst[-1])).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('State', 'Amount'),\n",
       " ('NY', '100.00750.00'),\n",
       " ('TX', '450.00200.00'),\n",
       " ('CA', '200.00500.00')]"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Continue with reduceByKey\n",
    "# Notice how it assumes that the first item is the key!\n",
    "cleanServ.map(lambda lst: (lst[3],lst[-1]))\\\n",
    "         .reduceByKey(lambda amt1,amt2 : amt1+amt2)\\\n",
    "         .collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Uh oh! Looks like we forgot that the amounts are still strings! Let's fix that:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('State', 'Amount'), ('NY', 850.0), ('TX', 650.0), ('CA', 700.0)]"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Continue with reduceByKey\n",
    "# Notice how it assumes that the first item is the key!\n",
    "cleanServ.map(lambda lst: (lst[3],lst[-1]))\\\n",
    "         .reduceByKey(lambda amt1,amt2 : float(amt1)+float(amt2))\\\n",
    "         .collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can continue our analysis by sorting this output:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('NY', 850.0), ('CA', 700.0), ('TX', 650.0)]"
      ]
     },
     "execution_count": 69,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Grab state and amounts\n",
    "# Add them\n",
    "# Get rid of ('State','Amount')\n",
    "# Sort them by the amount value\n",
    "cleanServ.map(lambda lst: (lst[3],lst[-1]))\\\n",
    ".reduceByKey(lambda amt1,amt2 : float(amt1)+float(amt2))\\\n",
    ".filter(lambda x: not x[0]=='State')\\\n",
    ".sortBy(lambda stateAmount: stateAmount[1], ascending=False)\\\n",
    ".collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** Remember to try to use unpacking for readability. For example: **"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "x = ['ID','State','Amount']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def func1(lst):\n",
    "    return lst[-1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def func2(id_st_amt):\n",
    "    # Unpack Values\n",
    "    (Id,st,amt) = id_st_amt\n",
    "    return amt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Amount'"
      ]
     },
     "execution_count": 84,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "func1(x)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Amount'"
      ]
     },
     "execution_count": 85,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "func2(x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Great Job!"
   ]
  }
 ],
 "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.5.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}