dataframes20201204.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## CARLOS SÁNCHEZ VEGA\n",
    "\n",
    "# DATAFRAME EXERCISES"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We import libraries:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import Row\n",
    "from pyspark.sql import functions as func\n",
    "from pyspark.sql.functions import col, asc,desc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create the spark configuration:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"FriendsByAge\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines = spark.read.option(\"header\", \"true\").option(\"inferSchema\", \"true\").csv(\"fakefriends-header.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+---+-------+\n",
      "|userID|    name|age|friends|\n",
      "+------+--------+---+-------+\n",
      "|     0|    Will| 33|    385|\n",
      "|     1|Jean-Luc| 26|      2|\n",
      "|     2|    Hugh| 55|    221|\n",
      "|     3|  Deanna| 40|    465|\n",
      "|     4|   Quark| 68|     21|\n",
      "|     5|  Weyoun| 59|    318|\n",
      "|     6|  Gowron| 37|    220|\n",
      "|     7|    Will| 54|    307|\n",
      "|     8|  Jadzia| 38|    380|\n",
      "|     9|    Hugh| 27|    181|\n",
      "|    10|     Odo| 53|    191|\n",
      "|    11|     Ben| 57|    372|\n",
      "|    12|   Keiko| 54|    253|\n",
      "|    13|Jean-Luc| 56|    444|\n",
      "|    14|    Hugh| 43|     49|\n",
      "|    15|     Rom| 36|     49|\n",
      "|    16|  Weyoun| 22|    323|\n",
      "|    17|     Odo| 35|     13|\n",
      "|    18|Jean-Luc| 45|    455|\n",
      "|    19|  Geordi| 60|    246|\n",
      "+------+--------+---+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "lines.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<ul>\n",
    "<li>First column: identifier</li>\n",
    "<li>Second column: name</li>\n",
    "<li>Third column: age</li>\n",
    "<li>Fourth column: number of friends</li>\n",
    "</ul> "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userID: integer (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- age: integer (nullable = true)\n",
      " |-- friends: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "lines.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-14-fa7555bb2397>, line 1)",
     "output_type": "error",
     "traceback": [
      "\u001b[0;36m  File \u001b[0;32m\"<ipython-input-14-fa7555bb2397>\"\u001b[0;36m, line \u001b[0;32m1\u001b[0m\n\u001b[0;31m    1. <strong>Count the number of people aged 21, at most</strong>\u001b[0m\n\u001b[0m                     ^\u001b[0m\n\u001b[0;31mSyntaxError\u001b[0m\u001b[0;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "1. <strong>Count the number of people aged 21, at most</strong>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "2. <strong>Get the average number of friends broken down by age using dataframes</strong>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this case, we should select the columns of age and friends"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The are asking us to get the number of occurrences of a mark in the RDD, so we could use, for example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "friendsAndAge=lines.select(\"age\", \"friends\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+\n",
      "|age|friends|\n",
      "+---+-------+\n",
      "| 33|    385|\n",
      "| 26|      2|\n",
      "| 55|    221|\n",
      "| 40|    465|\n",
      "| 68|     21|\n",
      "| 59|    318|\n",
      "| 37|    220|\n",
      "| 54|    307|\n",
      "| 38|    380|\n",
      "| 27|    181|\n",
      "| 53|    191|\n",
      "| 57|    372|\n",
      "| 54|    253|\n",
      "| 56|    444|\n",
      "| 43|     49|\n",
      "| 36|     49|\n",
      "| 22|    323|\n",
      "| 35|     13|\n",
      "| 45|    455|\n",
      "| 60|    246|\n",
      "+---+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "friendsAndAge.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We get the average number of friends per age"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------------------+\n",
      "|age|      avg(friends)|\n",
      "+---+------------------+\n",
      "| 31|            267.25|\n",
      "| 65|             298.2|\n",
      "| 53|222.85714285714286|\n",
      "| 34|             245.5|\n",
      "| 28|             209.1|\n",
      "| 26|242.05882352941177|\n",
      "| 27|           228.125|\n",
      "| 44| 282.1666666666667|\n",
      "| 22|206.42857142857142|\n",
      "| 47|233.22222222222223|\n",
      "| 52| 340.6363636363636|\n",
      "| 40| 250.8235294117647|\n",
      "| 20|             165.0|\n",
      "| 57| 258.8333333333333|\n",
      "| 54| 278.0769230769231|\n",
      "| 48|             281.4|\n",
      "| 19|213.27272727272728|\n",
      "| 64| 281.3333333333333|\n",
      "| 41|268.55555555555554|\n",
      "| 43|230.57142857142858|\n",
      "+---+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "friendsAndAge.groupBy(\"age\").avg(\"friends\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can see we can format the output as:\n",
    "   * We can limit number of decimals shown\n",
    "   * We can sort the result to find which is the age most friendly (considered as the one having the max number of friends)\n",
    "   * We can change column name \"avg(friends)\" for a more readible name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+\n",
      "|age|friends_avg|\n",
      "+---+-----------+\n",
      "| 18|     343.38|\n",
      "| 19|     213.27|\n",
      "| 20|      165.0|\n",
      "| 21|     350.88|\n",
      "| 22|     206.43|\n",
      "| 23|      246.3|\n",
      "| 24|      233.8|\n",
      "| 25|     197.45|\n",
      "| 26|     242.06|\n",
      "| 27|     228.13|\n",
      "| 28|      209.1|\n",
      "| 29|     215.92|\n",
      "| 30|     235.82|\n",
      "| 31|     267.25|\n",
      "| 32|     207.91|\n",
      "| 33|     325.33|\n",
      "| 34|      245.5|\n",
      "| 35|     211.63|\n",
      "| 36|      246.6|\n",
      "| 37|     249.33|\n",
      "+---+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "friendsAndAge.groupBy(\"age\").agg(func.round(func.avg(\"friends\"), 2)\n",
    "  .alias(\"friends_avg\")).sort(\"age\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "3. <strong>Get the the number of people who are teenagers (aged 13 - 19) and the number of people by age</strong>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"FriendsByAge\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines = spark.sparkContext.textFile(\"fakefriends.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0,Will,33,385\n",
      "1,Jean-Luc,26,2\n",
      "2,Hugh,55,221\n",
      "3,Deanna,40,465\n",
      "4,Quark,68,21\n",
      "5,Weyoun,59,318\n",
      "6,Gowron,37,220\n",
      "7,Will,54,307\n",
      "8,Jadzia,38,380\n",
      "9,Hugh,27,181\n"
     ]
    }
   ],
   "source": [
    "for i in lines.take(10):print(i)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we can se, input file does not have header nor doest it have any strucure of columns. Firstly, we have to format the input"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "def mapper(line):\n",
    "    fields = line.split(',')\n",
    "    return Row(ID=int(fields[0]), name=str(fields[1].encode(\"utf-8\")), \\\n",
    "               age=int(fields[2]), numFriends=int(fields[3]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "linesFormatted = lines.map(mapper)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row(ID=0, name=\"b'Will'\", age=33, numFriends=385)\n",
      "Row(ID=1, name=\"b'Jean-Luc'\", age=26, numFriends=2)\n",
      "Row(ID=2, name=\"b'Hugh'\", age=55, numFriends=221)\n",
      "Row(ID=3, name=\"b'Deanna'\", age=40, numFriends=465)\n",
      "Row(ID=4, name=\"b'Quark'\", age=68, numFriends=21)\n",
      "Row(ID=5, name=\"b'Weyoun'\", age=59, numFriends=318)\n",
      "Row(ID=6, name=\"b'Gowron'\", age=37, numFriends=220)\n",
      "Row(ID=7, name=\"b'Will'\", age=54, numFriends=307)\n",
      "Row(ID=8, name=\"b'Jadzia'\", age=38, numFriends=380)\n",
      "Row(ID=9, name=\"b'Hugh'\", age=27, numFriends=181)\n"
     ]
    }
   ],
   "source": [
    "for i in linesFormatted.take(10):print(i)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As shown above, we have an RDD, but to make it simple we will solve the problem using dataframes. For that purpose, we convert the RDD to dataframe.\n",
    "<strong>It is important to note that, to improve performance, as we are going for various tasks, we will use the cache function.</strong>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "peopleDf= linesFormatted.toDF().cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+---+----------+\n",
      "| ID|       name|age|numFriends|\n",
      "+---+-----------+---+----------+\n",
      "|  0|    b'Will'| 33|       385|\n",
      "|  1|b'Jean-Luc'| 26|         2|\n",
      "|  2|    b'Hugh'| 55|       221|\n",
      "|  3|  b'Deanna'| 40|       465|\n",
      "|  4|   b'Quark'| 68|        21|\n",
      "|  5|  b'Weyoun'| 59|       318|\n",
      "|  6|  b'Gowron'| 37|       220|\n",
      "|  7|    b'Will'| 54|       307|\n",
      "|  8|  b'Jadzia'| 38|       380|\n",
      "|  9|    b'Hugh'| 27|       181|\n",
      "| 10|     b'Odo'| 53|       191|\n",
      "| 11|     b'Ben'| 57|       372|\n",
      "| 12|   b'Keiko'| 54|       253|\n",
      "| 13|b'Jean-Luc'| 56|       444|\n",
      "| 14|    b'Hugh'| 43|        49|\n",
      "| 15|     b'Rom'| 36|        49|\n",
      "| 16|  b'Weyoun'| 22|       323|\n",
      "| 17|     b'Odo'| 35|        13|\n",
      "| 18|b'Jean-Luc'| 45|       455|\n",
      "| 19|  b'Geordi'| 60|       246|\n",
      "+---+-----------+---+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "peopleDf.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "we will solve it by SQL means"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "peopleDf.createOrReplaceTempView(\"people\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "An now the query to gett the people aged between 13 & 19 (teenagers) is so easy :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [],
   "source": [
    "teenagers = spark.sql(\"select count(age) as number_of_teenagers from people where age>=13 and age <= 19\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|number_of_teenagers|\n",
      "+-------------------+\n",
      "|                 19|\n",
      "+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "teenagers.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we will have to get the number of people by age"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+\n",
      "|age|count|\n",
      "+---+-----+\n",
      "| 18|    8|\n",
      "| 19|   11|\n",
      "| 20|    5|\n",
      "| 21|    8|\n",
      "| 22|    7|\n",
      "| 23|   10|\n",
      "| 24|    5|\n",
      "| 25|   11|\n",
      "| 26|   17|\n",
      "| 27|    8|\n",
      "| 28|   10|\n",
      "| 29|   12|\n",
      "| 30|   11|\n",
      "| 31|    8|\n",
      "| 32|   11|\n",
      "| 33|   12|\n",
      "| 34|    6|\n",
      "| 35|    8|\n",
      "| 36|   10|\n",
      "| 37|    9|\n",
      "+---+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "peopleDf.groupBy(\"age\").count().orderBy(\"age\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "4. <strong>Implement the word count solution with dataframes</strong>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"WordCount\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We read the book file as a dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "book = spark.read.text(\"Book\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will split the text using a regular expression that extracts words.\n",
    "\"explode\" function works as \"flatMap\" function, using a defined function for each of the elements (the regular expression)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[word: string]"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "words = book.select(func.explode(func.split(book.value, \"\\\\W+\")).alias(\"word\"))\n",
    "words.filter(words.word != \"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|      word|\n",
      "+----------+\n",
      "|      Self|\n",
      "|Employment|\n",
      "|  Building|\n",
      "|        an|\n",
      "|  Internet|\n",
      "|  Business|\n",
      "|        of|\n",
      "|       One|\n",
      "| Achieving|\n",
      "| Financial|\n",
      "|       and|\n",
      "|  Personal|\n",
      "|   Freedom|\n",
      "|   through|\n",
      "|         a|\n",
      "| Lifestyle|\n",
      "|Technology|\n",
      "|  Business|\n",
      "|        By|\n",
      "|     Frank|\n",
      "+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "words.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we can see, we have a dataframe composed of just one column (word). We will have to create a function to apply a lower case function to all words)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "lowercaseWords = words.select(func.lower(words.word).alias(\"word\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|      word|\n",
      "+----------+\n",
      "|      self|\n",
      "|employment|\n",
      "|  building|\n",
      "|        an|\n",
      "|  internet|\n",
      "|  business|\n",
      "|        of|\n",
      "|       one|\n",
      "| achieving|\n",
      "| financial|\n",
      "|       and|\n",
      "|  personal|\n",
      "|   freedom|\n",
      "|   through|\n",
      "|         a|\n",
      "| lifestyle|\n",
      "|technology|\n",
      "|  business|\n",
      "|        by|\n",
      "|     frank|\n",
      "+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "lowercaseWords.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we will group by word to count its appearances"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'lowercaseWords' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-33-576afd2d4580>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mlowercaseWords\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgroupBy\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"word\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcount\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0morderBy\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcol\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"count\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdesc\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m: name 'lowercaseWords' is not defined"
     ]
    }
   ],
   "source": [
    "lowercaseWords.groupBy(\"word\").count().orderBy(col(\"count\").desc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "5. <strong>Get the minimum temperature by station</strong>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"minimumTemperatureByStation\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "minTemperaturesFile = spark.read.csv(\"1800.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------+----+----+----+----+---+----+\n",
      "|        _c0|     _c1| _c2| _c3| _c4| _c5|_c6| _c7|\n",
      "+-----------+--------+----+----+----+----+---+----+\n",
      "|ITE00100554|18000101|TMAX| -75|null|null|  E|null|\n",
      "|ITE00100554|18000101|TMIN|-148|null|null|  E|null|\n",
      "|GM000010962|18000101|PRCP|   0|null|null|  E|null|\n",
      "|EZE00100082|18000101|TMAX| -86|null|null|  E|null|\n",
      "|EZE00100082|18000101|TMIN|-135|null|null|  E|null|\n",
      "|ITE00100554|18000102|TMAX| -60|null|   I|  E|null|\n",
      "|ITE00100554|18000102|TMIN|-125|null|null|  E|null|\n",
      "|GM000010962|18000102|PRCP|   0|null|null|  E|null|\n",
      "|EZE00100082|18000102|TMAX| -44|null|null|  E|null|\n",
      "|EZE00100082|18000102|TMIN|-130|null|null|  E|null|\n",
      "|ITE00100554|18000103|TMAX| -23|null|null|  E|null|\n",
      "|ITE00100554|18000103|TMIN| -46|null|   I|  E|null|\n",
      "|GM000010962|18000103|PRCP|   4|null|null|  E|null|\n",
      "|EZE00100082|18000103|TMAX| -10|null|null|  E|null|\n",
      "|EZE00100082|18000103|TMIN| -73|null|null|  E|null|\n",
      "|ITE00100554|18000104|TMAX|   0|null|null|  E|null|\n",
      "|ITE00100554|18000104|TMIN| -13|null|null|  E|null|\n",
      "|GM000010962|18000104|PRCP|   0|null|null|  E|null|\n",
      "|EZE00100082|18000104|TMAX| -55|null|null|  E|null|\n",
      "|EZE00100082|18000104|TMIN| -74|null|null|  E|null|\n",
      "+-----------+--------+----+----+----+----+---+----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "minTemperaturesFile.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we can see, columns don't have a descriptive name, so we are going to define a schema for all the necessary columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField, StringType, IntegerType, FloatType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = StructType([ \\\n",
    "                     StructField(\"stationID\", StringType(), True), \\\n",
    "                     StructField(\"date\", IntegerType(), True), \\\n",
    "                     StructField(\"measure_type\", StringType(), True), \\\n",
    "                     StructField(\"temperature\", FloatType(), True)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- stationID: string (nullable = true)\n",
      " |-- date: integer (nullable = true)\n",
      " |-- measure_type: string (nullable = true)\n",
      " |-- temperature: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = spark.read.schema(schema).csv(\"1800.csv\")\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------+------------+-----------+\n",
      "|  stationID|    date|measure_type|temperature|\n",
      "+-----------+--------+------------+-----------+\n",
      "|ITE00100554|18000101|        TMAX|      -75.0|\n",
      "|ITE00100554|18000101|        TMIN|     -148.0|\n",
      "|GM000010962|18000101|        PRCP|        0.0|\n",
      "|EZE00100082|18000101|        TMAX|      -86.0|\n",
      "|EZE00100082|18000101|        TMIN|     -135.0|\n",
      "|ITE00100554|18000102|        TMAX|      -60.0|\n",
      "|ITE00100554|18000102|        TMIN|     -125.0|\n",
      "|GM000010962|18000102|        PRCP|        0.0|\n",
      "|EZE00100082|18000102|        TMAX|      -44.0|\n",
      "|EZE00100082|18000102|        TMIN|     -130.0|\n",
      "|ITE00100554|18000103|        TMAX|      -23.0|\n",
      "|ITE00100554|18000103|        TMIN|      -46.0|\n",
      "|GM000010962|18000103|        PRCP|        4.0|\n",
      "|EZE00100082|18000103|        TMAX|      -10.0|\n",
      "|EZE00100082|18000103|        TMIN|      -73.0|\n",
      "|ITE00100554|18000104|        TMAX|        0.0|\n",
      "|ITE00100554|18000104|        TMIN|      -13.0|\n",
      "|GM000010962|18000104|        PRCP|        0.0|\n",
      "|EZE00100082|18000104|        TMAX|      -55.0|\n",
      "|EZE00100082|18000104|        TMIN|      -74.0|\n",
      "+-----------+--------+------------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we will create the solution for the problem"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------+\n",
      "|  stationID|temperature|\n",
      "+-----------+-----------+\n",
      "|ITE00100554|     -148.0|\n",
      "|EZE00100082|     -135.0|\n",
      "|ITE00100554|     -125.0|\n",
      "|EZE00100082|     -130.0|\n",
      "|ITE00100554|      -46.0|\n",
      "|EZE00100082|      -73.0|\n",
      "|ITE00100554|      -13.0|\n",
      "|EZE00100082|      -74.0|\n",
      "|ITE00100554|       -6.0|\n",
      "|EZE00100082|      -58.0|\n",
      "|ITE00100554|       13.0|\n",
      "|EZE00100082|      -57.0|\n",
      "|ITE00100554|       10.0|\n",
      "|EZE00100082|      -50.0|\n",
      "|ITE00100554|       14.0|\n",
      "|EZE00100082|      -31.0|\n",
      "|ITE00100554|       23.0|\n",
      "|EZE00100082|      -46.0|\n",
      "|ITE00100554|       31.0|\n",
      "|EZE00100082|      -75.0|\n",
      "+-----------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df.measure_type==\"TMIN\").select(df.stationID, df.temperature).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------------+\n",
      "|  stationID|min(temperature)|\n",
      "+-----------+----------------+\n",
      "|ITE00100554|          -148.0|\n",
      "|EZE00100082|          -135.0|\n",
      "+-----------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df.measure_type==\"TMIN\").select(df.stationID, df.temperature).groupBy(df.stationID).min(\"temperature\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "6. <strong>Get the total amount spent by customers using dataframes</strong>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"totalSpentCustomers\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField, StringType, IntegerType, FloatType"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We define the inout schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = StructType([ \\\n",
    "                     StructField(\"customerID\", StringType(), True), \\\n",
    "                     StructField(\"orderID\", IntegerType(), True), \\\n",
    "                     StructField(\"amount\", FloatType(), True)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- customerID: string (nullable = true)\n",
      " |-- orderID: integer (nullable = true)\n",
      " |-- amount: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_customers = spark.read.schema(schema).csv(\"customer-orders.csv\")\n",
    "df_customers.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+------+\n",
      "|customerID|orderID|amount|\n",
      "+----------+-------+------+\n",
      "|        44|   8602| 37.19|\n",
      "|        35|   5368| 65.89|\n",
      "|         2|   3391| 40.64|\n",
      "|        47|   6694| 14.98|\n",
      "|        29|    680| 13.08|\n",
      "|        91|   8900| 24.59|\n",
      "|        70|   3959| 68.68|\n",
      "|        85|   1733| 28.53|\n",
      "|        53|   9900| 83.55|\n",
      "|        14|   1505|  4.32|\n",
      "|        51|   3378|  19.8|\n",
      "|        42|   6926| 57.77|\n",
      "|         2|   4424| 55.77|\n",
      "|        79|   9291| 33.17|\n",
      "|        50|   3901| 23.57|\n",
      "|        20|   6633|  6.49|\n",
      "|        15|   6148| 65.53|\n",
      "|        44|   8331| 99.19|\n",
      "|         5|   3505| 64.18|\n",
      "|        48|   5539| 32.42|\n",
      "+----------+-------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_customers.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "They are asking to sum all amounts grouped by customer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+\n",
      "|customerID|total_amount|\n",
      "+----------+------------+\n",
      "|        68|     6375.45|\n",
      "|        73|      6206.2|\n",
      "|        39|     6193.11|\n",
      "|        54|     6065.39|\n",
      "|        71|     5995.66|\n",
      "|         2|     5994.59|\n",
      "|        97|     5977.19|\n",
      "|        46|     5963.11|\n",
      "|        42|     5696.84|\n",
      "|        59|     5642.89|\n",
      "|        41|     5637.62|\n",
      "|         0|     5524.95|\n",
      "|         8|     5517.24|\n",
      "|        85|     5503.43|\n",
      "|        61|     5497.48|\n",
      "|        32|     5496.05|\n",
      "|        58|     5437.73|\n",
      "|        63|     5415.15|\n",
      "|        15|     5413.51|\n",
      "|         6|     5397.88|\n",
      "+----------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import functions as func\n",
    "from pyspark.sql.functions import col, asc,desc\n",
    "df_customers.select(\"customerID\", \"amount\").groupBy(\"customerID\").agg(func.round(func.sum(\"amount\"), 2).alias(\"total_amount\")).orderBy(col(\"total_amount\").desc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "7. <strong>sort all movies by popularity in one line</strong>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"moviePopularity\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We define a schema for the input file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField, IntegerType, LongType\n",
    "schema = StructType([\\\n",
    "                    StructField(\"userID\", IntegerType(), True),\\\n",
    "                    StructField(\"movieID\", IntegerType(), True),\\\n",
    "                    StructField(\"rating\", IntegerType(), True),\\\n",
    "                    StructField(\"timestamp\", LongType(), True)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "moviesDf = spark.read.option(\"sep\", \"\\t\").schema(schema).csv(\"u.data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+---------+\n",
      "|userID|movieID|rating|timestamp|\n",
      "+------+-------+------+---------+\n",
      "|   196|    242|     3|881250949|\n",
      "|   186|    302|     3|891717742|\n",
      "|    22|    377|     1|878887116|\n",
      "|   244|     51|     2|880606923|\n",
      "|   166|    346|     1|886397596|\n",
      "|   298|    474|     4|884182806|\n",
      "|   115|    265|     2|881171488|\n",
      "|   253|    465|     5|891628467|\n",
      "|   305|    451|     3|886324817|\n",
      "|     6|     86|     3|883603013|\n",
      "|    62|    257|     2|879372434|\n",
      "|   286|   1014|     5|879781125|\n",
      "|   200|    222|     5|876042340|\n",
      "|   210|     40|     3|891035994|\n",
      "|   224|     29|     3|888104457|\n",
      "|   303|    785|     3|879485318|\n",
      "|   122|    387|     5|879270459|\n",
      "|   194|    274|     2|879539794|\n",
      "|   291|   1042|     4|874834944|\n",
      "|   234|   1184|     2|892079237|\n",
      "+------+-------+------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "moviesDf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------+\n",
      "|movieID|timesWatched|\n",
      "+-------+------------+\n",
      "|     50|         583|\n",
      "|    258|         509|\n",
      "|    100|         508|\n",
      "|    181|         507|\n",
      "|    294|         485|\n",
      "|    286|         481|\n",
      "|    288|         478|\n",
      "|      1|         452|\n",
      "|    300|         431|\n",
      "|    121|         429|\n",
      "|    174|         420|\n",
      "|    127|         413|\n",
      "|     56|         394|\n",
      "|      7|         392|\n",
      "|     98|         390|\n",
      "|    237|         384|\n",
      "|    117|         378|\n",
      "|    172|         367|\n",
      "|    222|         365|\n",
      "|    204|         350|\n",
      "+-------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "moviesDf.groupBy(\"movieID\").agg(func.count(\"movieID\").alias(\"timesWatched\")).orderBy(col(\"timesWatched\").desc()).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Or even, using with a simple line:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+\n",
      "|movieID|count|\n",
      "+-------+-----+\n",
      "|     50|  583|\n",
      "|    258|  509|\n",
      "|    100|  508|\n",
      "|    181|  507|\n",
      "|    294|  485|\n",
      "|    286|  481|\n",
      "|    288|  478|\n",
      "|      1|  452|\n",
      "|    300|  431|\n",
      "|    121|  429|\n",
      "|    174|  420|\n",
      "|    127|  413|\n",
      "|     56|  394|\n",
      "|      7|  392|\n",
      "|     98|  390|\n",
      "|    237|  384|\n",
      "|    117|  378|\n",
      "|    172|  367|\n",
      "|    222|  365|\n",
      "|    204|  350|\n",
      "+-------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "moviesDf.groupBy(\"movieID\").count().orderBy(func.desc(\"count\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "8. <strong>Get the most popular movies including its names (using u.ITEM file, which contains movie names)</strong>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"PopularMovies\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "movieNames = spark.sparkContext.textFile(\"u.item\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1|Toy Story (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Toy%20Story%20(1995)|0|0|0|1|1|1|0|0|0|0|0|0|0|0|0|0|0|0|0\n",
      "2|GoldenEye (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?GoldenEye%20(1995)|0|1|1|0|0|0|0|0|0|0|0|0|0|0|0|0|1|0|0\n",
      "3|Four Rooms (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Four%20Rooms%20(1995)|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|1|0|0\n",
      "4|Get Shorty (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Get%20Shorty%20(1995)|0|1|0|0|0|1|0|0|1|0|0|0|0|0|0|0|0|0|0\n",
      "5|Copycat (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Copycat%20(1995)|0|0|0|0|0|0|1|0|1|0|0|0|0|0|0|0|1|0|0\n",
      "6|Shanghai Triad (Yao a yao yao dao waipo qiao) (1995)|01-Jan-1995||http://us.imdb.com/Title?Yao+a+yao+yao+dao+waipo+qiao+(1995)|0|0|0|0|0|0|0|0|1|0|0|0|0|0|0|0|0|0|0\n",
      "7|Twelve Monkeys (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Twelve%20Monkeys%20(1995)|0|0|0|0|0|0|0|0|1|0|0|0|0|0|0|1|0|0|0\n",
      "8|Babe (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Babe%20(1995)|0|0|0|0|1|1|0|0|1|0|0|0|0|0|0|0|0|0|0\n",
      "9|Dead Man Walking (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Dead%20Man%20Walking%20(1995)|0|0|0|0|0|0|0|0|1|0|0|0|0|0|0|0|0|0|0\n",
      "10|Richard III (1995)|22-Jan-1996||http://us.imdb.com/M/title-exact?Richard%20III%20(1995)|0|0|0|0|0|0|0|0|1|0|0|0|0|0|0|0|0|1|0\n"
     ]
    }
   ],
   "source": [
    "for i in movieNames.take(10):print(i)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we can see from the RDD shown above, we have to git its first two columns (1.movieID, 2. MovieName). Then, we should join this dataframe with the one of the previous exercise to add \"movieName\" column). We could implement the solution by using one of the next methods:\n",
    "1) broadcasting the movieNames dataframe accross all nodes so as to make every node match each task with its movieName.\n",
    "\n",
    "2) join dataframe from the previuos section with name dataframe"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<strong>Solution 1) By using broadcast</strong>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as func\n",
    "from pyspark.sql.types import StructType, StructField, IntegerType, LongType\n",
    "import codecs\n",
    "from pyspark.sql.functions import col"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will create a dictiones with the next structure {movieID:movieName}. For that purpose, we will create a function to be used later, as a udf function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "import codecs\n",
    "\n",
    "def loadMovieNames():\n",
    "    movieNames = {}\n",
    "    with codecs.open(\"u.item\", \"r\", encoding='ISO-8859-1', errors='ignore') as f:\n",
    "        for line in f:\n",
    "            fields = line.split('|')\n",
    "            movieNames[int(fields[0])] = fields[1]\n",
    "    return movieNames\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "nameDict = spark.sparkContext.broadcast(loadMovieNames())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, wil will read the file containing movies and its voted times:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then, we will define the schema for that file (the same from the exercise above)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = StructType([\\\n",
    "                    StructField(\"userID\", IntegerType(), True), \\\n",
    "                    StructField(\"movieID\", IntegerType(), True), \\\n",
    "                    StructField(\"rating\", IntegerType(), True), \\\n",
    "                    StructField(\"timestamp\", LongType(), True)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "moviesDF = spark.read.option(\"sep\", \"\\t\").schema(schema).csv(\"u.data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+---------+\n",
      "|userID|movieID|rating|timestamp|\n",
      "+------+-------+------+---------+\n",
      "|   196|    242|     3|881250949|\n",
      "|   186|    302|     3|891717742|\n",
      "|    22|    377|     1|878887116|\n",
      "|   244|     51|     2|880606923|\n",
      "|   166|    346|     1|886397596|\n",
      "|   298|    474|     4|884182806|\n",
      "|   115|    265|     2|881171488|\n",
      "|   253|    465|     5|891628467|\n",
      "|   305|    451|     3|886324817|\n",
      "|     6|     86|     3|883603013|\n",
      "|    62|    257|     2|879372434|\n",
      "|   286|   1014|     5|879781125|\n",
      "|   200|    222|     5|876042340|\n",
      "|   210|     40|     3|891035994|\n",
      "|   224|     29|     3|888104457|\n",
      "|   303|    785|     3|879485318|\n",
      "|   122|    387|     5|879270459|\n",
      "|   194|    274|     2|879539794|\n",
      "|   291|   1042|     4|874834944|\n",
      "|   234|   1184|     2|892079237|\n",
      "+------+-------+------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "moviesDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "movieCounts= moviesDF.groupBy(\"movieID\").count().orderBy(func.desc(\"count\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+\n",
      "|movieID|count|\n",
      "+-------+-----+\n",
      "|     50|  583|\n",
      "|    258|  509|\n",
      "|    100|  508|\n",
      "|    181|  507|\n",
      "|    294|  485|\n",
      "|    286|  481|\n",
      "|    288|  478|\n",
      "|      1|  452|\n",
      "|    300|  431|\n",
      "|    121|  429|\n",
      "|    174|  420|\n",
      "|    127|  413|\n",
      "|     56|  394|\n",
      "|      7|  392|\n",
      "|     98|  390|\n",
      "|    237|  384|\n",
      "|    117|  378|\n",
      "|    172|  367|\n",
      "|    222|  365|\n",
      "|    204|  350|\n",
      "+-------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieCounts.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we will create a UDF (user_defined function) to search movie names from the broadcasted dictionary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "def lookupName(movieID):\n",
    "    # the boradcasted dictionary\n",
    "    return nameDict.value[movieID]\n",
    "\n",
    "lookupNameUDF = func.udf(lookupName)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, after having created the UDF, we have to add the column of the movie title to our dataframe (movieCounts)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "moviesWithNames = movieCounts.withColumn(\"movieTitle\", lookupNameUDF(func.col(\"movieID\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+--------------------+\n",
      "|movieID|count|          movieTitle|\n",
      "+-------+-----+--------------------+\n",
      "|     50|  583|    Star Wars (1977)|\n",
      "|    258|  509|      Contact (1997)|\n",
      "|    100|  508|        Fargo (1996)|\n",
      "|    181|  507|Return of the Jed...|\n",
      "|    294|  485|    Liar Liar (1997)|\n",
      "|    286|  481|English Patient, ...|\n",
      "|    288|  478|       Scream (1996)|\n",
      "|      1|  452|    Toy Story (1995)|\n",
      "|    300|  431|Air Force One (1997)|\n",
      "|    121|  429|Independence Day ...|\n",
      "|    174|  420|Raiders of the Lo...|\n",
      "|    127|  413|Godfather, The (1...|\n",
      "|     56|  394| Pulp Fiction (1994)|\n",
      "|      7|  392|Twelve Monkeys (1...|\n",
      "|     98|  390|Silence of the La...|\n",
      "|    237|  384|Jerry Maguire (1996)|\n",
      "|    117|  378|    Rock, The (1996)|\n",
      "|    172|  367|Empire Strikes Ba...|\n",
      "|    222|  365|Star Trek: First ...|\n",
      "|    204|  350|Back to the Futur...|\n",
      "+-------+-----+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "moviesWithNames.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<strong>Solution 2) By using join</strong>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We consider the problem from the beginning, so we read input files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [],
   "source": [
    "moviesDF = spark.read.option(\"sep\", \"\\t\").schema(schema).csv(\"u.data\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We find out the most rated movies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "movieCounts= moviesDF.groupBy(\"movieID\").count().orderBy(func.desc(\"count\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We define the schema of the movie file with names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField, StringType, IntegerType, FloatType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = StructType([\\\n",
    "                    StructField(\"movieID\", IntegerType(), True), \\\n",
    "                    StructField(\"movieName\", StringType(), True)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [],
   "source": [
    "movieNames = spark.read.option(\"sep\", \"|\").schema(schema).csv(\"u.item\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+\n",
      "|movieID|           movieName|\n",
      "+-------+--------------------+\n",
      "|      1|    Toy Story (1995)|\n",
      "|      2|    GoldenEye (1995)|\n",
      "|      3|   Four Rooms (1995)|\n",
      "|      4|   Get Shorty (1995)|\n",
      "|      5|      Copycat (1995)|\n",
      "|      6|Shanghai Triad (Y...|\n",
      "|      7|Twelve Monkeys (1...|\n",
      "|      8|         Babe (1995)|\n",
      "|      9|Dead Man Walking ...|\n",
      "|     10|  Richard III (1995)|\n",
      "|     11|Seven (Se7en) (1995)|\n",
      "|     12|Usual Suspects, T...|\n",
      "|     13|Mighty Aphrodite ...|\n",
      "|     14|  Postino, Il (1994)|\n",
      "|     15|Mr. Holland's Opu...|\n",
      "|     16|French Twist (Gaz...|\n",
      "|     17|From Dusk Till Da...|\n",
      "|     18|White Balloon, Th...|\n",
      "|     19|Antonia's Line (1...|\n",
      "|     20|Angels and Insect...|\n",
      "+-------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieNames.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+\n",
      "|movieID|count|\n",
      "+-------+-----+\n",
      "|    405|  737|\n",
      "|    655|  685|\n",
      "|     13|  636|\n",
      "|    450|  540|\n",
      "|    276|  518|\n",
      "|    416|  493|\n",
      "|    537|  490|\n",
      "|    303|  484|\n",
      "|    234|  480|\n",
      "|    393|  448|\n",
      "|    181|  435|\n",
      "|    279|  434|\n",
      "|    429|  414|\n",
      "|    846|  405|\n",
      "|      7|  403|\n",
      "|     94|  400|\n",
      "|    682|  399|\n",
      "|    308|  397|\n",
      "|    293|  388|\n",
      "|     92|  388|\n",
      "+-------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieCounts.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movieID: integer (nullable = true)\n",
      " |-- count: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieCounts.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [],
   "source": [
    "moviesJoin = movieNames.join(movieCounts, movieNames.movieID == movieCounts.movieID, 'inner').select(movieCounts.movieID,movieNames.movieName, \"count\").orderBy(func.desc(\"count\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+-----+\n",
      "|movieID|           movieName|count|\n",
      "+-------+--------------------+-----+\n",
      "|    405|Mission: Impossib...|  737|\n",
      "|    655|  Stand by Me (1986)|  685|\n",
      "|     13|Mighty Aphrodite ...|  636|\n",
      "|    450|Star Trek V: The ...|  540|\n",
      "|    276|Leaving Las Vegas...|  518|\n",
      "|    416|   Old Yeller (1957)|  493|\n",
      "|    537|My Own Private Id...|  490|\n",
      "|    303|  Ulee's Gold (1997)|  484|\n",
      "|    234|         Jaws (1975)|  480|\n",
      "|    393|Mrs. Doubtfire (1...|  448|\n",
      "|    181|Return of the Jed...|  435|\n",
      "|    279|Once Upon a Time....|  434|\n",
      "|    429|Day the Earth Sto...|  414|\n",
      "|    846|To Gillian on Her...|  405|\n",
      "|      7|Twelve Monkeys (1...|  403|\n",
      "|     94|   Home Alone (1990)|  400|\n",
      "|    682|I Know What You D...|  399|\n",
      "|    308|FairyTale: A True...|  397|\n",
      "|     92| True Romance (1993)|  388|\n",
      "|    293|Donnie Brasco (1997)|  388|\n",
      "+-------+--------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "moviesJoin.show()"
   ]
  }
 ],
 "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.8.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}