week-4-total-spent-by-customer-sorted.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Based on\n",
    "# https://github.com/PacktPublishing/Frank-Kanes-Taming-Big-Data-with-Apache-Spark-and-Python/blob/master/total-spent-by-customer-sorted.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext\n",
    "\n",
    "conf = SparkConf().setMaster(\"local\").setAppName(\"SpendByCustomerSorted\")\n",
    "sc = SparkContext(conf = conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extractCustomerPricePairs(line):\n",
    "    fields = line.split(',')\n",
    "    return (int(fields[0]), float(fields[2]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['44,8602,37.19',\n",
       " '35,5368,65.89',\n",
       " '2,3391,40.64',\n",
       " '47,6694,14.98',\n",
       " '29,680,13.08']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "csv_input = sc.textFile(\"./data/customer-orders.csv\")\n",
    "csv_input.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(44, 37.19), (35, 65.89), (2, 40.64), (47, 14.98), (29, 13.08)]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mappedInput = csv_input.map(extractCustomerPricePairs)\n",
    "mappedInput.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(44, 4756.8899999999985),\n",
       " (35, 5155.419999999999),\n",
       " (2, 5994.59),\n",
       " (47, 4316.299999999999),\n",
       " (29, 5032.529999999999)]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "totalByCustomer = mappedInput.reduceByKey(lambda x, y: x + y)\n",
    "totalByCustomer.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(4756.8899999999985, 44),\n",
       " (5155.419999999999, 35),\n",
       " (5994.59, 2),\n",
       " (4316.299999999999, 47),\n",
       " (5032.529999999999, 29)]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Changed for Python 3 compatibility:\n",
    "#flipped = totalByCustomer.map(lambda (x,y):(y,x))\n",
    "flipped = totalByCustomer.map(lambda x: (x[1], x[0]))\n",
    "flipped.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "totalByCustomerSorted = flipped.sortByKey()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(3309.38, 45)\n",
      "(3790.570000000001, 79)\n",
      "(3924.230000000001, 96)\n",
      "(4042.6499999999987, 23)\n",
      "(4172.289999999998, 99)\n",
      "(4178.500000000001, 75)\n",
      "(4278.049999999997, 36)\n",
      "(4297.260000000001, 98)\n",
      "(4316.299999999999, 47)\n",
      "(4327.729999999999, 77)\n",
      "(4367.62, 13)\n",
      "(4384.33, 48)\n",
      "(4394.599999999999, 49)\n",
      "(4475.569999999999, 94)\n",
      "(4505.79, 67)\n",
      "(4517.27, 50)\n",
      "(4524.509999999999, 78)\n",
      "(4561.069999999999, 5)\n",
      "(4628.4, 57)\n",
      "(4635.799999999997, 83)\n",
      "(4642.259999999999, 91)\n",
      "(4647.129999999999, 74)\n",
      "(4652.939999999999, 84)\n",
      "(4659.63, 3)\n",
      "(4664.589999999998, 12)\n",
      "(4681.919999999999, 66)\n",
      "(4701.019999999999, 56)\n",
      "(4707.41, 21)\n",
      "(4727.860000000001, 80)\n",
      "(4735.030000000001, 14)\n",
      "(4735.200000000002, 37)\n",
      "(4755.070000000001, 7)\n",
      "(4756.8899999999985, 44)\n",
      "(4765.05, 31)\n",
      "(4812.489999999998, 82)\n",
      "(4815.050000000002, 4)\n",
      "(4819.700000000001, 10)\n",
      "(4830.549999999999, 88)\n",
      "(4836.859999999999, 20)\n",
      "(4851.479999999999, 89)\n",
      "(4876.840000000002, 95)\n",
      "(4898.460000000002, 38)\n",
      "(4904.209999999999, 76)\n",
      "(4908.81, 86)\n",
      "(4915.889999999999, 27)\n",
      "(4921.27, 18)\n",
      "(4945.299999999999, 53)\n",
      "(4958.600000000001, 1)\n",
      "(4975.22, 51)\n",
      "(4979.06, 16)\n",
      "(4990.72, 30)\n",
      "(5000.709999999998, 28)\n",
      "(5019.449999999999, 22)\n",
      "(5032.529999999999, 29)\n",
      "(5032.679999999999, 17)\n",
      "(5040.709999999999, 60)\n",
      "(5057.610000000001, 25)\n",
      "(5059.4299999999985, 19)\n",
      "(5112.709999999999, 81)\n",
      "(5123.010000000001, 69)\n",
      "(5140.3499999999985, 65)\n",
      "(5152.290000000002, 11)\n",
      "(5155.419999999999, 35)\n",
      "(5186.429999999999, 40)\n",
      "(5206.4, 87)\n",
      "(5245.059999999999, 52)\n",
      "(5250.4, 26)\n",
      "(5253.3200000000015, 62)\n",
      "(5254.659999999998, 33)\n",
      "(5259.920000000003, 24)\n",
      "(5265.750000000001, 93)\n",
      "(5288.689999999996, 64)\n",
      "(5290.409999999998, 90)\n",
      "(5298.090000000002, 55)\n",
      "(5322.649999999999, 9)\n",
      "(5330.8, 34)\n",
      "(5337.44, 72)\n",
      "(5368.249999999999, 70)\n",
      "(5368.83, 43)\n",
      "(5379.280000000002, 92)\n",
      "(5397.879999999998, 6)\n",
      "(5413.510000000001, 15)\n",
      "(5415.150000000001, 63)\n",
      "(5437.7300000000005, 58)\n",
      "(5496.050000000004, 32)\n",
      "(5497.479999999998, 61)\n",
      "(5503.43, 85)\n",
      "(5517.240000000001, 8)\n",
      "(5524.949999999998, 0)\n",
      "(5637.62, 41)\n",
      "(5642.89, 59)\n",
      "(5696.840000000003, 42)\n",
      "(5963.109999999999, 46)\n",
      "(5977.189999999995, 97)\n",
      "(5994.59, 2)\n",
      "(5995.660000000003, 71)\n",
      "(6065.389999999999, 54)\n",
      "(6193.109999999999, 39)\n",
      "(6206.199999999999, 73)\n",
      "(6375.449999999997, 68)\n"
     ]
    }
   ],
   "source": [
    "results = totalByCustomerSorted.collect();\n",
    "for result in results:\n",
    "    print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}