Clase 9 - Ejercicios score y NBA spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "import pyspark\n",
    "\n",
    "from pyspark.sql import *\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "5ENPglW_4Cco"
   },
   "outputs": [],
   "source": [
    "# create the Spark Session\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "\n",
    "# create the Spark Context\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "Gr9ZHM8c7x8h"
   },
   "source": [
    "### 2016 2C 1 Parcial"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "-__EzR357o4r"
   },
   "source": [
    "En este ejercicio queremos programar un sistema que recomiende textos a usuarios en base a sus gustos sobre ciertos términos (palabras).\n",
    "\n",
    "Se cuenta con un RDD de textos de la forma (docId, texto) donde texto es un string de longitud variable.\n",
    "\n",
    "Además contamos con un RDD que indica qué términos le gustan o no a cada usuario de la forma (userId, término, score) por ejemplo (23, “calesita”, -2).\n",
    "\n",
    "Se pide programar en Spark un programa que calcule el score total de cada documento para cada usuario generando un RDD de la forma (userId, docId, score) en donde el score es simplemente la suma de los scores del usuario para los términos que aparecen en el documento.\n",
    "\n",
    "Puede haber términos en los documentos para los cuales no exista score de algunos usuarios, en estos casos simplemente los consideramos neutros (score=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "lNvzWFoH70p7"
   },
   "outputs": [],
   "source": [
    "documents_raw = [\n",
    "    (1, 'pablo honey'),\n",
    "    (2, 'the bends'),\n",
    "    (3, 'ok computer'),\n",
    "    (4, 'kid a'),\n",
    "    (5, 'amnesiac'),\n",
    "    (6, 'hail to the thief'),\n",
    "    (7, 'in rainbows'),\n",
    "    (8, 'the king of limbs'),\n",
    "    (9, 'a moon shaped pool')\n",
    "]\n",
    "\n",
    "scores_raw = [\n",
    "    ('thom', 'pablo', 1),\n",
    "    ('thom', 'honey', 1),\n",
    "    ('martin', 'pablo', -1),\n",
    "    ('martin', 'honey', -1),\n",
    "    ('martin', 'ok', 30),\n",
    "    ('martin', 'computer', 30),\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "nzs9K63A7-dr"
   },
   "outputs": [],
   "source": [
    "documents = sc.parallelize(documents_raw)\n",
    "scores = sc.parallelize(scores_raw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "WVs7cmvcKI-C"
   },
   "outputs": [],
   "source": [
    "terms = documents.flatMap(lambda x: [(word, x[0]) for word in x[1].split()])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "YiHW6FoVK9r_"
   },
   "outputs": [],
   "source": [
    "scores_by_word = scores.map(lambda x: (x[1], (x[0], x[2])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 119
    },
    "colab_type": "code",
    "id": "Lq-Z6aT0LMNj",
    "outputId": "18f4ab83-d1e7-441e-b316-9446d92888e6"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('honey', (1, ('thom', 1))),\n",
       " ('honey', (1, ('martin', -1))),\n",
       " ('ok', (3, ('martin', 30))),\n",
       " ('pablo', (1, ('thom', 1))),\n",
       " ('pablo', (1, ('martin', -1))),\n",
       " ('computer', (3, ('martin', 30)))]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "total = terms.join(scores_by_word)\n",
    "total.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "akPXU_UJLdX_",
    "outputId": "4bb797cf-4d2d-4e57-e29f-de5b77ca8963"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(('thom', 1), 2)"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "by_user = total.map(lambda x: ((x[1][1][0], x[1][0]), x[1][1][1])).reduceByKey(lambda x, y: x + y).cache()\n",
    "by_user.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 323
    },
    "colab_type": "code",
    "id": "qnkheR5YVP2W",
    "outputId": "07446279-2e98-4e3f-f9a0-71d210ed0ed3"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(('thom', 1), 0),\n",
       " (('thom', 2), 0),\n",
       " (('thom', 3), 0),\n",
       " (('thom', 4), 0),\n",
       " (('thom', 5), 0),\n",
       " (('thom', 6), 0),\n",
       " (('thom', 7), 0),\n",
       " (('thom', 8), 0),\n",
       " (('thom', 9), 0),\n",
       " (('martin', 1), 0),\n",
       " (('martin', 2), 0),\n",
       " (('martin', 3), 0),\n",
       " (('martin', 4), 0),\n",
       " (('martin', 5), 0),\n",
       " (('martin', 6), 0),\n",
       " (('martin', 7), 0),\n",
       " (('martin', 8), 0),\n",
       " (('martin', 9), 0)]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "users = scores.map(lambda x: x[0]).distinct() # Usuarios unicos\n",
    "docs = documents.map(lambda x: x[0])\n",
    "users_docs = users.cartesian(docs).map(lambda x: ((x[0],x[1]), 0)).cache()\n",
    "users_docs.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 323
    },
    "colab_type": "code",
    "id": "qIsRgg3VbdzM",
    "outputId": "5edea2af-d2f6-4445-c1d8-5366b889287d"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('thom', 1, 2),\n",
       " ('thom', 2, 0),\n",
       " ('thom', 3, 0),\n",
       " ('thom', 5, 0),\n",
       " ('thom', 4, 0),\n",
       " ('thom', 6, 0),\n",
       " ('thom', 7, 0),\n",
       " ('thom', 9, 0),\n",
       " ('thom', 8, 0),\n",
       " ('martin', 9, 0),\n",
       " ('martin', 8, 0),\n",
       " ('martin', 5, 0),\n",
       " ('martin', 4, 0),\n",
       " ('martin', 6, 0),\n",
       " ('martin', 7, 0),\n",
       " ('martin', 1, -2),\n",
       " ('martin', 2, 0),\n",
       " ('martin', 3, 60)]"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "by_user.rightOuterJoin(users_docs).map(lambda x: (x[0][0], x[0][1], 0 if x[1][0] is None else x[1][0])).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "UVu85ePR8DT3"
   },
   "source": [
    "### 2017 1C 1 Parcial"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "_xD-DVqW8ghO"
   },
   "source": [
    "Se tiene información estadística de la temporada regular de todos los jugadores de la NBA en un RDD de tuplas con el siguiente formato: (id_jugador, nombre, promedio_puntos, promedio_asistencias, promedio_robos, promedio_bloqueos, promedio_rebotes, promedio_faltas).\n",
    "\n",
    "Un analista de la cadena ESPN está trabajando con un RDD que corresponde a la primera ronda de playoffs y que tiene el siguiente formato: (id_jugador, id_partido, timestamp, cantidad_puntos, cantidad_rebotes, cantidad_bloqueos, cantidad_robos, cantidad_asistencias, cantidad_faltas).\n",
    "\n",
    "En base a estos RDDs se quiere programar en PySpark un programa que genere un RDD con los nombres (sin duplicados) de los jugadores que lograron en algún partido de playoffs una cantidad de asistencias mayor a su promedio histórico."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "vTpCLayaoZfh"
   },
   "outputs": [],
   "source": [
    "# (id_jugador, nombre, promedio_asistencias)\n",
    "players_all_time_stats = [\n",
    "    (1, 'Manu Ginobili', 800),\n",
    "    (2, 'Kobe Bryant', 100),\n",
    "    (3, 'Marc Gasol', 25),\n",
    "    (4, 'James Harden', 1000)]\n",
    "\n",
    "# (id_jugador, id_partido, timestamp, cantidad_asistencias)\n",
    "scores = [\n",
    "  (1, 1, 1, 100),\n",
    "  (1, 1, 3, 100),\n",
    "  (2, 1, 1, 150),\n",
    "  (2, 1, 3, 150),\n",
    "  (3, 2, 2, 50),\n",
    "  (3, 2, 3, 50),      \n",
    "  (1, 2, 1, 150),\n",
    "  (1, 2, 3, 150),\n",
    "]\n",
    "\n",
    "stats = sc.parallelize(players_all_time_stats)\n",
    "scores = sc.parallelize(scores)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "zDg7WpYsgzds",
    "outputId": "04e1d23a-5d87-4ff1-8843-d9bbd8e62b41"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(1, ('Manu Ginobili', 800))"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "stats = stats.map(lambda x: (x[0], (x[1],x[2])))\n",
    "stats.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "GKfuyH4whgcA",
    "outputId": "71676afe-3583-49e3-9f1d-f92841f2b2b4"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(1, 300)"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "scores_by_match = scores.map(lambda x: ((x[0], x[1]), x[3])).reduceByKey(lambda x, y: x + y)\\\n",
    "                                  .map(lambda x: (x[0][0], x[1]))\n",
    "scores_by_match.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "DG8PbfYNhydT"
   },
   "outputs": [],
   "source": [
    "resul = scores_by_match.join(stats).filter(lambda x: x[1][0] > x[1][1][1]).map(lambda x: (x[1][1][0])).distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "iqsTlobCj1aW",
    "outputId": "d54e0d75-6f73-48f8-d73f-2cb420d6a2cb"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Kobe Bryant', 'Marc Gasol']"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "resul.collect()"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "Clase 9 - Ejercicios score y NBA",
   "provenance": []
  },
  "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}