scala-fp-streaming.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Indice\n",
    "\n",
    "* Mi viaje en programacion: procedural -> orientada a objetos -> funcional\n",
    "* MapReduce como simplificacion de la computacion distribuida\n",
    "* Reglas para distribuir el trabajo:\n",
    "    * Funciones puras (sin mutabilidad)\n",
    "    * Sin Excepciones (Monads: Option y Try)\n",
    "* Practica local\n",
    "    * map\n",
    "    * flatMap\n",
    "    * reduce\n",
    "    * Mutabilidad y condiciones de carrera\n",
    "    * manejo de excepciones\n",
    "* En streaming\n",
    "    * akka-streams\n",
    "    * dataflow / scio\n",
    "    * apache spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Mi viaje: Basic -> C# -> Scala\n",
    "\n",
    "## Procedural\n",
    "GOTO?\n",
    "\n",
    "## OOP: clases y objetos\n",
    "perfecto, hasta que tenemos que usar varios procesadores. Lock/mutex nos restringen el uso de multiples procesadores\n",
    "\n",
    "## Funcional\n",
    "Incomodo y dificil de razonar en un principio. El mejor camino para sacarle el mejor performance a multiples procesadores"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# MapReduce como simplificacion de la computacion distribuida\n",
    "¿Como procesarias bigdata, si solo pudieras usar personas para procesar todo?\n",
    "\n",
    "* Enseñar a cada uno como procesar los datos (funcion/algoritmia)\n",
    "* Fraccionar el trabajo (en chunks) para poder distribuir parte de los datos a cada usuario (shuffle)\n",
    "* Enviar datos a los usuarios para que los procesen y nos devuelva el resultado (Map o Reduce)\n",
    "* Juntar todos los datos\n",
    "\n",
    "## Map: N -> N (chunk -> chunk)\n",
    "Entrar N registros y sale la misma cantidad\n",
    "\n",
    "## Reduce: N -> M (chunk -> dato)\n",
    "Entra N registros y sale M registros. Muchas veces sale 1 solo registro por chunk\n",
    "\n",
    "\n",
    "# Reglas para distribuir el trabajo\n",
    "\n",
    "## Funciones puras\n",
    "Cada dato debe ser calculado independiente de los datos anteriores, para evitar dependencias y condiciones de carrera\n",
    "\n",
    "## No usar excepciones\n",
    "Se deben capturar los errores sin que esos arrojen una excepcion al sistema (o uno de los trabajadores podria caerse y dejar el trabajo en un estado incierto).\n",
    "Los monad (Option/Try) nos facilitan esta tarea (su explicacion en unos minutos)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Practica local"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[32mimport \u001b[39m\u001b[36m$ivy.$                                                             \n",
       "\u001b[39m\n",
       "\u001b[32mimport \u001b[39m\u001b[36mscala.collection.parallel.CollectionConverters._\n",
       "\u001b[39m\n",
       "\u001b[32mimport \u001b[39m\u001b[36mscala.util.{Try, Success, Failure}\u001b[39m"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "// Carguemos dependencias\n",
    "import $ivy.`org.scala-lang.modules:scala-parallel-collections_2.13:1.0.3`\n",
    "import scala.collection.parallel.CollectionConverters._\n",
    "import scala.util.{Try, Success, Failure}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## map"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mduplica\u001b[39m: \u001b[32mInt\u001b[39m => \u001b[32mInt\u001b[39m = ammonite.$sess.cmd1$Helper$$Lambda$2385/1240126282@3075689a\n",
       "defined \u001b[32mfunction\u001b[39m \u001b[36mtriplica\u001b[39m\n",
       "\u001b[36mres1_2\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mVector\u001b[39m(\u001b[32m2\u001b[39m, \u001b[32m4\u001b[39m, \u001b[32m6\u001b[39m, \u001b[32m8\u001b[39m, \u001b[32m10\u001b[39m, \u001b[32m12\u001b[39m, \u001b[32m14\u001b[39m, \u001b[32m16\u001b[39m, \u001b[32m18\u001b[39m, \u001b[32m20\u001b[39m)\n",
       "\u001b[36mres1_3\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mVector\u001b[39m(\u001b[32m3\u001b[39m, \u001b[32m6\u001b[39m, \u001b[32m9\u001b[39m, \u001b[32m12\u001b[39m, \u001b[32m15\u001b[39m, \u001b[32m18\u001b[39m, \u001b[32m21\u001b[39m, \u001b[32m24\u001b[39m, \u001b[32m27\u001b[39m, \u001b[32m30\u001b[39m)"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val duplica: Int => Int = { i => i * 2 }\n",
    "def triplica(v: Int): Int = v * 3\n",
    "\n",
    "(1 to 10).map(duplica)\n",
    "(1 to 10).map(triplica)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defined \u001b[32mfunction\u001b[39m \u001b[36maumenta\u001b[39m\n",
       "\u001b[36mres2_1\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mVector\u001b[39m(\u001b[32m2\u001b[39m, \u001b[32m4\u001b[39m, \u001b[32m6\u001b[39m, \u001b[32m8\u001b[39m, \u001b[32m10\u001b[39m, \u001b[32m12\u001b[39m, \u001b[32m14\u001b[39m, \u001b[32m16\u001b[39m, \u001b[32m18\u001b[39m, \u001b[32m20\u001b[39m)\n",
       "\u001b[36mres2_2\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mVector\u001b[39m(\u001b[32m3\u001b[39m, \u001b[32m6\u001b[39m, \u001b[32m9\u001b[39m, \u001b[32m12\u001b[39m, \u001b[32m15\u001b[39m, \u001b[32m18\u001b[39m, \u001b[32m21\u001b[39m, \u001b[32m24\u001b[39m, \u001b[32m27\u001b[39m, \u001b[32m30\u001b[39m)"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def aumenta(n:Int)(v: Int): Int = n * v\n",
    "\n",
    "(1 to 10).map(aumenta(2))\n",
    "(1 to 10).map(aumenta(3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mcuatriplica\u001b[39m: \u001b[32mInt\u001b[39m => \u001b[32mInt\u001b[39m = ammonite.$sess.cmd3$Helper$$Lambda$2459/1076319808@c89cf46\n",
       "\u001b[36mres3_1\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mVector\u001b[39m(\u001b[32m4\u001b[39m, \u001b[32m8\u001b[39m, \u001b[32m12\u001b[39m, \u001b[32m16\u001b[39m, \u001b[32m20\u001b[39m, \u001b[32m24\u001b[39m, \u001b[32m28\u001b[39m, \u001b[32m32\u001b[39m, \u001b[32m36\u001b[39m, \u001b[32m40\u001b[39m)"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val cuatriplica = aumenta(4)(_)\n",
    "\n",
    "(1 to 10).map(cuatriplica)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## flatMap"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mcenso\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mString\u001b[39m] = \u001b[33mList\u001b[39m(\u001b[32m\"onka,panda,quimera\"\u001b[39m, \u001b[32m\"monty\"\u001b[39m, \u001b[32m\"canaima\"\u001b[39m, \u001b[32m\"obi,max\"\u001b[39m)\n",
       "\u001b[36mres4_1\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mArray\u001b[39m[\u001b[32mString\u001b[39m]] = \u001b[33mList\u001b[39m(\n",
       "  \u001b[33mArray\u001b[39m(\u001b[32m\"onka\"\u001b[39m, \u001b[32m\"panda\"\u001b[39m, \u001b[32m\"quimera\"\u001b[39m),\n",
       "  \u001b[33mArray\u001b[39m(\u001b[32m\"monty\"\u001b[39m),\n",
       "  \u001b[33mArray\u001b[39m(\u001b[32m\"canaima\"\u001b[39m),\n",
       "  \u001b[33mArray\u001b[39m(\u001b[32m\"obi\"\u001b[39m, \u001b[32m\"max\"\u001b[39m)\n",
       ")"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val censo = List (\n",
    "    \"onka,panda,quimera\", // casa 1\n",
    "    \"monty\", // casa 2\n",
    "    \"canaima\", // casa 3\n",
    "    \"obi,max\" // casa 4\n",
    ")\n",
    "\n",
    "censo.map { perros => perros.split(\",\") }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mres5\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mString\u001b[39m] = \u001b[33mList\u001b[39m(\n",
       "  \u001b[32m\"onka\"\u001b[39m,\n",
       "  \u001b[32m\"panda\"\u001b[39m,\n",
       "  \u001b[32m\"quimera\"\u001b[39m,\n",
       "  \u001b[32m\"monty\"\u001b[39m,\n",
       "  \u001b[32m\"canaima\"\u001b[39m,\n",
       "  \u001b[32m\"obi\"\u001b[39m,\n",
       "  \u001b[32m\"max\"\u001b[39m\n",
       ")"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "censo.map { perros => perros.split(\",\") }.flatten"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mres6\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mString\u001b[39m] = \u001b[33mList\u001b[39m(\n",
       "  \u001b[32m\"onka\"\u001b[39m,\n",
       "  \u001b[32m\"panda\"\u001b[39m,\n",
       "  \u001b[32m\"quimera\"\u001b[39m,\n",
       "  \u001b[32m\"monty\"\u001b[39m,\n",
       "  \u001b[32m\"canaima\"\u001b[39m,\n",
       "  \u001b[32m\"obi\"\u001b[39m,\n",
       "  \u001b[32m\"max\"\u001b[39m\n",
       ")"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "censo.flatMap { perros => perros.split(\",\") }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## reduce"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36medades\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mVector\u001b[39m(\u001b[32m48\u001b[39m, \u001b[32m29\u001b[39m, \u001b[32m71\u001b[39m, \u001b[32m7\u001b[39m, \u001b[32m2\u001b[39m, \u001b[32m12\u001b[39m, \u001b[32m29\u001b[39m, \u001b[32m31\u001b[39m, \u001b[32m28\u001b[39m, \u001b[32m16\u001b[39m)\n",
       "\u001b[36mmaximo\u001b[39m: (\u001b[32mInt\u001b[39m, \u001b[32mInt\u001b[39m) => \u001b[32mInt\u001b[39m = ammonite.$sess.cmd7$Helper$$Lambda$2607/1053818651@69e51f04\n",
       "\u001b[36mminimo\u001b[39m: (\u001b[32mInt\u001b[39m, \u001b[32mInt\u001b[39m) => \u001b[32mInt\u001b[39m = ammonite.$sess.cmd7$Helper$$Lambda$2608/120085331@388e9f34\n",
       "\u001b[36msuma\u001b[39m: (\u001b[32mInt\u001b[39m, \u001b[32mInt\u001b[39m) => \u001b[32mInt\u001b[39m = ammonite.$sess.cmd7$Helper$$Lambda$2609/1320308301@49015052\n",
       "\u001b[36mres7_4\u001b[39m: \u001b[32mInt\u001b[39m = \u001b[32m71\u001b[39m\n",
       "\u001b[36mres7_5\u001b[39m: \u001b[32mInt\u001b[39m = \u001b[32m2\u001b[39m\n",
       "\u001b[36mres7_6\u001b[39m: \u001b[32mInt\u001b[39m = \u001b[32m273\u001b[39m"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val edades = {    \n",
    "    val rnd = new java.util.Random\n",
    "    \n",
    "    (1 to 10).map { i =>\n",
    "        val value = rnd.nextInt % 75\n",
    "        \n",
    "        if (value > 0)\n",
    "            value\n",
    "        else\n",
    "            value * -1\n",
    "    }\n",
    "}\n",
    "\n",
    "val maximo: (Int, Int) => Int = { case (a, b) => if(a > b) a else b }\n",
    "val minimo: (Int, Int) => Int = { case (a, b) => if(a < b) a else b }\n",
    "val suma: (Int, Int) => Int = _ + _\n",
    "\n",
    "edades.reduce(maximo)\n",
    "edades.reduce(minimo)\n",
    "edades.reduce(suma)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mres8_0\u001b[39m: \u001b[32mInt\u001b[39m = \u001b[32m71\u001b[39m\n",
       "\u001b[36mres8_1\u001b[39m: \u001b[32mInt\u001b[39m = \u001b[32m2\u001b[39m\n",
       "\u001b[36mres8_2\u001b[39m: \u001b[32mInt\u001b[39m = \u001b[32m273\u001b[39m"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "edades.par.reduce(maximo)\n",
    "edades.par.reduce(minimo)\n",
    "edades.par.reduce(suma)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Mutabilidad y condiciones de carrera"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mnumTuplas\u001b[39m: \u001b[32mInt\u001b[39m = \u001b[32m8\u001b[39m\n",
       "\u001b[36mtestTupla\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[(\u001b[32mInt\u001b[39m, \u001b[32mInt\u001b[39m)] = \u001b[33mVector\u001b[39m(\n",
       "  (\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m),\n",
       "  (\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m),\n",
       "  (\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m),\n",
       "  (\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m),\n",
       "  (\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m),\n",
       "  (\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m),\n",
       "  (\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m),\n",
       "  (\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m),\n",
       "  (\u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m),\n",
       "  (\u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m),\n",
       "  (\u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m),\n",
       "  (\u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m),\n",
       "  (\u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m),\n",
       "  (\u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m),\n",
       "  (\u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m),\n",
       "  (\u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m)\n",
       ")"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val numTuplas = 8\n",
    "val testTupla =\n",
    "    (1 to numTuplas).map { _ => (1, 2) } ++\n",
    "    (1 to numTuplas).map { _ => (2, 3) }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defined \u001b[32mfunction\u001b[39m \u001b[36msuma\u001b[39m\n",
       "\u001b[36mres10_1\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mVector\u001b[39m(\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m\n",
       ")"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "// funcion pura\n",
    "def suma(a: Int, b: Int) = a + b\n",
    "\n",
    "testTupla.map { case (a,b) => suma(a, b) }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mres11\u001b[39m: \u001b[32mcollection\u001b[39m.\u001b[32mparallel\u001b[39m.\u001b[32mimmutable\u001b[39m.\u001b[32mParSeq\u001b[39m[\u001b[32mInt\u001b[39m] = ParVector(3, 3, 3, 3, 3, 3, 3, 3, 5, 5, 5, 5, 5, 5, 5, 5)"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "testTupla.par.map { case (a,b) => suma(a, b) }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div class=\"jp-RenderedText\">\n",
       "<pre><code><span style=\"color: rgb(0, 187, 187)\"><span class=\"ansi-cyan-fg\">acumulado</span></span>: <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">Int</span></span> = <style>@keyframes fadein { from { opacity: 0; } to { opacity: 1; } }</style><span style=\"animation: fadein 2s;\"><span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">15</span></span></span>\n",
       "defined <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">function</span></span> <span style=\"color: rgb(0, 187, 187)\"><span class=\"ansi-cyan-fg\">acumula</span></span>\n",
       "<span style=\"color: rgb(0, 187, 187)\"><span class=\"ansi-cyan-fg\">res12_2</span></span>: <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">IndexedSeq</span></span>[<span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">Int</span></span>] = <span style=\"color: yellow\"><span class=\"ansi-yellow-fg\">Vector</span></span>(\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">1</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">2</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">3</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">4</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">5</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">6</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">7</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">8</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">10</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">12</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">14</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">16</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">18</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">20</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">22</span></span>,\n",
       "  <span style=\"color: rgb(0, 187, 0)\"><span class=\"ansi-green-fg\">24</span></span>\n",
       ")</code></pre>\n",
       "</div>"
      ],
      "text/plain": [
       "\u001b[36macumulado\u001b[39m: \u001b[32mInt\u001b[39m = \u001b[32m15\u001b[39m\n",
       "defined \u001b[32mfunction\u001b[39m \u001b[36macumula\u001b[39m\n",
       "\u001b[36mres12_2\u001b[39m: \u001b[32mIndexedSeq\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mVector\u001b[39m(\n",
       "  \u001b[32m1\u001b[39m,\n",
       "  \u001b[32m2\u001b[39m,\n",
       "  \u001b[32m3\u001b[39m,\n",
       "  \u001b[32m4\u001b[39m,\n",
       "  \u001b[32m5\u001b[39m,\n",
       "  \u001b[32m6\u001b[39m,\n",
       "  \u001b[32m7\u001b[39m,\n",
       "  \u001b[32m8\u001b[39m,\n",
       "  \u001b[32m10\u001b[39m,\n",
       "  \u001b[32m12\u001b[39m,\n",
       "  \u001b[32m14\u001b[39m,\n",
       "  \u001b[32m16\u001b[39m,\n",
       "  \u001b[32m18\u001b[39m,\n",
       "  \u001b[32m20\u001b[39m,\n",
       "  \u001b[32m22\u001b[39m,\n",
       "  \u001b[32m24\u001b[39m\n",
       ")"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "// funcion impura\n",
    "var acumulado = 0\n",
    "def acumula(a: Int) = {\n",
    "    acumulado = acumulado + a\n",
    "    \n",
    "    acumulado\n",
    "}\n",
    "\n",
    "testTupla.map { case (a,b) => acumula(a) }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mres13_1\u001b[39m: \u001b[32mcollection\u001b[39m.\u001b[32mparallel\u001b[39m.\u001b[32mimmutable\u001b[39m.\u001b[32mParSeq\u001b[39m[\u001b[32mInt\u001b[39m] = ParVector(1, 2, 4, 5, 5, 6, 6, 7, 2, 3, 5, 7, 9, 11, 13, 15)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "// reseteamos acumulador\n",
    "acumulado = 0\n",
    "// resultado no es igual al anterior\n",
    "testTupla.par.map { case (a,b) => acumula(a) }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Manejo de excepciones"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mintText\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mString\u001b[39m] = \u001b[33mList\u001b[39m(\u001b[32m\"1\"\u001b[39m, \u001b[32m\"2\"\u001b[39m, \u001b[32m\"3\"\u001b[39m, \u001b[32m\"cuatro\"\u001b[39m, \u001b[32m\"5\"\u001b[39m, \u001b[32m\"-1\"\u001b[39m)"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val intText = List(\"1\", \"2\", \"3\", \"cuatro\", \"5\", \"-1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "ename": "",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[31mjava.lang.NumberFormatException: For input string: \"cuatro\"\u001b[39m\n  java.lang.NumberFormatException.forInputString(\u001b[32mNumberFormatException.java\u001b[39m:\u001b[32m65\u001b[39m)\n  java.lang.Integer.parseInt(\u001b[32mInteger.java\u001b[39m:\u001b[32m580\u001b[39m)\n  java.lang.Integer.parseInt(\u001b[32mInteger.java\u001b[39m:\u001b[32m615\u001b[39m)\n  scala.collection.StringOps$.toInt$extension(\u001b[32mStringOps.scala\u001b[39m:\u001b[32m889\u001b[39m)\n  ammonite.$sess.cmd15$Helper.$anonfun$res15$1(\u001b[32mcmd15.sc\u001b[39m:\u001b[32m1\u001b[39m)\n  ammonite.$sess.cmd15$Helper.$anonfun$res15$1$adapted(\u001b[32mcmd15.sc\u001b[39m:\u001b[32m1\u001b[39m)\n  scala.collection.immutable.List.map(\u001b[32mList.scala\u001b[39m:\u001b[32m250\u001b[39m)\n  ammonite.$sess.cmd15$Helper.<init>(\u001b[32mcmd15.sc\u001b[39m:\u001b[32m1\u001b[39m)\n  ammonite.$sess.cmd15$.<clinit>(\u001b[32mcmd15.sc\u001b[39m:\u001b[32m7\u001b[39m)"
     ]
    }
   ],
   "source": [
    "intText.map { t => t.toInt }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mres16\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mList\u001b[39m(\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m, \u001b[32m-1\u001b[39m, \u001b[32m5\u001b[39m, \u001b[32m-1\u001b[39m)"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "intText.map{ i => try {\n",
    "        i.toInt\n",
    "    } catch {\n",
    "        // que devuelvo si no puedo procesar\n",
    "        case ex: Throwable => -1\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mres17\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mTry\u001b[39m[\u001b[32mInt\u001b[39m]] = \u001b[33mList\u001b[39m(\n",
       "  \u001b[33mSuccess\u001b[39m(\u001b[32m1\u001b[39m),\n",
       "  \u001b[33mSuccess\u001b[39m(\u001b[32m2\u001b[39m),\n",
       "  \u001b[33mSuccess\u001b[39m(\u001b[32m3\u001b[39m),\n",
       "  \u001b[33mFailure\u001b[39m(java.lang.NumberFormatException: For input string: \"cuatro\"),\n",
       "  \u001b[33mSuccess\u001b[39m(\u001b[32m5\u001b[39m),\n",
       "  \u001b[33mSuccess\u001b[39m(\u001b[32m-1\u001b[39m)\n",
       ")"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "intText.map { i => Try { i.toInt } }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\u001b[36mres18_0\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mOption\u001b[39m[\u001b[32mInt\u001b[39m]] = \u001b[33mList\u001b[39m(\n",
       "  \u001b[33mSome\u001b[39m(\u001b[32m1\u001b[39m),\n",
       "  \u001b[33mSome\u001b[39m(\u001b[32m2\u001b[39m),\n",
       "  \u001b[33mSome\u001b[39m(\u001b[32m3\u001b[39m),\n",
       "  \u001b[32mNone\u001b[39m,\n",
       "  \u001b[33mSome\u001b[39m(\u001b[32m5\u001b[39m),\n",
       "  \u001b[33mSome\u001b[39m(\u001b[32m-1\u001b[39m)\n",
       ")\n",
       "\u001b[36mres18_1\u001b[39m: \u001b[32mList\u001b[39m[\u001b[32mInt\u001b[39m] = \u001b[33mList\u001b[39m(\u001b[32m1\u001b[39m, \u001b[32m2\u001b[39m, \u001b[32m3\u001b[39m, \u001b[32m5\u001b[39m, \u001b[32m-1\u001b[39m)"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "intText.map { i => Try { i.toInt }.toOption }\n",
    "intText.flatMap { i => Try { i.toInt }.toOption }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# En streaming\n",
    "\n",
    "## akka-streams\n",
    "vamos al notebook akka-streams!\n",
    "\n",
    "## [apache beam](https://beam.apache.org) / [scio](https://github.com/spotify/scio) y [apache spark](https://spark.apache.org)\n",
    "ambos siguen los mismos principios de programacion funcional.\n",
    "las [UDF](https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/expressions/UserDefinedFunction.html) de scala usan una funcion como entrada "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ¡Muchas gracias a todos!"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Scala 2.13",
   "language": "scala",
   "name": "scala213"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".sc",
   "mimetype": "text/x-scala",
   "name": "scala",
   "nbconvert_exporter": "script",
   "version": "2.13.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}