2. Spark DF Popular Movie.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as f\n",
    "import collections\n",
    "import codecs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"DF-Popular-Movie\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Broadcast Variables\n",
    "\n",
    "Broadcast Variables broadcast ojects to the executors, such that they are always there whenever needed. Use sc.broadcast() to ship off whatever you want and value() to get the object back. Use the broadcasted object however you want to map functions and UDFs.    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def loadMovieNames():\n",
    "    movieNames = {}\n",
    "    with codecs.open('C:/Users/dennis/Spark/Spark/resources/u.item',\"r\",encoding='ISO-8859-1',errors='ignore') as file:\n",
    "        for line in file:\n",
    "            fields = line.split('|')\n",
    "            movieNames[int(fields[0])]=fields[1]\n",
    "    return movieNames            "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "nameDict = spark.sparkContext.broadcast(loadMovieNames())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load and Verify Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create schema when reading u.data\n",
    "from pyspark.sql.types import StringType, DoubleType, IntegerType, StructType, StructField\n",
    "schema = StructType([StructField('userId', IntegerType(), True),\n",
    "                     StructField('movieId', IntegerType(), True),\n",
    "                     StructField('rating', IntegerType(), True),\n",
    "                     StructField('timestamp', DoubleType(), True)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = spark.read.csv('resources/u.data',sep = '\\t', header = False, schema = schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userId: integer (nullable = true)\n",
      " |-- movieId: integer (nullable = true)\n",
      " |-- rating: integer (nullable = true)\n",
      " |-- timestamp: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(userId=196, movieId=242, rating=3, timestamp=881250949.0),\n",
       " Row(userId=186, movieId=302, rating=3, timestamp=891717742.0),\n",
       " Row(userId=22, movieId=377, rating=1, timestamp=878887116.0)]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data.head(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['userId', 'movieId', 'rating', 'timestamp']"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+------------------+------------------+-----------------+\n",
      "|summary|            userId|           movieId|            rating|        timestamp|\n",
      "+-------+------------------+------------------+------------------+-----------------+\n",
      "|  count|            100000|            100000|            100000|           100000|\n",
      "|   mean|         462.48475|         425.53013|           3.52986|8.8352885148862E8|\n",
      "| stddev|266.61442012750905|330.79835632558473|1.1256735991443214|5343856.189502848|\n",
      "|    min|                 1|                 1|                 1|      8.7472471E8|\n",
      "|    max|               943|              1682|                 5|     8.93286638E8|\n",
      "+-------+------------------+------------------+------------------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create UDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a user-defined functions to look up movie names from our broadcasted dictionary \n",
    "def lookupName(movieID):\n",
    "    return nameDict.value[movieID]\n",
    "lookupNameUDF = f.udf(lookupName)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Display most popular movies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+\n",
      "|movieId|count|\n",
      "+-------+-----+\n",
      "|    496|  231|\n",
      "|    471|  221|\n",
      "|    463|   71|\n",
      "+-------+-----+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieCounts = data.groupBy('movieId').count()\n",
    "movieCounts.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+--------------------+\n",
      "|movieId|count|          movieTitle|\n",
      "+-------+-----+--------------------+\n",
      "|    496|  231|It's a Wonderful ...|\n",
      "|    471|  221|Courage Under Fir...|\n",
      "|    463|   71|Secret of Roan In...|\n",
      "+-------+-----+--------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieWithNames = movieCounts.withColumn('movieTitle',lookupNameUDF(f.col(\"movieId\")))\n",
    "movieWithNames.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------------------+-----+\n",
      "|movieId|movieTitle                   |count|\n",
      "+-------+-----------------------------+-----+\n",
      "|50     |Star Wars (1977)             |583  |\n",
      "|258    |Contact (1997)               |509  |\n",
      "|100    |Fargo (1996)                 |508  |\n",
      "|181    |Return of the Jedi (1983)    |507  |\n",
      "|294    |Liar Liar (1997)             |485  |\n",
      "|286    |English Patient, The (1996)  |481  |\n",
      "|288    |Scream (1996)                |478  |\n",
      "|1      |Toy Story (1995)             |452  |\n",
      "|300    |Air Force One (1997)         |431  |\n",
      "|121    |Independence Day (ID4) (1996)|429  |\n",
      "+-------+-----------------------------+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieWithNames.select('movieId','movieTitle','count').orderBy(f.desc(\"count\")).show(10,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.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.8.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}