03 - RDDs con pares clave valor.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![RDD key pair](media/04.rdd_key_pair.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "# 03 - RDDs con pares clave/valor (aka Pair RDDs)\n",
    "\n",
    "-   Tipos de datos muy usados en Big Data (MapReduce)\n",
    "-   Spark dispone de operaciones especiales para su manejo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: pyspark in /home/xergioalex/Envs/tensorflow/lib/python3.5/site-packages\n",
      "Requirement already satisfied: py4j==0.10.7 in /home/xergioalex/Envs/tensorflow/lib/python3.5/site-packages (from pyspark)\n",
      "\u001b[33mYou are using pip version 9.0.1, however version 18.0 is available.\n",
      "You should consider upgrading via the 'pip install --upgrade pip' command.\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "!pip install pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create apache spark context\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext(master=\"local\", appName=\"Mi app\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Stop apache spark context\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creación de *Pair RDDs*\n",
    "Los RDDs clave/valor pueden crearse a partir de una lista de tuplas, a partir de otro RDD o mediante un zip de dos RDDs.\n",
    "-   A partir de una lista de tuplas\n",
    "-   A partir de otro RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('a', 2), ('b', 5), ('a', 3)]\n",
      "[('a', 0), ('b', 1), ('c', 2)]\n"
     ]
    }
   ],
   "source": [
    "prdd = sc.parallelize([('a',2), ('b',5), ('a',3)])\n",
    "print(prdd.collect())\n",
    "\n",
    "prdd = sc.parallelize(zip(['a', 'b', 'c'], range(3)))\n",
    "print(prdd.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 179,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Par (1ª palabra, línea): [('arenga.', 'arenga. Al cabo de lo cual, dijo:')]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Ejemplo usando un fichero\n",
    "# Para cada línea ontenemos una tupla, siendo el primer elemento\n",
    "# la primera palabra de la línes, y el segundo la línea completa\n",
    "linesrdd = sc.textFile(\"data/quijote.txt\")\n",
    "prdd = linesrdd.map(lambda x: (x.split(\" \")[0], x))\n",
    "print('Par (1ª palabra, línea): {0}\\n'.format(prdd.takeSample(False, 1)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 181,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(4, 2), (9, 3), (16, 4)]\n"
     ]
    }
   ],
   "source": [
    "nrdd = sc.parallelize(range(2,5))\n",
    "prdd = nrdd.keyBy(lambda x: x*x)\n",
    "\n",
    "print(prdd.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 182,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[['a', 'b'], ['c', 'd'], ['e', 'f', 'g', 'h']]\n",
      "[('a', 0), ('b', 1), ('c', 2), ('d', 3), ('e', 4), ('f', 5), ('g', 6), ('h', 7)]\n"
     ]
    }
   ],
   "source": [
    "# zipWithIndex(): Zipea el RDD con los índices de sus elementos.\n",
    "rdd = sc.parallelize(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'], 3)\n",
    "prdd = rdd.zipWithIndex()\n",
    "print(rdd.glom().collect())\n",
    "\n",
    "print(prdd.collect())\n",
    "\n",
    "# Este método dispara un Spark job cuando el RDD tiene más de una partición."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 183,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Particionado del RDD: [['a', 'b'], ['c', 'd'], ['e', 'f', 'g', 'h']]\n",
      "[('a', 0), ('b', 3), ('c', 1), ('d', 4), ('e', 2), ('f', 5), ('g', 8), ('h', 11)]\n"
     ]
    }
   ],
   "source": [
    "# zipWithUniqueId(): Zipea el RDD con identificadores únicos (long) para cada elemento.\n",
    "# Los elementos en la partición k-ésima obtienen los ids k, n+k, 2*n+k,... siendo n = nº de particiones\n",
    "# No dispara un trabajo Spark\n",
    "rdd = sc.parallelize(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'], 3)\n",
    "print(\"Particionado del RDD: {0}\".format(rdd.glom().collect()))\n",
    "prdd = rdd.zipWithUniqueId()\n",
    "\n",
    "print(prdd.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Mediante un zip de dos RDDs\n",
    "    * Los RDDs deben tener el mismo número de particiones y el mismo número de elementos en cada partición"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 186,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(0, 1000), (1, 1001), (2, 1002), (3, 1003), (4, 1004)]\n"
     ]
    }
   ],
   "source": [
    "rdd1 = sc.parallelize(range(0, 5), 2)\n",
    "rdd2 = sc.parallelize(range(1000, 1005), 2)\n",
    "prdd = rdd1.zip(rdd2)\n",
    "\n",
    "print(prdd.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformaciones sobre un único RDD clave/valor\n",
    "\n",
    "Sobre un único RDD clave/valor podemos efectuar transformaciones de agregación a nivel de clave y transformaciones que afectan a las claves o a los valores"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transformaciones de agregación\n",
    "\n",
    "* `reduceByKey(func)`/`foldByKey(func)`\n",
    "    -   Devuelven un RDD, agrupando los valores asociados a la misma clave mediante `func`\n",
    "    -   Similares a `reduce` y `fold` sobre RDDs simples"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 188,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('a', 10), ('b', 13)]\n"
     ]
    }
   ],
   "source": [
    "from operator import add\n",
    "prdd   = sc.parallelize([('a', 2), ('b', 5), ('a', 8), ('b', 6), ('b', 2)]).cache()\n",
    "redrdd = prdd.reduceByKey(add)\n",
    "\n",
    "print(redrdd.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-   `groupByKey()` agrupa valores asociados a misma clave\n",
    "    - Operación muy costosa en comunicaciones\n",
    "    - Mejor usar operaciones de reducción"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 190,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('a', <pyspark.resultiterable.ResultIterable object at 0x7f07c711f860>), ('b', <pyspark.resultiterable.ResultIterable object at 0x7f07c711fe80>)]\n",
      "[('a', [2, 8]), ('b', [5, 6, 2])]\n"
     ]
    }
   ],
   "source": [
    "grouprdd = prdd.groupByKey()\n",
    "\n",
    "print(grouprdd.collect())\n",
    "print\n",
    "\n",
    "lista = [(k, list(v)) for k, v in grouprdd.collect()]\n",
    "print(lista)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `combineByKey(createCombiner(func1), mergeValue(func2), mergeCombiners(func3))`\n",
    "    - Método general para agregación por clave, similar a `aggregate`\n",
    "    - Especifica tres funciones:\n",
    "\n",
    "     1.  `createCombiner` al recorrer los elementos de cada partición, si nos encontramos una clave nueva se crea un acumulador y se inicializa con `func1`\n",
    "\n",
    "     2.  `mergeValue` mezcla los valores de cada clave en cada partición usando `func2`\n",
    "\n",
    "     3.  `mergeCombiners` mezcla los resultados de las diferentes particiones mediante `func3`\n",
    "\n",
    "- Los valores del RDD de salida pueden tener un tipo diferente al de los valores del RDD de entrada."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 192,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('a', (10, 2)), ('b', (13, 3))]\n",
      "[('a', 5.0), ('b', 4.333333333333333)]\n"
     ]
    }
   ],
   "source": [
    "# Para cada clave, obten una tupla que tenga la suma y el número de valores\n",
    "sumCount = prdd.combineByKey(\n",
    "                            (lambda x: (x, 1)),\n",
    "                            (lambda x, y: (x[0]+y, x[1]+1)),\n",
    "                            (lambda x, y: (x[0]+y[0], x[1]+y[1])))\n",
    "\n",
    "print(sumCount.collect())\n",
    "\n",
    "# Con el RDD anterior, obtenemos la media de los valores\n",
    "m = sumCount.mapValues(lambda v: float(v[0])/v[1])\n",
    "print(m.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transformaciones sobre claves o valores\n",
    "-   `keys()` devuelve un RDD con las claves\n",
    "-   `values()` devuelve un RDD con los valores\n",
    "-   `sortByKey()` devuelve un RDD clave/valor con las claves ordenadas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 206,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "RDD completo: [('a', 2), ('b', 5), ('a', 8), ('b', 6), ('b', 2)]\n",
      "RDD con las claves: ['a', 'b', 'a', 'b', 'b']\n",
      "RDD con los valores: [2, 5, 8, 6, 2]\n",
      "RDD con las claves ordenadas: [('a', 2), ('a', 8), ('b', 5), ('b', 6), ('b', 2)]\n"
     ]
    }
   ],
   "source": [
    "prdd   = sc.parallelize([('a', 2), ('b', 5), ('a', 8), ('b', 6), ('b', 2)]).cache()\n",
    "print(\"RDD completo: {0}\".format(prdd.collect()))\n",
    "print(\"RDD con las claves: {0}\".format(prdd.keys().collect()))\n",
    "print(\"RDD con los valores: {0}\".format(prdd.values().collect()))\n",
    "print(\"RDD con las claves ordenadas: {0}\".format(prdd.sortByKey().collect()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-   `mapValues(func)` devuelve un RDD aplicando una función sobre los valores\n",
    "-   `flatMapValues(func)` devuelve un RDD aplicando una función sobre los valores y “aplanando” la salida"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 208,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('a', (2, 20)), ('b', (5, 50)), ('a', (8, 80)), ('b', (6, 60)), ('b', (2, 20))]\n",
      "[('a', 2), ('a', 20), ('b', 5), ('b', 50), ('a', 8), ('a', 80), ('b', 6), ('b', 60), ('b', 2), ('b', 20)]\n"
     ]
    }
   ],
   "source": [
    "mapv = prdd.mapValues(lambda x: (x, 10*x))\n",
    "print(mapv.collect())\n",
    "\n",
    "fmapv = prdd.flatMapValues(lambda x: (x, 10*x))\n",
    "print(fmapv.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transformaciones sobre dos RDDs clave/valor\n",
    "Combinan dos RDDs de tipo clave/valor para obtener un tercer RDD."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`join`/`leftOuterJoin`/`rightOuterJoin`/`fullOuterJoin` realizan inner/outer/full joins entre los dos RDDs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 214,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Join:  [('a', (2, 1)), ('a', (8, 1))]\n",
      "LeftOuterJoin:  [('b', (5, None)), ('a', (2, 1)), ('a', (8, 1))]\n",
      "RightOuterJoin:  [('c', (None, 7)), ('a', (2, 1)), ('a', (8, 1))]\n",
      "FullOuterJoin:  [('c', (None, 7)), ('b', (5, None)), ('a', (2, 1)), ('a', (8, 1))]\n"
     ]
    }
   ],
   "source": [
    "rdd1 = sc.parallelize([(\"a\", 2), (\"b\", 5), (\"a\", 8)]).cache()\n",
    "rdd2 = sc.parallelize([(\"c\", 7), (\"a\", 1)]).cache()\n",
    "\n",
    "rdd3 = rdd1.join(rdd2)\n",
    "print(\"Join: \", rdd3.collect())\n",
    "\n",
    "rdd3 = rdd1.leftOuterJoin(rdd2)\n",
    "print(\"LeftOuterJoin: \", rdd3.collect())\n",
    "\n",
    "rdd3 = rdd1.rightOuterJoin(rdd2)\n",
    "print(\"RightOuterJoin: \", rdd3.collect())\n",
    "\n",
    "rdd3 = rdd1.fullOuterJoin(rdd2)\n",
    "print(\"FullOuterJoin: \", rdd3.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-   `subtractByKey` elimina elementos con una clave presente en otro RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 220,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rd1:  [('a', 2), ('b', 5), ('a', 8)]\n",
      "rd2:  [('c', 7), ('a', 1)]\n",
      "subtractByKey:  [('b', 5)]\n"
     ]
    }
   ],
   "source": [
    "rdd3 = rdd1.subtractByKey(rdd2)\n",
    "\n",
    "print(\"rd1: \", rdd1.collect())\n",
    "print(\"rd2: \", rdd2.collect())\n",
    "print(\"subtractByKey: \", rdd3.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-   `cogroup` agrupa los datos que comparten la misma clave en ambos RDDs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 223,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('c', (<pyspark.resultiterable.ResultIterable object at 0x7f07c71feda0>, <pyspark.resultiterable.ResultIterable object at 0x7f07c71fecf8>)), ('b', (<pyspark.resultiterable.ResultIterable object at 0x7f07c71feba8>, <pyspark.resultiterable.ResultIterable object at 0x7f07c71fea90>)), ('a', (<pyspark.resultiterable.ResultIterable object at 0x7f07c71fe7b8>, <pyspark.resultiterable.ResultIterable object at 0x7f07c731db00>))]\n",
      "rd1:  [('a', 2), ('b', 5), ('a', 8)]\n",
      "rd2:  [('c', 7), ('a', 1)]\n",
      "{'b': [[5], []], 'a': [[2, 8], [1]], 'c': [[], [7]]}\n"
     ]
    }
   ],
   "source": [
    "rdd3 = rdd1.cogroup(rdd2)\n",
    "\n",
    "print(rdd3.collect())\n",
    "\n",
    "map = rdd3.mapValues(lambda v: [list(l) for l in v]).collectAsMap()\n",
    "print(\"rd1: \", rdd1.collect())\n",
    "print(\"rd2: \", rdd2.collect())\n",
    "print(map)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## Acciones sobre RDDs clave/valor\n",
    "Sobre los RDDs clave/valor podemos aplicar las acciones para RDDs simples y algunas adicionales."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-   `collectAsMap()` obtiene el RDD en forma de mapa"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "prdd:  [('a', 7), ('b', 5), ('a', 8)]\n",
      "collectAsMap:  {'b': 5, 'a': 8}\n"
     ]
    }
   ],
   "source": [
    "prdd = sc.parallelize([(\"a\", 7), (\"b\", 5), (\"a\", 8)]).cache()\n",
    "\n",
    "rddMap = prdd.collectAsMap()\n",
    "\n",
    "print(\"prdd: \", prdd.collect())\n",
    "print(\"collectAsMap: \", rddMap)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-   `countByKey()` devuelve un mapa indicando el número de ocurrencias de cada clave"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "prdd:  [('a', 7), ('b', 5), ('a', 8)]\n",
      "collectAsMap:  defaultdict(<class 'int'>, {'b': 1, 'a': 2})\n"
     ]
    }
   ],
   "source": [
    "countMap = prdd.countByKey()\n",
    "\n",
    "print(\"prdd: \", prdd.collect())\n",
    "print(\"collectAsMap: \", countMap)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `lookup(key)` devuelve una lista con los valores asociados con una clave"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "prdd:  [('a', 7), ('b', 5), ('a', 8)]\n",
      "lookup:  [7, 8]\n"
     ]
    }
   ],
   "source": [
    "listA = prdd.lookup('a')\n",
    "\n",
    "print(\"prdd: \", prdd.collect())\n",
    "print(\"lookup: \", listA)"
   ]
  }
 ],
 "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}