SPARK_STREAMING_CARLOS_SANCHEZ_VEGA.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Práctica de Spark Streaming (2017 - 2018) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Carlos Sánchez Vega (Campus Ferraz) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sistemas distribuidos 3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Importación de dependencias y funciones"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark.streaming import StreamingContext\n",
    "from pyspark.streaming.kafka import KafkaUtils\n",
    "from operator import add\n",
    "from operator import sub\n",
    "from pandas import pandas as pd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Cargamos los paquetes externos que necesitamos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "packages = \"org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.1\"\n",
    "\n",
    "os.environ[\"PYSPARK_SUBMIT_ARGS\"] = (\n",
    "    \"--packages {0} pyspark-shell\".format(packages)\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creamos una ventana de 5 segundos, tal y como dice en el enunciado"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc = SparkContext(appName=\"gp\")\n",
    "ssc = StreamingContext(sc, 5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "kafkaParams = {\"metadata.broker.list\": \"localhost:9092\"}\n",
    "directKafkaStream = KafkaUtils.createDirectStream(ssc, [\"gp\"], kafkaParams)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### a) Calcular el número total de menciones recibidas por cada cuenta de usuario durante el intervalo de 5 segundos.   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "ssc.checkpoint(\"checkpoint\")\n",
    "\n",
    "# Parseamos el mensaje de entrada\n",
    "stream = directKafkaStream.map(lambda line:line[1])\n",
    "\n",
    "# El csv divide los campos por \",\". La columna que nos interesa es la de \"Mentions\"\n",
    "# que es la cuarta columna (el primer índice es el índice 0)\n",
    "total_count = (stream.map(lambda line:line.split(',')[3])\n",
    "                       # hay casos en los que la columna \"Mentions\" está vacía\n",
    "                       # en esos casos, no tendremos en cuenta el campo para el conteo\n",
    "                      .filter(lambda line: len(line)>0)                       \n",
    "                      .map(lambda line:(line,1))\n",
    "                      .reduceByKey(add))\n",
    "total_count.pprint()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:00:05\n",
      "-------------------------------------------\n",
      "('\"valeyellow46', 1)\n",
      "('\"jessansan', 2)\n",
      "('\"valeyellow46\"', 1)\n",
      "('\"marcmarquez93\"', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:00:10\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:00:15\n",
      "-------------------------------------------\n",
      "('\"libertaddigital\"', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:00:20\n",
      "-------------------------------------------\n",
      "('\"valeyellow46\"', 1)\n",
      "('\"motogp', 1)\n",
      "('\"prischilala', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:00:25\n",
      "-------------------------------------------\n",
      "('\"motogp', 3)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:00:30\n",
      "-------------------------------------------\n",
      "('\"motogp', 1)\n",
      "('\"birtymotogp\"', 1)\n",
      "('\"dhika46', 1)\n",
      "('\"bahtiar_', 1)\n",
      "('\"cindypoluta\"', 1)\n",
      "('\"marcmarquez93\"', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:00:35\n",
      "-------------------------------------------\n",
      "('\"marcmarquez93', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:00:40\n",
      "-------------------------------------------\n",
      "('\"valeyellow46\"', 1)\n",
      "('\"motogp', 1)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ssc.start()\n",
    "ssc.awaitTermination()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### b) Calcular la frecuencia total acumulada de apariciones de cada hashtag en el campo body, actualizando un ranking con los 5 hashtags con mayor frecuencia de aparición."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "ssc.checkpoint(\"checkpoint\")\n",
    "\n",
    "# Parseamos el mensaje de entrada\n",
    "stream = directKafkaStream.map(lambda line:line[1])\n",
    "\n",
    "# Los campos del csv están separados por el campo \",\". La columna que nos interesa \n",
    "# es la de \"Body\", que se corresponde con la séptima columna \n",
    "# (los índices de las columnas empiezan por 0)\n",
    "hashtags = (stream.map(lambda line: line.split(',')[6])\n",
    "                    # se separan las palabras del campo y se toman aquellos que sean \"hashtags\"\n",
    "                    # (los hashtags son los que comienzan por \"#\")\n",
    "                  .flatMap(lambda line: filter(lambda word: word[:1]=='#', list(set(line.split()))))\n",
    "                  .map(lambda hastag:(hastag ,1))                    \n",
    "                   # se realiza el conteo parcial de los hashtags encontrados \n",
    "                  .updateStateByKey(lambda aux, totalSum: \n",
    "                                    sum(aux) + totalSum if totalSum != None else sum(aux))\n",
    ")\n",
    "\n",
    "# se ordenan los hashtags para poder mostrarlos en orden descendente\n",
    "most_frequency =  (hashtags.transform(lambda rdd: rdd.context.parallelize( \n",
    "                                rdd.takeOrdered(5, key = lambda hashtag: -hashtag[1])\n",
    "                            ) \n",
    "          ))\n",
    "\n",
    "most_frequency.pprint()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:10\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:15\n",
      "-------------------------------------------\n",
      "('#marc', 1)\n",
      "('#marquesista', 1)\n",
      "('#repsol', 1)\n",
      "('#93', 1)\n",
      "('#marquez', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:20\n",
      "-------------------------------------------\n",
      "('#marc', 1)\n",
      "('#marquesista', 1)\n",
      "('#repsol', 1)\n",
      "('#motogp', 1)\n",
      "('#93', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:25\n",
      "-------------------------------------------\n",
      "('#marc', 1)\n",
      "('#marquesista', 1)\n",
      "('#repsol', 1)\n",
      "('#motogp', 1)\n",
      "('#93', 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:30\n",
      "-------------------------------------------\n",
      "('#motogp', 4)\n",
      "('#', 3)\n",
      "('#marquez', 3)\n",
      "('#qatar', 2)\n",
      "('#93', 2)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:35\n",
      "-------------------------------------------\n",
      "('#', 7)\n",
      "('#motogp', 7)\n",
      "('#marquez', 5)\n",
      "('#gp', 5)\n",
      "('#moto', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:40\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:45\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:50\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:03:55\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:00\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:05\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:10\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:15\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:20\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:25\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:30\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:35\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:40\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:45\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:50\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:04:55\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:00\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:05\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:10\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:15\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:20\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:25\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:30\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:35\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:40\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:45\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:50\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:05:55\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:00\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:05\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:10\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:15\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:20\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:25\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:30\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:35\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:40\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:45\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:50\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:06:55\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:00\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:05\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:10\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:15\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:20\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:25\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:30\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:35\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:40\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:45\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:50\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:07:55\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:00\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:05\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:10\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:15\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:20\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:25\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:30\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:35\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 18:08:40\n",
      "-------------------------------------------\n",
      "('#', 8)\n",
      "('#motogp', 7)\n",
      "('#gp', 6)\n",
      "('#moto', 6)\n",
      "('#marquez', 5)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ssc.start()\n",
    "ssc.awaitTermination()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### c) Calcular en una ventana temporal 20 segundos con offset de 10 segundos la frecuencia de aparición de cada uno de los 3 posibles tipos de tweets (TW-RT-MT)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "ssc.checkpoint(\"checkpoint\")\n",
    "stream = directKafkaStream.map(lambda line:line[1])\n",
    "\n",
    "# se define un conjunto, entre los que están los diferentes tipos de tweets\n",
    "# se escapan los caracteres de doble comilla(\")\n",
    "type_tw_rt_mt = [\"\\\"MT\\\"\", \"\\\"RT\\\"\", \"\\\"TW\\\"\"]\n",
    "\n",
    "total_count = (stream.map(lambda line: line.split(',')[9])\n",
    "                               #si el campo está entre los diferentes tipos de tweets\n",
    "                               # lo tomo en cuenta para el conteo total\n",
    "                             .filter(lambda line: line in type_tw_rt_mt)                              \n",
    "                             .map(lambda line:(line,1))\n",
    "                               # añado los parámetros de la ventana temporal (20) y el offset (10)\n",
    "                             .reduceByKeyAndWindow(add, sub, 20, 10))\n",
    "total_count.pprint()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:57:35\n",
      "-------------------------------------------\n",
      "('\"TW\"', 24)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:57:45\n",
      "-------------------------------------------\n",
      "('\"TW\"', 34)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:57:55\n",
      "-------------------------------------------\n",
      "('\"TW\"', 10)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:58:05\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:58:15\n",
      "-------------------------------------------\n",
      "('\"TW\"', 7)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:58:25\n",
      "-------------------------------------------\n",
      "('\"TW\"', 23)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:58:35\n",
      "-------------------------------------------\n",
      "('\"MT\"', 3)\n",
      "('\"TW\"', 28)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:58:45\n",
      "-------------------------------------------\n",
      "('\"MT\"', 5)\n",
      "('\"TW\"', 27)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:58:55\n",
      "-------------------------------------------\n",
      "('\"MT\"', 2)\n",
      "('\"TW\"', 31)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:59:05\n",
      "-------------------------------------------\n",
      "('\"MT\"', 5)\n",
      "('\"TW\"', 20)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:59:15\n",
      "-------------------------------------------\n",
      "('\"MT\"', 5)\n",
      "('\"TW\"', 19)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:59:25\n",
      "-------------------------------------------\n",
      "('\"RT\"', 1)\n",
      "('\"TW\"', 28)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-07-15 17:59:35\n",
      "-------------------------------------------\n",
      "('\"RT\"', 1)\n",
      "('\"TW\"', 19)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ssc.start()\n",
    "ssc.awaitTermination()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "ssc.stop()"
   ]
  }
 ],
 "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.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}