1-MapReduce-Solutions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# MapReduce\n",
    "## El nacimiento del BigData\n",
    "---\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Según un análisis rápido de la popularidad de términos de busqueda en google terms, es a partir de 2012 cuando explota la popularidad del termino **Big Data**. A pesar de que es difícil explicar toda la casuística que pudo desencadenar la viralización del termino, es bastante probable que fuese iniciada por su mención especial por parte del World Economy Forum en su informe [Big Data, Big Impact: New Possibilities for International Development](http://www3.weforum.org/docs/WEF_TC_MFS_BigDataBigImpact_Briefing_2012.pdf) y su posterior difusión gracias a un artículo del New York Times con el suculento título [The Age of Big Data](http://www.nytimes.com/2012/02/12/sunday-review/big-datas-impact-in-the-world.html)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src='img/data_terms_trends.png' width=400>\n",
    "$$\\text{Fig.1 Importancia relativa de términos de busqueda en google relacionados con el mundo del dato}$$"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "No obstante, en contra de lo que parece indicar la gráfica anterior, deberíamos situar el nacimiento del Big Data un poco antes, concretamente en 2004, cuando J. Dean y S. Ghemawat, dos ingenieros de Google, publican un artículo titulado [MapReduce: Simplified Data Processing on Large Clusters](https://static.googleusercontent.com/media/research.google.com/en//archive/mapreduce-osdi04.pdf).\n",
    "\n",
    "En este artículo, Jeffrey Sanjay exponen de forma clara un nuevo modelo de programación que reduce a dos funciones básicas el procesado de un fichero. Estás dos funciones, *Map* y *Reduce*, inspiradas en las funciones primitivas homónimas de Lisp y otros lenguajes funcionales, consisten básicamente en contar (map) cuantas ocurrencias hay de cada tipo de elemento (por ejemplo: número de veces que se repite una palabra en un texto) y posteriormente sumar (reducir) el número de elementos que hay de cada tipo.\n",
    "\n",
    "A pesar de su sencillez, este paradigma resulta ser muy versatil ya que muchas de las tareas a realizar en grandes volumenes de datos se pueden expresar bajo este modelo.\n",
    "\n",
    "<img src='img/map-reduce-execution.png'>\n",
    "\n",
    "A pesar de su sencillez, este paradigma resulta ser muy versatil ya que muchas de las tareas a realizar en grandes volumenes de datos se pueden expresar bajo este modelo. La gracia del concepto MapReduce es que es facilmente paralelizable y permite a usuarios sin experiencia en computación distribuida usar los recursos de grandes *clusters* (conjuntos de máquinas)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Map(y)Reduce: una versión simplificada en python\n",
    "Adaptada de Data Science from scratch."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Como punto de partida, vamos a crear una función que cuente las palabras de una serie de documentos."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cargar los docs desde github"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "\n",
    "path_othello = 'https://raw.githubusercontent.com/juanhuguetgarcia/MADM-2017/master/4-BigData/data/othello.txt'\n",
    "path_mcbeth = 'https://raw.githubusercontent.com/juanhuguetgarcia/MADM-2017/master/4-BigData/data/mcbeth.txt'\n",
    "\n",
    "paths = [path_othello, path_mcbeth]\n",
    "\n",
    "docs = []\n",
    "for path in paths:\n",
    "    data = requests.get(path)\n",
    "    docs.append(data.text)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "O bien, desde local..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "local_paths = ['data/othello.txt', 'data/mcbeth.txt']\n",
    "\n",
    "docs = []\n",
    "for path in local_paths:\n",
    "    with open(path, 'r') as f:\n",
    "        docs.append(f.read())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Crea una funcion `tokenize` que dada una frase como cadena de caracteres devuelva una lista con las palabras en minúsculas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def tokenize(mensaje):\n",
    "    \"\"\"\n",
    "    Dado un mensaje, es decir, una serie de palabras en forma de cadena realiza:\n",
    "    1. Convierte a minusculas\n",
    "    2. Extrae las palabras usando una expresión regular\n",
    "    \"\"\"\n",
    "    mensaje = mensaje.lower()\n",
    "    palabras = mensaje.split()\n",
    "    return palabras"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Método clásico: Crea una función que cuente todas las palabras de los documentos de forma secuencial"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from sklearn.feature_extraction.stop_words import ENGLISH_STOP_WORDS\n",
    "\n",
    "def cuenta_palabras_secuencial(documents, stop_words={}):\n",
    "    \"\"\"\n",
    "    Cuenta las palabras que existen en una serie de documentos.\n",
    "    \"\"\"\n",
    "    # Inicializa un diccionario vacio. En el pondremos la palabras y el número\n",
    "    # de veces que aparecen en forma de clave, valor\n",
    "    wordcount = {}\n",
    "    \n",
    "    # Crea un loop que recorra los documentos.\n",
    "    for doc in documents:\n",
    "        # tokeniza el documento\n",
    "        doc_words = tokenize(doc)\n",
    "        #crea un loop que recorra las palabras del documento\n",
    "        for word in doc_words:\n",
    "            ## comprueba que la palabra no esta en la lista negra stop_words\n",
    "            if word not in stop_words:\n",
    "                # Comprueba si la palabra existe ya en el diccionario inicial.\n",
    "                # Si no existe, asignale al diccionario una nueva clave con el nombre de la\n",
    "                # palabra y dale el valor 1.\n",
    "                # En caso contrario, actualiza el valor que contiene y súmale uno.\n",
    "                if word not in wordcount:\n",
    "                    wordcount[word] = 1\n",
    "                else:\n",
    "                    wordcount[word] += 1\n",
    "            else:\n",
    "                continue\n",
    "    return wordcount  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Vamos a ver si funciona..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('othello', 303),\n",
       " ('iago', 287),\n",
       " ('macbeth', 257),\n",
       " ('thou', 219),\n",
       " ('desdemona', 188)]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "words_seq = cuenta_palabras_secuencial(docs, ENGLISH_STOP_WORDS)\n",
    "\n",
    "sorted_words_seq = sorted(words_seq.items(), key=lambda x: x[1], reverse=True)\n",
    "\n",
    "sorted_words_seq[:5]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "En el snippet de código anterior vemos que crear un contador de palabras no es complicado. Sin embargo, esta aproximación tiene dos inconvenientes mayores:\n",
    "\n",
    "1. Al ser secuencial, solo puede procesar un documento a la vez\n",
    "1. La máquina que hace el procesado tiene que tener el documento en memoria\n",
    "\n",
    "Sin embargo, si tuviesemos acceso a 1000 máquinas, podríamos utilizar el esquema Map-Reduce de forma que:\n",
    "\n",
    "1. Repartir los documentos entre las distintas máquinas\n",
    "1. Que cada máquina **mapee** los documentos generando montones de pares clave-valor\n",
    "1. Que un colector recoja los pares clave-valor y los envíe a un **reductor** de forma que cada clave única este en un sólo colector\n",
    "1. Que cada **reductor agrupe** todos los pares clave-valor **por clave y reduzca** (sume) todos los valores\n",
    "\n",
    "Vamos a definir las funciones de mapeo y reducción que nos permitirían, dada la infraestructura, hacer un conteo de palabras de forma distribuída.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## funcion de mapeo\n",
    "\n",
    "def word_count_mapper(document, stop_words={}):\n",
    "    \"\"\"\n",
    "    Para cada una de las palabras del documento, emite una tupla ('palabra', 1)\n",
    "    \"\"\"\n",
    "    ## Inicializa una lista vacia\n",
    "    word_counter = []\n",
    "    \n",
    "    # haz un loop para cada una de las palabras tokenizadas del documento y \n",
    "    # si no está en la lista negra, haz un append a la lista vacia con una tupla\n",
    "    # (word, 1)\n",
    "    for word in tokenize(document):\n",
    "        if word not in stop_words:\n",
    "            word_counter.append((word, 1))\n",
    "    return word_counter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## funcion de reducción\n",
    "\n",
    "def word_count_reducer(word_counter):\n",
    "    \"\"\"\n",
    "    Dada una tupla (word, lista_con_unos), devuelve una tupla con la palabra\n",
    "    y la suma de apariciones\n",
    "    \"\"\"\n",
    "    word = word_counter[0]\n",
    "    count_list = word_counter[1]\n",
    "    return (word, sum(count_list))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## poniendolo todo junto...\n",
    "\n",
    "from collections import defaultdict\n",
    "\n",
    "def cuenta_palabras_mapreduce(documents, stop_words={}):\n",
    "    \"\"\"\n",
    "    Cuenta las palabras de los documentos usando las funciones de wc_mapper y wc_reducer (MapReduce)\n",
    "    \"\"\"\n",
    "    collector = defaultdict(list)\n",
    "    \n",
    "    ## Map\n",
    "    for document in documents:\n",
    "        for word, count in word_count_mapper(document, stop_words):\n",
    "            collector[word].append(count)\n",
    "    \n",
    "    ## Reduce\n",
    "    reduced_words = []\n",
    "    for word_counts in collector.items():\n",
    "        reduced_words.append(word_count_reducer(word_counts))\n",
    "\n",
    "    return reduced_words"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('othello', 303),\n",
       " ('iago', 287),\n",
       " ('macbeth', 257),\n",
       " ('thou', 219),\n",
       " ('desdemona', 188)]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "words = cuenta_palabras_mapreduce(docs, ENGLISH_STOP_WORDS)\n",
    "\n",
    "sorted_words = sorted(words, key=lambda x: x[1], reverse=True)\n",
    "\n",
    "sorted_words[:5]"
   ]
  }
 ],
 "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.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}