Lab_ex_2_1_solution.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Ex 2.1 Hadoop MapReduce with Python\n",
    "There are two prominent *Python* APIs for interfacing *Hadoop MapReduce* clusters:\n",
    "\n",
    "\n",
    "## *MRJOB* for *MapReduce* job execution\n",
    "The ``mrjob`` lib -> [see docu](https://mrjob.readthedocs.io/en/latest/index.html) is a power full *MapReduce* client for *Python*. Some of the key features are:\n",
    "\n",
    "* local emulation (single and multi-core) a *Hadoop* cluster for development and debugging\n",
    "* simple access, authentication and file transfer to *Hadoop* clusters\n",
    "* powerful API for common cloud services, such as AWS or Azure   "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise\n",
    "Use  *mrjob*  to  compute  employee  **top  annual  salaries** and  **gross pay** in the *CSV* table ``Baltimore_City_employee_Salaries_FY2014.csv``.\n",
    "\n",
    "* use  ``import csv`` to read the data -> [API docs](https://docs.python.org/3/library/csv.html)\n",
    "* use ``yield`` to return *producers* from *map* and *reduce* functions\n",
    "* return top entries in both categories "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing salaries.py\n"
     ]
    }
   ],
   "source": [
    "%%file salaries.py \n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "import csv\n",
    "\n",
    "#column names\n",
    "cols = 'Name,JobTitle,AgencyID,Agency,HireDate,AnnualSalary,GrossPay'.split(',')\n",
    "\n",
    "class salarymax(MRJob):\n",
    "\n",
    "    def mapper(self, _, line):\n",
    "        # Convert each line into a dictionary\n",
    "        row = dict(zip(cols, [ a.strip() for a in next(csv.reader([line]))]))\n",
    "\n",
    "        # Yield the salary\n",
    "        try:\n",
    "            yield 'salary', (float(row['AnnualSalary']), line)\n",
    "        except ValueError:\n",
    "            self.increment_counter('warn', 'missing salary', 1)\n",
    "        \n",
    "        # Yield the gross pay\n",
    "        try:\n",
    "            yield 'gross', (float(row['GrossPay']), line)\n",
    "        except ValueError:\n",
    "            self.increment_counter('warn', 'missing gross', 1)\n",
    "\n",
    "    def reducer(self, key, values):\n",
    "        topten = []\n",
    "\n",
    "        # For 'salary' and 'gross' compute the top 10\n",
    "        for p in values:\n",
    "            topten.append(p)\n",
    "            topten.sort()\n",
    "            topten = topten[-10:]\n",
    "\n",
    "        for p in topten:\n",
    "            yield key, p\n",
    "\n",
    "    combiner = reducer\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    salarymax.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No configs found; falling back on auto-configuration\n",
      "No configs specified for local runner\n",
      "Creating temp directory /tmp/salaries.keuper.20200618.164945.846521\n",
      "Running step 1 of 1...\n",
      "\n",
      "Counters: 2\n",
      "\twarn\n",
      "\t\tmissing gross=3224\n",
      "\t\tmissing salary=1\n",
      "\n",
      "job output is in /tmp/salaries.keuper.20200618.164945.846521/output\n",
      "Streaming final output from /tmp/salaries.keuper.20200618.164945.846521/output...\n",
      "\"gross\"\t[165108.5,\"\\\"Cheelsman III,Charles H\\\",Battalion Fire Chief EMS EMT-P,A64460,Fire Department ,12\\/08\\/1980,107307.00,165108.50\"]\n",
      "\"gross\"\t[165270.01,\"\\\"Makanjuola,Rafiu T\\\",POLICE OFFICER (EID),A99061,Police Department ,07\\/30\\/1997,67535.00,165270.01\"]\n",
      "\"gross\"\t[165892.21,\"\\\"Harris Jr,William\\\",POLICE SERGEANT,A99309,Police Department ,10\\/24\\/2000,80612.00,165892.21\"]\n",
      "\"gross\"\t[166442.42,\"\\\"Stokes,Charline B\\\",Battalion Fire Chief EMS EMT-P,A64460,Fire Department ,01\\/18\\/1988,107307.00,166442.42\"]\n",
      "\"gross\"\t[173876.84,\"\\\"Marcus Sr,Albert M\\\",POLICE OFFICER (EID),A99322,Police Department ,02\\/03\\/1975,73012.00,173876.84\"]\n",
      "\"gross\"\t[176141.33,\"\\\"Nalewajko Jr,Stephen C\\\",POLICE LIEUTENANT EID,A99264,Police Department ,08\\/21\\/1981,95087.00,176141.33\"]\n",
      "\"gross\"\t[185741.81,\"\\\"Charles,Ronnie E\\\",EXECUTIVE LEVEL III,A83001,HR-Human Resources ,07\\/05\\/2012,200000.00,185741.81\"]\n",
      "\"gross\"\t[188328.5,\"\\\"Black,Harry E\\\",EXECUTIVE LEVEL III,A23001,FIN-Admin & Budgets ,01\\/30\\/2012,190000.00,188328.50\"]\n",
      "\"gross\"\t[193653.69,\"\\\"Batts,Anthony W\\\",EXECUTIVE LEVEL III,A99390,Police Department ,09\\/25\\/2012,193800.00,193653.69\"]\n",
      "\"gross\"\t[238772.04,\"\\\"Bernstein,Gregg L\\\",STATE'S ATTORNEY,A29001,States Attorneys Office ,01\\/03\\/2011,238772.00,238772.04\"]\n",
      "\"salary\"\t[163200.0,\"\\\"Chow,Rudolph S\\\",DIRECTOR PUBLIC WORKS,A41101,DPW-Administration ,02\\/01\\/2011,163200.00,145513.79\"]\n",
      "\"salary\"\t[163200.0,\"\\\"Nilson,George A\\\",CITY SOLICITOR,A30001,Law Department ,01\\/16\\/2007,163200.00,164332.32\"]\n",
      "\"salary\"\t[163365.0,\"\\\"Rawlings-Blake,Stephanie C\\\",MAYOR,A01001,Mayors Office ,12\\/07\\/1995,163365.00,161219.24\"]\n",
      "\"salary\"\t[165000.0,\"\\\"Ford,Niles R\\\",EXECUTIVE LEVEL III,A64006,Fire Department ,01\\/15\\/2014,165000.00,69807.64\"]\n",
      "\"salary\"\t[172000.0,\"\\\"Parthemos,Kaliope\\\",EXECUTIVE LEVEL III,A01020,Mayor's Office ,12\\/26\\/2006,172000.00,154654.39\"]\n",
      "\"salary\"\t[187200.0,\"\\\"Swift,Michael\\\",CONTRACT SERV SPEC II,A02003,City Council ,05\\/19\\/2008,187200.00,3510.00\"]\n",
      "\"salary\"\t[190000.0,\"\\\"Black,Harry E\\\",EXECUTIVE LEVEL III,A23001,FIN-Admin & Budgets ,01\\/30\\/2012,190000.00,188328.50\"]\n",
      "\"salary\"\t[193800.0,\"\\\"Batts,Anthony W\\\",EXECUTIVE LEVEL III,A99390,Police Department ,09\\/25\\/2012,193800.00,193653.69\"]\n",
      "\"salary\"\t[200000.0,\"\\\"Charles,Ronnie E\\\",EXECUTIVE LEVEL III,A83001,HR-Human Resources ,07\\/05\\/2012,200000.00,185741.81\"]\n",
      "\"salary\"\t[238772.0,\"\\\"Bernstein,Gregg L\\\",STATE'S ATTORNEY,A29001,States Attorneys Office ,01\\/03\\/2011,238772.00,238772.04\"]\n",
      "Removing temp directory /tmp/salaries.keuper.20200618.164945.846521...\n"
     ]
    }
   ],
   "source": [
    "! python salaries.py -r local Baltimore_City_Employee_Salaries_FY2014.csv"
   ]
  },
  {
   "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}