map_reduce.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Map Reduce\n",
    "\n",
    "[Python](#Python)<br/>\n",
    "[Hadoop (MR Job)](#Hadoop)<br/>\n",
    "[Spark](#Spark)<br/>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from __future__ import division"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "text = ['Map step: Each worker node applies the map() function to the local data, and writes the output to a temporary storage.',\n",
    "        'Shuffle step: Worker nodes redistribute data based on the output keys (produced by the map() function), such that all data belonging to one key is located on the same worker node.',\n",
    "        'Reduce step: Worker nodes now process each group of output data, per key, in parallel.']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='Python'></a>\n",
    "### Python\n",
    "\n",
    "Not distributed at all but map and reduce exist with the same basic idea.\n",
    "\n",
    "This is how they can be used:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[118, 179, 86]\n",
      "383\n"
     ]
    }
   ],
   "source": [
    "# count of words using map and reduce\n",
    "mapped = map(lambda x: len(x), text)\n",
    "print mapped\n",
    "reduced = reduce(lambda x, y: x + y, mapped)\n",
    "print reduced"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Python reduce is basically functional (think Haskell) `foldl`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "383"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# count of words using just reduce\n",
    "reduce(lambda x, y: x + len(y), text, 0) # start with 0 as the first x"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='Hadoop'></a>\n",
    "### Hadoop (MR Job)\n",
    "\n",
    "[MapReduce - Wikipedia](https://en.wikipedia.org/wiki/MapReduce)<br/>\n",
    "[MR Job Documentation](https://pythonhosted.org/mrjob/job.html)\n",
    "\n",
    "**HDFS** - File system for data storage when it does not fit on single disk.<br/>\n",
    "**Hadoop** MapReduce - Operations on distributed data, stored in HDFS. Name Node (single point of failure) and Data Nodes, data stored in blocks (usually 128 MB), each block stored 3 times (by default, can be changed).\n",
    "\n",
    "MapReduce steps:\n",
    "- **Map** - Each worker node applies the \"map()\" function to the local data, and writes the output to a temporary storage.\n",
    "- **Shuffle** - Worker nodes redistribute data based on the output keys (produced by the \"map()\" function), such that all data belonging to one key is located on the same worker node.\n",
    "- **Reduce** - Worker nodes now process each group of output data, per key, in parallel.\n",
    "\n",
    "<img src=\"./img/mapreduce_hadoop.png\" width=\"400\">\n",
    "\n",
    "MRJob lets you write Hadoop like MapReduce jobs in Python.\n",
    "\n",
    "Usefuls things to remember:\n",
    "\n",
    "- Need key value pairs throughout the whole process\n",
    "- Can yield more than one thing\n",
    "- Do not evaluate generators into lists!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from mrjob.job import MRJob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting wordlen.py\n"
     ]
    }
   ],
   "source": [
    "%%file wordlen.py\n",
    "from __future__ import division\n",
    "from mrjob.job import MRJob\n",
    "import os\n",
    "\n",
    "# avg len of words in a newsgroup topic\n",
    "class AvgWordLen(MRJob):\n",
    "\n",
    "    def mapper(self, _, line):\n",
    "        file_name = os.environ['map_input_file'].split('/')[2] #get the topic from path\n",
    "        words = line.split()\n",
    "        for word in words:\n",
    "            yield file_name, len(word)\n",
    "\n",
    "    # one combiner per mapper > same key can be in multiple combiners!\n",
    "    # combiners reduce the amount of data transferred from mapper to reucer\n",
    "    def combiner(self, key, values):\n",
    "        word_cnt = 0\n",
    "        word_len = 0\n",
    "        for word in values:\n",
    "            word_cnt += 1\n",
    "            word_len += word        \n",
    "        yield key, (word_len, word_cnt)\n",
    "            \n",
    "    def reducer(self, key, values):\n",
    "        word_cnt = 0\n",
    "        word_len = 0\n",
    "        for x, y in values:\n",
    "            word_cnt += y\n",
    "            word_len += x\n",
    "        yield key, word_len / word_cnt\n",
    "            \n",
    "    def reducer_without_combiner(self, key, values):\n",
    "        word_cnt = 0\n",
    "        word_len = 0\n",
    "        for word in values:\n",
    "            word_cnt += 1\n",
    "            word_len += word\n",
    "        yield key, word_len / word_cnt\n",
    "    \n",
    "    # this is needed for more steps or for other than default names (mapper, combiner, reducer)\n",
    "    def steps(self):\n",
    "        return [self.mr(mapper=self.mapper, reducer=self.reducer_without_combiner)]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is kind of a hack to have MR Job run in iPython notebook, the correct way is writing it into `wordlen.py` file and than calling from command as follows: \n",
    "\n",
    "```\n",
    "python wordlen.py input_file > outputs.txt\n",
    "```\n",
    "\n",
    "Running MR Job from iPython and printing out the results:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "comp.windows.x 5.8665388883\n",
      "rec.motorcycles 5.9326556544\n",
      "sci.med 6.2812920592\n"
     ]
    }
   ],
   "source": [
    "import wordlen\n",
    "reload(wordlen)\n",
    "\n",
    "mr_job = wordlen.AvgWordLen(args=['data/mini_20_newsgroups'])\n",
    "with mr_job.make_runner() as runner:\n",
    "    runner.run()\n",
    "    for line in runner.stream_output():\n",
    "        key, value = mr_job.parse_output_line(line)\n",
    "        print key, value"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='Spark'></a>\n",
    "### Apache Spark\n",
    "\n",
    "[Spark Documentation](http://spark.apache.org/docs/latest/api/python/)\n",
    "\n",
    "Not a map reduce in Hadoop sense, fully in memory.\n",
    "\n",
    "<img src=\"\">\n",
    "\n",
    "RDD = resilient distributed dataset \n",
    "\n",
    "Useful things to remember:\n",
    "- Does not need key value pairs\n",
    "- Saves a lot of I/O time compared to Hadoop (much faster)\n",
    "- Does not need key value pairs\n",
    "- RDDS are not evaluated until requested (first(), collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark as ps\n",
    "import json\n",
    "import multiprocessing\n",
    "from operator import add"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4\n"
     ]
    }
   ],
   "source": [
    "print multiprocessing.cpu_count() # figure out how many cores I have\n",
    "sc = ps.SparkContext('local[4]') # use all of them"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# create RDD from a Python list\n",
    "my_rdd = sc.parallelize(text)\n",
    "\n",
    "# create RDD from text file\n",
    "# sc.textFile('data/whatever.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Map step: Each worker node applies the map() function to the local data, and writes the output to a temporary storage.',\n",
       " 'Shuffle step: Worker nodes redistribute data based on the output keys (produced by the map() function), such that all data belonging to one key is located on the same worker node.',\n",
       " 'Reduce step: Worker nodes now process each group of output data, per key, in parallel.']"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "my_rdd.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Map step: Each worker node applies the map() function to the local data, and writes the output to a temporary storage.',\n",
       " 'Shuffle step: Worker nodes redistribute data based on the output keys (produced by the map() function), such that all data belonging to one key is located on the same worker node.']"
      ]
     },
     "execution_count": 64,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# return only elements longer than 100\n",
    "my_rdd_filtered = my_rdd.filter(lambda x: len(x) > 100)\n",
    "my_rdd_filtered.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "383\n",
      "383\n"
     ]
    }
   ],
   "source": [
    "# count of words\n",
    "my_rdd_count = (\n",
    "    my_rdd\n",
    "        .map(lambda x: len(x))\n",
    "        .sum()\n",
    ")\n",
    "print my_rdd_count\n",
    "\n",
    "my_rdd_count = (\n",
    "    my_rdd\n",
    "        .map(lambda x: len(x))\n",
    "        .reduce(lambda x, y: x + y)\n",
    ")\n",
    "print my_rdd_count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5015\n",
      "15\n"
     ]
    }
   ],
   "source": [
    "# working with numbers\n",
    "my_rdd = sc.parallelize([1,2,3,4,5])\n",
    "\n",
    "print my_rdd.fold(1000, lambda x, y: x + y)\n",
    "print my_rdd.reduce(lambda x, y: x + y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}