618_11a_Big_Data_I.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SI 618: Data Manipulation and Analysis\n",
    "## 11a - Big Data I\n",
    "\n",
    "### Dr. Chris Teplovs, School of Information, University of Michigan\n",
    "<small><a rel=\"license\" href=\"http://creativecommons.org/licenses/by-nc-sa/4.0/\"><img alt=\"Creative Commons License\" style=\"border-width:0\" src=\"https://i.creativecommons.org/l/by-nc-sa/4.0/88x31.png\" /></a>This work is licensed under a <a rel=\"license\" href=\"http://creativecommons.org/licenses/by-nc-sa/4.0/\">Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License</a>."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction to Big Data\n",
    "\n",
    "### **NOTE: conda install -y -c conda-forge mrjob**\n",
    "\n",
    "[Distributed Computing I](assets/distcomp1.pdf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generators"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "def f():\n",
    "    return 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def f123():\n",
    "    yield 1\n",
    "    yield 2\n",
    "    yield 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def f123x(x):\n",
    "    for i in range(0,x):\n",
    "        yield i"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "x = f()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<generator object f123 at 0x1070f0d68>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "f123()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "x = f123()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<generator object f123 at 0x1070f0de0>"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "2\n",
      "3\n"
     ]
    }
   ],
   "source": [
    "for thing in x:\n",
    "    print(thing)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "x = f123x(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<generator object f123x at 0x1070f0e58>"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0\n",
      "1\n",
      "2\n",
      "3\n",
      "4\n",
      "5\n",
      "6\n",
      "7\n",
      "8\n",
      "9\n"
     ]
    }
   ],
   "source": [
    "for thing in x:\n",
    "    print(thing)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "x = f123x(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0\n"
     ]
    }
   ],
   "source": [
    "for thing in x:\n",
    "    print(thing)\n",
    "    break;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n"
     ]
    }
   ],
   "source": [
    "for thing in x:\n",
    "    print(thing)\n",
    "    break;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "next(x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Our first mrjob script"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note the use of the magic command ```%%file```.  You can use this to write the contents of a cell out to a file, which is what we need to do to use mrjob:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing word_count.py\n"
     ]
    }
   ],
   "source": [
    "%%file word_count.py\n",
    "from mrjob.job import MRJob\n",
    "import re\n",
    "\n",
    "class MRWordFrequencyCount(MRJob):\n",
    "\n",
    "  ### input: self, in_key, in_value\n",
    "  def mapper(self, _, line):\n",
    "    yield \"chars\", len(line)\n",
    "    yield \"words\", len(line.split())\n",
    "    yield \"lines\", 1\n",
    "\n",
    "  ### input: self, in_key from mapper, in_value from mapper\n",
    "  def reducer(self, key, values):\n",
    "    yield key, sum(values)\n",
    "if __name__ == \"__main__\":\n",
    "    MRWordFrequencyCount.run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can run the script from the \"command line\", supplying a file to\n",
    "analyze as the first argument.  We have supplied data/simpletext.txt, which contains the text from the slides, in today's zip file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No configs found; falling back on auto-configuration\n",
      "No configs specified for inline runner\n",
      "Running step 1 of 1...\n",
      "Creating temp directory /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/word_count.xuxinye.20190408.222715.086558\n",
      "job output is in /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/word_count.xuxinye.20190408.222715.086558/output\n",
      "Streaming final output from /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/word_count.xuxinye.20190408.222715.086558/output...\n",
      "\"words\"\t9\n",
      "\"lines\"\t3\n",
      "\"chars\"\t41\n",
      "Removing temp directory /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/word_count.xuxinye.20190408.222715.086558...\n"
     ]
    }
   ],
   "source": [
    "!python word_count.py data/simpletext.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <font color=\"magenta\">Q1: Explain what the yield statements in the  above script do.  Provide a list of the first few iterations through the mapper() step would yield."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The yield statement suspends function’s execution and sends a value back to caller, but retains enough state to enable function to resume where it is left off. \n",
    "\n",
    "For Mapper(), the firest few iteration will be:\n",
    "\"chars\" 15\n",
    "\"words\" 3\n",
    "\"lines\" 1\n",
    "\n",
    "\"chars\" 13\n",
    "\"words\" 3\n",
    "\"lines\" 1\n",
    "\n",
    "\"chars\" 13\n",
    "\"words\" 3\n",
    "\"lines\" 1\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <font color=\"magenta\">Q2.  Repeat the above cell using the Tale of Two Cities text file (data/totc.txt).  Report your findings below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No configs found; falling back on auto-configuration\n",
      "No configs specified for inline runner\n",
      "Running step 1 of 1...\n",
      "Creating temp directory /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/word_count.xuxinye.20190408.223341.850439\n",
      "job output is in /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/word_count.xuxinye.20190408.223341.850439/output\n",
      "Streaming final output from /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/word_count.xuxinye.20190408.223341.850439/output...\n",
      "\"words\"\t138883\n",
      "\"lines\"\t16273\n",
      "\"chars\"\t760429\n",
      "Removing temp directory /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/word_count.xuxinye.20190408.223341.850439...\n"
     ]
    }
   ],
   "source": [
    "!python word_count.py data/totc.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Insert your interpretation of the output here."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Now let's look at a slightly more complicated example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting most_used_word.py\n"
     ]
    }
   ],
   "source": [
    "%%file most_used_word.py\n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "import re\n",
    "\n",
    "WORD_RE = re.compile(r\"[\\w']+\")\n",
    "\n",
    "\n",
    "class MRMostUsedWord(MRJob):\n",
    "    STOPWORDS = {'i', 'we', 'ourselves', 'hers', 'between', 'yourself', 'but', 'again', 'there', 'about', 'once', 'during', 'out', 'very', 'having', 'with', 'they', 'own', 'an', 'be', 'some', 'for', 'do', 'its', 'yours', 'such', 'into', 'of', 'most', 'itself', 'other', 'off', 'is', 's', 'am', 'or', 'who', 'as', 'from', 'him', 'each', 'the', 'themselves', 'until', 'below', 'are', 'we', 'these', 'your', 'his', 'through', 'don', 'nor', 'me', 'were', 'her', 'more', 'himself', 'this', 'down', 'should', 'our', 'their', 'while', 'above', 'both', 'up', 'to', 'ours', 'had', 'she', 'all', 'no', 'when', 'at', 'any', 'before', 'them', 'same', 'and', 'been', 'have', 'in', 'will', 'on', 'does', 'yourselves', 'then', 'that', 'because', 'what', 'over', 'why', 'so', 'can', 'did', 'not', 'now', 'under', 'he', 'you', 'herself', 'has', 'just', 'where', 'too', 'only', 'myself', 'which', 'those', 'i', 'after', 'few', 'whom', 't', 'being', 'if', 'theirs', 'my', 'against', 'a', 'by', 'doing', 'it', 'how', 'further', 'was', 'here', 'than'}\n",
    "    def steps(self):\n",
    "        return [\n",
    "            MRStep(mapper=self.mapper_get_words,)\n",
    "#                    combiner=self.combiner_count_words,\n",
    "#                    reducer=self.reducer_count_words),\n",
    "#             MRStep(reducer=self.reducer_find_max_word)\n",
    "        ]\n",
    "\n",
    "    def mapper_get_words(self, _, line):\n",
    "        # yield each word in the line\n",
    "        for word in WORD_RE.findall(line):\n",
    "            if word.lower() not in self.STOPWORDS:\n",
    "                yield (word.lower(), 1)\n",
    "\n",
    "    def combiner_count_words(self, word, counts):\n",
    "        # optimization: sum the words we've seen so far\n",
    "        yield (word, sum(counts))\n",
    "\n",
    "    def reducer_count_words(self, word, counts):\n",
    "        # send all (num_occurrences, word) pairs to the same reducer.\n",
    "        # num_occurrences is so we can easily use Python's max() function.\n",
    "        yield None, (sum(counts), word)\n",
    "\n",
    "    # discard the key; it is just None\n",
    "    def reducer_find_max_word(self, _, word_count_pairs):\n",
    "        # each item of word_count_pairs is (count, word),\n",
    "        # so yielding one results in key=counts, value=word\n",
    "        yield max(word_count_pairs)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    MRMostUsedWord.run()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No configs found; falling back on auto-configuration\n",
      "No configs specified for inline runner\n",
      "Running step 1 of 1...\n",
      "Creating temp directory /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/most_used_word.xuxinye.20190408.224745.454278\n",
      "job output is in /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/most_used_word.xuxinye.20190408.224745.454278/output\n",
      "Streaming final output from /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/most_used_word.xuxinye.20190408.224745.454278/output...\n",
      "\"deer\"\t1\n",
      "\"car\"\t1\n",
      "\"bear\"\t1\n",
      "\"car\"\t1\n",
      "\"car\"\t1\n",
      "\"river\"\t1\n",
      "\"dear\"\t1\n",
      "\"bear\"\t1\n",
      "\"river\"\t1\n",
      "Removing temp directory /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/most_used_word.xuxinye.20190408.224745.454278...\n"
     ]
    }
   ],
   "source": [
    "!python most_used_word.py data/simpletext.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <font color=\"magenta\">Q3: Explain what the yield statements in the  above script do.  Provide a list of the first few iterations through the mapper() step would yield."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- yield in mapper_get_words returns with each word as key and 1 as value.\n",
    "\n",
    "- yield in combiner_count_words returns a dict with each word as key and its appearance time as value.\n",
    "\n",
    "- yield in reducer_count_words returns the pairs with the from null and a list including the accumulated word count and the word.\n",
    "\n",
    "- yield in reducer_find_max_word returns the word with max word count and its counting value.\n",
    "\n",
    "we can know the first few iterations are: The mapper() yields:\n",
    "\"deer\" 1\n",
    "\"car\" 1\n",
    "\"bear\" 1\n",
    "\"car\" 1\n",
    "\"car\" 1\n",
    "\"river\" 1\n",
    "\"dear\" 1\n",
    "\"bear\" 1\n",
    "\"river\" 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### <font color=\"magenta\">Q4: Run the above script on the Tale of Two Cities text file (data/totc.txt).  What answer do you get?</font>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting most_used_word.py\n"
     ]
    }
   ],
   "source": [
    "%%file most_used_word.py\n",
    "from mrjob.job import MRJob\n",
    "from mrjob.step import MRStep\n",
    "import re\n",
    "\n",
    "WORD_RE = re.compile(r\"[\\w']+\")\n",
    "\n",
    "\n",
    "class MRMostUsedWord(MRJob):\n",
    "    STOPWORDS = {'i', 'we', 'ourselves', 'hers', 'between', 'yourself', 'but', 'again', 'there', 'about', 'once', 'during', 'out', 'very', 'having', 'with', 'they', 'own', 'an', 'be', 'some', 'for', 'do', 'its', 'yours', 'such', 'into', 'of', 'most', 'itself', 'other', 'off', 'is', 's', 'am', 'or', 'who', 'as', 'from', 'him', 'each', 'the', 'themselves', 'until', 'below', 'are', 'we', 'these', 'your', 'his', 'through', 'don', 'nor', 'me', 'were', 'her', 'more', 'himself', 'this', 'down', 'should', 'our', 'their', 'while', 'above', 'both', 'up', 'to', 'ours', 'had', 'she', 'all', 'no', 'when', 'at', 'any', 'before', 'them', 'same', 'and', 'been', 'have', 'in', 'will', 'on', 'does', 'yourselves', 'then', 'that', 'because', 'what', 'over', 'why', 'so', 'can', 'did', 'not', 'now', 'under', 'he', 'you', 'herself', 'has', 'just', 'where', 'too', 'only', 'myself', 'which', 'those', 'i', 'after', 'few', 'whom', 't', 'being', 'if', 'theirs', 'my', 'against', 'a', 'by', 'doing', 'it', 'how', 'further', 'was', 'here', 'than'}\n",
    "    def steps(self):\n",
    "        return [\n",
    "            MRStep(mapper=self.mapper_get_words,\n",
    "                   combiner=self.combiner_count_words,\n",
    "                   reducer=self.reducer_count_words),\n",
    "            MRStep(reducer=self.reducer_find_max_word)\n",
    "        ]\n",
    "\n",
    "    def mapper_get_words(self, _, line):\n",
    "        # yield each word in the line\n",
    "        for word in WORD_RE.findall(line):\n",
    "            if word.lower() not in self.STOPWORDS:\n",
    "                yield (word.lower(), 1)\n",
    "\n",
    "    def combiner_count_words(self, word, counts):\n",
    "        # optimization: sum the words we've seen so far\n",
    "        yield (word, sum(counts))\n",
    "\n",
    "    def reducer_count_words(self, word, counts):\n",
    "        # send all (num_occurrences, word) pairs to the same reducer.\n",
    "        # num_occurrences is so we can easily use Python's max() function.\n",
    "        yield None, (sum(counts), word)\n",
    "\n",
    "    # discard the key; it is just None\n",
    "    def reducer_find_max_word(self, _, word_count_pairs):\n",
    "        # each item of word_count_pairs is (count, word),\n",
    "        # so yielding one results in key=counts, value=word\n",
    "        yield max(word_count_pairs)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    MRMostUsedWord.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 inline runner\n",
      "Running step 1 of 2...\n",
      "Creating temp directory /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/most_used_word.xuxinye.20190408.224553.594293\n",
      "Running step 2 of 2...\n",
      "job output is in /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/most_used_word.xuxinye.20190408.224553.594293/output\n",
      "Streaming final output from /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/most_used_word.xuxinye.20190408.224553.594293/output...\n",
      "661\t\"said\"\n",
      "Removing temp directory /var/folders/dn/70vhnp752bbbcf4cz_6rpxmm0000gn/T/most_used_word.xuxinye.20190408.224553.594293...\n"
     ]
    }
   ],
   "source": [
    "!python most_used_word.py data/totc.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the Tale, we can get 661\t\"said\""
   ]
  },
  {
   "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.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}