PySpark_safaribooksonline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Resilient Distributed Dataset - Transformations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:175"
      ]
     },
     "execution_count": 88,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import numpy as np\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark import SparkContext\n",
    "sc =SparkContext()\n",
    "\n",
    "random_numbers = sc.parallelize([np.random.rand() for _ in range(1000)], 4)\n",
    "random_numbers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:175"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "display(random_numbers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0.3908827457638425, 0.9706352129593969, 0.28054105344126723]"
      ]
     },
     "execution_count": 90,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "random_numbers.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd_random_tuples = sc.parallelize([(\"Mark\", 43, \"6'1\\\"\"), (\"Stella\", 23, \"5'6\\\"\"), (\"Skye\", 6, \"3'11\\\"\"), (\"Albert\", 1, \"2'7\\\"\")])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Mark', 43, '6\\'1\"'), ('Stella', 23, '5\\'6\"'), ('Skye', 6, '3\\'11\"')]"
      ]
     },
     "execution_count": 92,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_random_tuples.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['OrderDate,Region,Rep,Item,Units,Unit Cost,Total',\n",
       " '1-6-16,East,Jones,Pencil,95, 1.99 , 189.05 ',\n",
       " '1-23-16,Central,Kivell,Binder,50, 19.99 , 999.50 ']"
      ]
     },
     "execution_count": 93,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file = sc.textFile('sample_data.csv', 4)\n",
    "rdd_from_file.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['OrderDate,Region,Rep,Item,Units,Unit Cost,Total',\n",
       " '1-6-16,East,Jones,Pencil,95, 1.99 , 189.05 ',\n",
       " '1-23-16,Central,Kivell,Binder,50, 19.99 , 999.50 ',\n",
       " '2-9-16,Central,Jardine,Pencil,36, 4.99 , 179.64 ',\n",
       " '2-26-16,Central,Gill,Pen,27, 19.99 , 539.73 ']"
      ]
     },
     "execution_count": 94,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Structured"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Tom', 0), ('Spark', 1)]"
      ]
     },
     "execution_count": 95,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_valid_structured = sc.parallelize([('Tom', 0), ('Spark', 1)])\n",
    "rdd_valid_structured.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[9] at parallelize at PythonRDD.scala:175"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "display(rdd_valid_structured)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Unstructured"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Jul 17 22:04:25 router dnsprobe[276]: dnsquery failed',\n",
       " 'Jul 17 22:04:29 router last message repeated two times',\n",
       " 'Jul 17 22:04:29 router dnsprob[276]: Primary DNS server is Down... Switching to']"
      ]
     },
     "execution_count": 97,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_valid_unstructured = sc.parallelize([('Jul 17 22:04:25 router dnsprobe[276]: dnsquery failed'), \n",
    "                                         ('Jul 17 22:04:29 router last message repeated two times'), \n",
    "                                        ('Jul 17 22:04:29 router dnsprob[276]: Primary DNS server is Down... Switching to'), \n",
    "                                        ('Jul 17 22:05:08 router dnsprob[276]: Switching back to primary DNS server'), \n",
    "                                        ('Jul 17 22:26:11 debian -- MARK --'), \n",
    "                                        ('Jul 17 22:46:11 debian -- Mark --'), \n",
    "                                        ('Jul 17 22:47:36 router -- MARK --'), \n",
    "                                        ('Jul 17 22:47:36 router dnsprob[276]: dns query failed')])\n",
    "rdd_valid_unstructured.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Heterogeneous Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Toll invoice number', 'TRE2321'),\n",
       " {'Last bill balance': 0.0, 'New bill balance:': 23.92},\n",
       " ['2017-11-12 12:32:34PM', '2017-11-14 1:32:56PM']]"
      ]
     },
     "execution_count": 98,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_valid_heterogeneous = sc.parallelize([('Toll invoice number', 'TRE2321'), \n",
    "                                         {'Last bill balance': 0.00, 'New bill balance:': 23.92}, \n",
    "                                         ['2017-11-12 12:32:34PM', '2017-11-14 1:32:56PM']])\n",
    "rdd_valid_heterogeneous.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Understanding lazy execution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Tom', 'Spark']"
      ]
     },
     "execution_count": 99,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "names_only = rdd_valid_structured.map(lambda element: element[0])\n",
    "names_only.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[18] at collect at <ipython-input-99-9e08de0ad59a>:2"
      ]
     },
     "execution_count": 100,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "names_only"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .map(...) transformation\n",
    "\n",
    "### Using lambdas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['OrderDate', 'Region', 'Rep', 'Item', 'Units', 'Unit Cost', 'Total']\n",
      "['1-6-16', 'East', 'Jones', 'Pencil', '95', ' 1.99 ', ' 189.05 ']\n"
     ]
    }
   ],
   "source": [
    "def splitOnComma(inputString):\n",
    "    return inputString.split(',')\n",
    "for el in rdd_from_file.map(splitOnComma).take(2):\n",
    "    print(el)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['OrderDate', 'Region', 'Rep', 'Item', 'Units', 'Unit Cost', 'Total']\n",
      "['1-6-16', 'East', 'Jones', 'Pencil', '95', ' 1.99 ', ' 189.05 ']\n"
     ]
    }
   ],
   "source": [
    "for el in rdd_from_file.map(lambda element: element.split(',')).take(2):\n",
    "    print(el)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transforming Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['OrderDate', 'Region', 'Rep', 'Item', -666, -666, -666],\n",
       " ['1-6-16', 'East', 'Jones', 'Pencil', 95.0, 1.99, 189.05],\n",
       " ['1-23-16', 'Central', 'Kivell', 'Binder', 50.0, 19.99, 999.5],\n",
       " ['2-9-16', 'Central', 'Jardine', 'Pencil', 36.0, 4.99, 179.64],\n",
       " ['2-26-16', 'Central', 'Gill', 'Pen', 27.0, 19.99, 539.73]]"
      ]
     },
     "execution_count": 103,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def converToFloat(inputString):\n",
    "    try:\n",
    "        return float(inputString)\n",
    "    except: \n",
    "        return -666\n",
    "\n",
    "to_filter = rdd_from_file.map(lambda element: element.split(',')).map(lambda element:\n",
    "                                                                     [e for e in element[:4]] + \n",
    "                                                                     [converToFloat(e) for e in element[4:]])\n",
    "to_filter.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .filter(...) transformation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['1-6-16', 'East', 'Jones', 'Pencil', 95.0, 1.99, 189.05],\n",
       " ['1-23-16', 'Central', 'Kivell', 'Binder', 50.0, 19.99, 999.5],\n",
       " ['2-9-16', 'Central', 'Jardine', 'Pencil', 36.0, 4.99, 179.64],\n",
       " ['2-26-16', 'Central', 'Gill', 'Pen', 27.0, 19.99, 539.73]]"
      ]
     },
     "execution_count": 104,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered = to_filter.filter(lambda element: element[4] != -666)\n",
    "filtered.take(4)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .flat(...) Map transformation\n",
    "\n",
    "#### .map to flatMap comparison"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[1, 30], [2, 40], [3, 50], [4, 60]]\n",
      "[1, 30, 2, 40, 3, 50, 4, 60]\n"
     ]
    }
   ],
   "source": [
    "def mapExample(inputList):\n",
    "    outputList = []\n",
    "    for item in inputList:\n",
    "        temp = item.copy()\n",
    "        temp[1] *= 10\n",
    "        outputList.append(temp)\n",
    "    return outputList\n",
    "\n",
    "def flatMapExample(inputList):\n",
    "    outputList = []\n",
    "    for item in inputList:\n",
    "        temp = item.copy()\n",
    "        temp[1] *= 10\n",
    "        outputList += temp\n",
    "    return outputList\n",
    "\n",
    "sampleList = [[1, 3], [2, 4], [3, 5], [4, 6]]\n",
    "print(mapExample(sampleList))\n",
    "print(flatMapExample(sampleList))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Filtering malformed record"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[],\n",
       " [[datetime.datetime(2016, 1, 6, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Pencil',\n",
       "   95,\n",
       "   1.99,\n",
       "   189.05]],\n",
       " [[datetime.datetime(2016, 1, 23, 0, 0),\n",
       "   'Central',\n",
       "   'Kivell',\n",
       "   'Binder',\n",
       "   50,\n",
       "   19.99,\n",
       "   999.5]]]"
      ]
     },
     "execution_count": 106,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import datetime as dt\n",
    "\n",
    "def parseCSVRow(inputRow):\n",
    "    try:\n",
    "        rowSplit = inputRow.split(',')\n",
    "        rowSplit[0] = dt.datetime.strptime(rowSplit[0], '%m-%d-%y')\n",
    "        rowSplit[4] = int(rowSplit[4])\n",
    "        \n",
    "        for i in [5,6]:\n",
    "            rowSplit[i] = float(rowSplit[i])\n",
    "            \n",
    "        return [rowSplit]\n",
    "    except:\n",
    "        return []\n",
    "rdd_from_file.map(parseCSVRow).take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[[datetime.datetime(2016, 1, 6, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Pencil',\n",
       "   95,\n",
       "   1.99,\n",
       "   189.05]],\n",
       " [[datetime.datetime(2016, 1, 23, 0, 0),\n",
       "   'Central',\n",
       "   'Kivell',\n",
       "   'Binder',\n",
       "   50,\n",
       "   19.99,\n",
       "   999.5]],\n",
       " [[datetime.datetime(2016, 2, 9, 0, 0),\n",
       "   'Central',\n",
       "   'Jardine',\n",
       "   'Pencil',\n",
       "   36,\n",
       "   4.99,\n",
       "   179.64]]]"
      ]
     },
     "execution_count": 107,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file.map(parseCSVRow).filter(lambda element: len(element)>0).take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[datetime.datetime(2016, 1, 6, 0, 0),\n",
       "  'East',\n",
       "  'Jones',\n",
       "  'Pencil',\n",
       "  95,\n",
       "  1.99,\n",
       "  189.05],\n",
       " [datetime.datetime(2016, 1, 23, 0, 0),\n",
       "  'Central',\n",
       "  'Kivell',\n",
       "  'Binder',\n",
       "  50,\n",
       "  19.99,\n",
       "  999.5],\n",
       " [datetime.datetime(2016, 2, 9, 0, 0),\n",
       "  'Central',\n",
       "  'Jardine',\n",
       "  'Pencil',\n",
       "  36,\n",
       "  4.99,\n",
       "  179.64]]"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_clean = rdd_from_file.flatMap(parseCSVRow)\n",
    "rdd_from_file_clean.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .distinct(...) transformation\n",
    "\n",
    "#### Understanding the distinct method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1, 4, 3, 6, 8, 9]\n"
     ]
    }
   ],
   "source": [
    "sampleList = [1,1,1,4,3,4,6,4,8,6,9]\n",
    "distinct = []\n",
    "seen = {}\n",
    "for elem in sampleList:\n",
    "    if elem in seen:\n",
    "        continue\n",
    "    else:\n",
    "        distinct.append(elem)\n",
    "        seen[elem] = 1\n",
    "print(distinct)        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Finding distinct in RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Desk', 'Pencil', 'Pen', 'Binder', 'Pen Set']"
      ]
     },
     "execution_count": 110,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "items = rdd_from_file_clean.map(lambda element: element[3]).distinct()\n",
    "items.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Jardine',\n",
       " 'Gill',\n",
       " 'Thompson',\n",
       " 'Howard',\n",
       " 'Smith',\n",
       " 'Jones',\n",
       " 'Sorvino',\n",
       " 'Andrews',\n",
       " 'Morgan',\n",
       " 'Parent',\n",
       " 'Kivell']"
      ]
     },
     "execution_count": 111,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_clean.map(lambda element: element[2]).distinct().collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11"
      ]
     },
     "execution_count": 112,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_clean.map(lambda element: element[2]).distinct().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .sample(...) transformation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "37"
      ]
     },
     "execution_count": 113,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_clean.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8"
      ]
     },
     "execution_count": 114,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_clean.sample(False, 0.2).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .join(...) transformation\n",
    "\n",
    "#### The mechanics of join"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('a', (2, 4)), ('b', (3, 4))]"
      ]
     },
     "execution_count": 115,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "a = sc.parallelize([('a', 2), ('b', 3)])\n",
    "b = sc.parallelize([('a', 4), ('b', 4), ('c', 4)])\n",
    "a.join(b).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Append location"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {},
   "outputs": [],
   "source": [
    "cities = sc.parallelize([('East', 'Boston'), ('Central', 'Chicago'), ('West', 'Seattle')])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[datetime.datetime(2016, 1, 6, 0, 0),\n",
       "  'East',\n",
       "  'Jones',\n",
       "  'Pencil',\n",
       "  95,\n",
       "  1.99,\n",
       "  189.05,\n",
       "  'Boston'],\n",
       " [datetime.datetime(2016, 4, 1, 0, 0),\n",
       "  'East',\n",
       "  'Jones',\n",
       "  'Binder',\n",
       "  60,\n",
       "  4.99,\n",
       "  299.4,\n",
       "  'Boston'],\n",
       " [datetime.datetime(2016, 6, 8, 0, 0),\n",
       "  'East',\n",
       "  'Jones',\n",
       "  'Binder',\n",
       "  60,\n",
       "  8.99,\n",
       "  539.4,\n",
       "  'Boston']]"
      ]
     },
     "execution_count": 117,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_clean \\\n",
    ".map(lambda element: (element[1], element)) \\\n",
    ".join(cities)\\\n",
    ".map(lambda element: element[1][0] + [element[1][1]]) \\\n",
    ".take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .repartition(...) transformation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 118,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_clean.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 119,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_repartitioned = rdd_from_file.repartition(2)\n",
    "rdd_from_file_repartitioned.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[(datetime.datetime(2017, 10, 31, 0, 0),\n",
       "   'Central',\n",
       "   'Andrews',\n",
       "   'Pencil',\n",
       "   14,\n",
       "   1.29,\n",
       "   18.06),\n",
       "  (datetime.datetime(2017, 11, 17, 0, 0),\n",
       "   'Central',\n",
       "   'Jardine',\n",
       "   'Binder',\n",
       "   11,\n",
       "   4.99,\n",
       "   54.89),\n",
       "  (datetime.datetime(2017, 5, 14, 0, 0),\n",
       "   'Central',\n",
       "   'Gill',\n",
       "   'Pencil',\n",
       "   53,\n",
       "   1.29,\n",
       "   68.37),\n",
       "  (datetime.datetime(2016, 12, 12, 0, 0),\n",
       "   'Central',\n",
       "   'Smith',\n",
       "   'Pencil',\n",
       "   67,\n",
       "   1.29,\n",
       "   86.43),\n",
       "  (datetime.datetime(2016, 8, 15, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Pencil',\n",
       "   35,\n",
       "   4.99,\n",
       "   174.65),\n",
       "  (datetime.datetime(2016, 9, 1, 0, 0),\n",
       "   'Central',\n",
       "   'Smith',\n",
       "   'Desk',\n",
       "   2,\n",
       "   125.0,\n",
       "   250.0),\n",
       "  (datetime.datetime(2017, 7, 21, 0, 0),\n",
       "   'Central',\n",
       "   'Morgan',\n",
       "   'Pen Set',\n",
       "   55,\n",
       "   12.49,\n",
       "   686.95)],\n",
       " [(datetime.datetime(2017, 9, 10, 0, 0),\n",
       "   'Central',\n",
       "   'Gill',\n",
       "   'Pencil',\n",
       "   7,\n",
       "   1.29,\n",
       "   9.03),\n",
       "  (datetime.datetime(2017, 2, 18, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Binder',\n",
       "   4,\n",
       "   4.99,\n",
       "   19.96),\n",
       "  (datetime.datetime(2016, 7, 12, 0, 0),\n",
       "   'East',\n",
       "   'Howard',\n",
       "   'Binder',\n",
       "   29,\n",
       "   1.99,\n",
       "   57.71),\n",
       "  (datetime.datetime(2016, 5, 22, 0, 0),\n",
       "   'West',\n",
       "   'Thompson',\n",
       "   'Pencil',\n",
       "   32,\n",
       "   1.99,\n",
       "   63.68),\n",
       "  (datetime.datetime(2017, 4, 10, 0, 0),\n",
       "   'Central',\n",
       "   'Andrews',\n",
       "   'Pencil',\n",
       "   66,\n",
       "   1.99,\n",
       "   131.34),\n",
       "  (datetime.datetime(2017, 3, 7, 0, 0),\n",
       "   'West',\n",
       "   'Sorvino',\n",
       "   'Binder',\n",
       "   7,\n",
       "   19.99,\n",
       "   139.93),\n",
       "  (datetime.datetime(2017, 12, 21, 0, 0),\n",
       "   'Central',\n",
       "   'Andrews',\n",
       "   'Binder',\n",
       "   28,\n",
       "   4.99,\n",
       "   139.72),\n",
       "  (datetime.datetime(2016, 4, 18, 0, 0),\n",
       "   'Central',\n",
       "   'Andrews',\n",
       "   'Pencil',\n",
       "   75,\n",
       "   1.99,\n",
       "   149.25),\n",
       "  (datetime.datetime(2017, 9, 27, 0, 0),\n",
       "   'West',\n",
       "   'Sorvino',\n",
       "   'Pen',\n",
       "   76,\n",
       "   1.99,\n",
       "   151.24),\n",
       "  (datetime.datetime(2016, 3, 15, 0, 0),\n",
       "   'West',\n",
       "   'Sorvino',\n",
       "   'Pencil',\n",
       "   56,\n",
       "   2.99,\n",
       "   167.44),\n",
       "  (datetime.datetime(2016, 2, 9, 0, 0),\n",
       "   'Central',\n",
       "   'Jardine',\n",
       "   'Pencil',\n",
       "   36,\n",
       "   4.99,\n",
       "   179.64),\n",
       "  (datetime.datetime(2016, 1, 6, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Pencil',\n",
       "   95,\n",
       "   1.99,\n",
       "   189.05),\n",
       "  (datetime.datetime(2017, 3, 24, 0, 0),\n",
       "   'Central',\n",
       "   'Jardine',\n",
       "   'Pen Set',\n",
       "   50,\n",
       "   4.99,\n",
       "   249.5),\n",
       "  (datetime.datetime(2016, 10, 5, 0, 0),\n",
       "   'Central',\n",
       "   'Morgan',\n",
       "   'Binder',\n",
       "   28,\n",
       "   8.99,\n",
       "   251.72),\n",
       "  (datetime.datetime(2016, 9, 18, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Pen Set',\n",
       "   16,\n",
       "   15.99,\n",
       "   255.84),\n",
       "  (datetime.datetime(2016, 4, 1, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Binder',\n",
       "   60,\n",
       "   4.99,\n",
       "   299.4),\n",
       "  (datetime.datetime(2016, 11, 8, 0, 0),\n",
       "   'East',\n",
       "   'Parent',\n",
       "   'Pen',\n",
       "   15,\n",
       "   19.99,\n",
       "   299.85),\n",
       "  (datetime.datetime(2017, 7, 4, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Pen Set',\n",
       "   62,\n",
       "   4.99,\n",
       "   309.38),\n",
       "  (datetime.datetime(2017, 1, 15, 0, 0),\n",
       "   'Central',\n",
       "   'Gill',\n",
       "   'Binder',\n",
       "   46,\n",
       "   8.99,\n",
       "   413.54),\n",
       "  (datetime.datetime(2016, 5, 5, 0, 0),\n",
       "   'Central',\n",
       "   'Jardine',\n",
       "   'Pencil',\n",
       "   90,\n",
       "   4.99,\n",
       "   449.1),\n",
       "  (datetime.datetime(2016, 6, 25, 0, 0),\n",
       "   'Central',\n",
       "   'Morgan',\n",
       "   'Pencil',\n",
       "   90,\n",
       "   4.99,\n",
       "   449.1),\n",
       "  (datetime.datetime(2016, 11, 25, 0, 0),\n",
       "   'Central',\n",
       "   'Kivell',\n",
       "   'Pen Set',\n",
       "   96,\n",
       "   4.99,\n",
       "   479.04),\n",
       "  (datetime.datetime(2017, 4, 27, 0, 0),\n",
       "   'East',\n",
       "   'Howard',\n",
       "   'Pen',\n",
       "   96,\n",
       "   4.99,\n",
       "   479.04),\n",
       "  (datetime.datetime(2016, 2, 26, 0, 0),\n",
       "   'Central',\n",
       "   'Gill',\n",
       "   'Pen',\n",
       "   27,\n",
       "   19.99,\n",
       "   539.73),\n",
       "  (datetime.datetime(2016, 6, 8, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Binder',\n",
       "   60,\n",
       "   8.99,\n",
       "   539.4),\n",
       "  (datetime.datetime(2016, 10, 22, 0, 0),\n",
       "   'East',\n",
       "   'Jones',\n",
       "   'Pen',\n",
       "   64,\n",
       "   8.99,\n",
       "   575.36),\n",
       "  (datetime.datetime(2017, 6, 17, 0, 0),\n",
       "   'Central',\n",
       "   'Kivell',\n",
       "   'Desk',\n",
       "   5,\n",
       "   125.0,\n",
       "   625.0),\n",
       "  (datetime.datetime(2017, 5, 31, 0, 0),\n",
       "   'Central',\n",
       "   'Gill',\n",
       "   'Binder',\n",
       "   80,\n",
       "   8.99,\n",
       "   719.2),\n",
       "  (datetime.datetime(2017, 8, 24, 0, 0),\n",
       "   'West',\n",
       "   'Sorvino',\n",
       "   'Desk',\n",
       "   3,\n",
       "   275.0,\n",
       "   825.0),\n",
       "  (datetime.datetime(2016, 1, 23, 0, 0),\n",
       "   'Central',\n",
       "   'Kivell',\n",
       "   'Binder',\n",
       "   50,\n",
       "   19.99,\n",
       "   999.5)]]"
      ]
     },
     "execution_count": 120,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_repartitioned_sorted = rdd_from_file_clean.map(lambda element: (int(element[6]), element)).repartitionAndSortWithinPartitions(2, lambda x: x).map(lambda element: tuple(element[1]))\n",
    "\n",
    "rdd_from_file_repartitioned_sorted.glom() \\\n",
    ".collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 121,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_from_file_repartitioned_sorted.getNumPartitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read in the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "OrderDate,Region,Rep,Item,Units,Unit Cost,Total\r\n",
      "1-6-16,East,Jones,Pencil,95, 1.99 , 189.05 \r\n"
     ]
    }
   ],
   "source": [
    "%%sh\n",
    "head -n 2 sample_data.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "def parseCSVRow(inputRow):\n",
    "    try:\n",
    "        rowSplit = inputRow.split(',')\n",
    "        rowSplit[0] = dt.datetime.strptime(rowSplit[0], '%m-%d-%y')\n",
    "        rowSplit[4] = int(rowSplit[4])\n",
    "        \n",
    "        for i in [5,6]:\n",
    "            rowSplit[i] = float(rowSplit[i])\n",
    "            \n",
    "        return [rowSplit]\n",
    "    except:\n",
    "        return []\n",
    "    \n",
    "rdd_clean = sc.textFile('sample_data.csv', 4).flatMap(parseCSVRow)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .take(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[datetime.datetime(2016, 1, 6, 0, 0),\n",
       "  'East',\n",
       "  'Jones',\n",
       "  'Pencil',\n",
       "  95,\n",
       "  1.99,\n",
       "  189.05],\n",
       " [datetime.datetime(2016, 1, 23, 0, 0),\n",
       "  'Central',\n",
       "  'Kivell',\n",
       "  'Binder',\n",
       "  50,\n",
       "  19.99,\n",
       "  999.5]]"
      ]
     },
     "execution_count": 124,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_clean.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[datetime.datetime(2016, 1, 6, 0, 0), 'East', 'Jones', 'Pencil', 95, 1.99, 189.05]\n",
      "[datetime.datetime(2016, 1, 23, 0, 0), 'Central', 'Kivell', 'Binder', 50, 19.99, 999.5]\n",
      "[datetime.datetime(2016, 2, 9, 0, 0), 'Central', 'Jardine', 'Pencil', 36, 4.99, 179.64]\n",
      "[datetime.datetime(2016, 2, 26, 0, 0), 'Central', 'Gill', 'Pen', 27, 19.99, 539.73]\n",
      "[datetime.datetime(2016, 3, 15, 0, 0), 'West', 'Sorvino', 'Pencil', 56, 2.99, 167.44]\n"
     ]
    }
   ],
   "source": [
    "for element in rdd_clean.takeOrdered(5, key = lambda el: el[0]):\n",
    "    print(element)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[datetime.datetime(2017, 6, 17, 0, 0), 'Central', 'Kivell', 'Desk', 5, 125.0, 625.0]\n",
      "[datetime.datetime(2016, 6, 8, 0, 0), 'East', 'Jones', 'Binder', 60, 8.99, 539.4]\n",
      "[datetime.datetime(2016, 8, 15, 0, 0), 'East', 'Jones', 'Pencil', 35, 4.99, 174.65]\n",
      "[datetime.datetime(2017, 3, 7, 0, 0), 'West', 'Sorvino', 'Binder', 7, 19.99, 139.93]\n",
      "[datetime.datetime(2016, 5, 5, 0, 0), 'Central', 'Jardine', 'Pencil', 90, 4.99, 449.1]\n"
     ]
    }
   ],
   "source": [
    "for element in rdd_clean.takeSample(False, 5, seed = 666):\n",
    "    print(element)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .collect(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 127,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(rdd_clean.collect()) == rdd_clean.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[datetime.datetime(2016, 1, 23, 0, 0), 'Central', 'Kivell', 'Binder', 50, 19.99, 999.5]\n",
      "[datetime.datetime(2016, 2, 26, 0, 0), 'Central', 'Gill', 'Pen', 27, 19.99, 539.73]\n",
      "[datetime.datetime(2016, 6, 8, 0, 0), 'East', 'Jones', 'Binder', 60, 8.99, 539.4]\n",
      "[datetime.datetime(2016, 10, 22, 0, 0), 'East', 'Jones', 'Pen', 64, 8.99, 575.36]\n",
      "[datetime.datetime(2017, 5, 31, 0, 0), 'Central', 'Gill', 'Binder', 80, 8.99, 719.2]\n",
      "[datetime.datetime(2017, 6, 17, 0, 0), 'Central', 'Kivell', 'Desk', 5, 125.0, 625.0]\n",
      "[datetime.datetime(2017, 7, 21, 0, 0), 'Central', 'Morgan', 'Pen Set', 55, 12.49, 686.95]\n",
      "[datetime.datetime(2017, 8, 24, 0, 0), 'West', 'Sorvino', 'Desk', 3, 275.0, 825.0]\n"
     ]
    }
   ],
   "source": [
    "for element in rdd_clean.filter(lambda el: el[-1] > 500).collect():\n",
    "    print(element)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .reduce(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11496.039999999999"
      ]
     },
     "execution_count": 133,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from operator import add\n",
    "total_value = rdd_clean.map(lambda el: el[-1]).reduce(add)\n",
    "total_value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11496.039999999999"
      ]
     },
     "execution_count": 134,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "total_value = rdd_clean.map(lambda el: el[-1]).reduce(lambda x, y: x+y)\n",
    "total_value"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .reduceByKey(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 135,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[datetime.datetime(2016, 1, 6, 0, 0),\n",
       "  'East',\n",
       "  'Jones',\n",
       "  'Pencil',\n",
       "  95,\n",
       "  1.99,\n",
       "  189.05],\n",
       " [datetime.datetime(2016, 1, 23, 0, 0),\n",
       "  'Central',\n",
       "  'Kivell',\n",
       "  'Binder',\n",
       "  50,\n",
       "  19.99,\n",
       "  999.5]]"
      ]
     },
     "execution_count": 135,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_clean.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 137,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('East', 3199.64), ('Central', 6949.109999999999), ('West', 1347.29)]"
      ]
     },
     "execution_count": 137,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sales_by_region = rdd_clean.map(lambda el: (el[1], el[-1])).reduceByKey(lambda x, y: x+y)\n",
    "sales_by_region.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 138,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('East', 3199.64)\n",
      "('Central', 6949.109999999999)\n",
      "('West', 1347.29)\n"
     ]
    }
   ],
   "source": [
    "for element in sales_by_region.collect():\n",
    "    print(element)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .count() action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 139,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "37"
      ]
     },
     "execution_count": 139,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_clean.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 140,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "37"
      ]
     },
     "execution_count": 140,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_clean.countApprox(1, confidence=0.7)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11"
      ]
     },
     "execution_count": 146,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sales = rdd_clean.map(lambda el: el[2])\n",
    "sales.countApproxDistinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 148,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "11"
      ]
     },
     "execution_count": 148,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sales.distinct().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .foreach(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {},
   "outputs": [],
   "source": [
    "distinct_sales = sales.distinct()\n",
    "distinct_sales.foreach(print)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .aggregate(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 157,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(11496.039999999999, 37)"
      ]
     },
     "execution_count": 157,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "seqOp = (lambda x, y: (x[0] + y, x[1] + 1))\n",
    "combOp = (lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "\n",
    "rdd_clean.map(lambda el: el[-1]).aggregate((0.0, 0), seqOp, combOp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 158,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[datetime.datetime(2016, 1, 6, 0, 0),\n",
       "  'East',\n",
       "  'Jones',\n",
       "  'Pencil',\n",
       "  95,\n",
       "  1.99,\n",
       "  189.05],\n",
       " [datetime.datetime(2016, 1, 23, 0, 0),\n",
       "  'Central',\n",
       "  'Kivell',\n",
       "  'Binder',\n",
       "  50,\n",
       "  19.99,\n",
       "  999.5],\n",
       " [datetime.datetime(2016, 2, 9, 0, 0),\n",
       "  'Central',\n",
       "  'Jardine',\n",
       "  'Pencil',\n",
       "  36,\n",
       "  4.99,\n",
       "  179.64]]"
      ]
     },
     "execution_count": 158,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_clean.take(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .aggregateByKey(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 159,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('Jardine', 933.13, 4, 233.2825)\n",
      "('Gill', 1749.8700000000001, 5, 349.97400000000005)\n",
      "('Thompson', 63.68, 1, 63.68)\n",
      "('Howard', 536.75, 2, 268.375)\n",
      "('Smith', 336.43, 2, 168.215)\n",
      "('Jones', 2363.04, 8, 295.38)\n",
      "('Sorvino', 1283.6100000000001, 4, 320.90250000000003)\n",
      "('Andrews', 438.37, 4, 109.5925)\n",
      "('Morgan', 1387.77, 3, 462.59)\n",
      "('Parent', 299.85, 1, 299.85)\n",
      "('Kivell', 2103.54, 3, 701.18)\n"
     ]
    }
   ],
   "source": [
    "seqOp = (lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "combOp = (lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "\n",
    "for element in rdd_clean.map(lambda el: (el[2], (el[-1], 1))).aggregateByKey((0.0, 0), seqOp, combOp).map(lambda el: (el[0], el[1][0], el[1][1], el[1][0]/el[1][1])).collect():\n",
    "    print(element)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .coalesce(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 160,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 160,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_clean.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 161,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 161,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_single = rdd_clean.coalesce(1)\n",
    "rdd_single.getNumPartitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### .combineByKey(...) action"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 162,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('Jardine', [('Pencil', 449.1), ('Pen Set', 249.5), ('Binder', 54.89)])\n",
      "('Gill', [('Pen', 539.73), ('Binder', 1132.74), ('Pencil', 77.4)])\n",
      "('Thompson', [('Pencil', 63.68)])\n",
      "('Howard', [('Binder', 57.71), ('Pen', 479.04)])\n",
      "('Smith', [('Desk', 250.0), ('Pencil', 86.43)])\n",
      "('Jones', [('Pencil', 363.70000000000005), ('Binder', 559.36), ('Pen Set', 565.22), ('Pen', 575.36)])\n",
      "('Sorvino', [('Pencil', 167.44), ('Binder', 139.93), ('Desk', 825.0), ('Pen', 151.24)])\n",
      "('Andrews', [('Pencil', 298.65000000000003), ('Binder', 139.72)])\n",
      "('Morgan', [('Pencil', 449.1), ('Binder', 251.72), ('Pen Set', 686.95)])\n",
      "('Parent', [('Pen', 299.85)])\n",
      "('Kivell', [('Binder', 999.5), ('Pen Set', 479.04), ('Desk', 625.0)])\n"
     ]
    }
   ],
   "source": [
    "def combiner(element):\n",
    "    return [element]\n",
    "\n",
    "def valueMerger(element1, element2):\n",
    "    element1.append(element2)\n",
    "    return element1\n",
    "\n",
    "def combinerMerger(element1, element2):\n",
    "    el1 = dict(element1)\n",
    "    for e in element2:\n",
    "        if e[0] not in el1:\n",
    "            el1[e[0]] = 0\n",
    "        el1[e[0]] += e[1]\n",
    "        \n",
    "    return list(el1.items())\n",
    "\n",
    "for element in rdd_clean.map(lambda el: (el[2], (el[3], el[-1]))).combineByKey(combiner, valueMerger, combinerMerger).collect():\n",
    "    print(element)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 165,
   "metadata": {},
   "outputs": [],
   "source": [
    "hist = rdd_clean.map(lambda el: el[-1]).histogram(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 169,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(9.0, 8)\n",
      "(108.0, 9)\n",
      "(207.0, 6)\n",
      "(306.0, 1)\n",
      "(405.0, 5)\n",
      "(504.0, 3)\n",
      "(603.0, 2)\n",
      "(702.0, 1)\n",
      "(801.0, 1)\n",
      "(900.0, 1)\n"
     ]
    }
   ],
   "source": [
    "for bucket in [(round(b, 0), v) for b, v in zip(hist[0], hist[1])]:\n",
    "    print(bucket)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}