Collab Filtering Netflix.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'C:\\\\spark-3.2.1-bin-hadoop3.2'"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "findspark.find()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import os\n",
    "# os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
    "# os.environ[\"SPARK_HOME\"] = \"spark-3.2.1-bin-hadoop3.2\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkContext, SparkConf\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = pyspark.SparkConf().setAppName('Netflix').setMaster('local')\n",
    "sc = pyspark.SparkContext(conf=conf)\n",
    "spark = SparkSession(sc)\n",
    "\n",
    "# sc = pyspark.SparkContext(appName=\"Netflix\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "combined_data_*.txt is a file that contains:\n",
    "movie_id:\n",
    "user_id, rating, date\n",
    "\n",
    "c_combined_data_*.csv is a file that transforms combined_data_*.txt into\n",
    "movie_id, user_id, rating, date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Combined Data 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_schema = StructType([\n",
    "    StructField('ID Film',IntegerType(), False),\n",
    "    StructField('ID User',IntegerType(), False),\n",
    "    StructField('Rating',IntegerType(), False),\n",
    "    StructField('Tanggal',DateType(), False)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_1 = spark.read.csv(\n",
    "    'Cleaned Input/dataset/c_combined_data_1.csv', \n",
    "    header=True, \n",
    "    schema=data_schema\n",
    ").cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+------+----------+\n",
      "|ID Film|ID User|Rating|   Tanggal|\n",
      "+-------+-------+------+----------+\n",
      "|      1| 822109|     5|2005-05-13|\n",
      "|      1| 885013|     4|2005-10-19|\n",
      "|      1|  30878|     4|2005-12-26|\n",
      "|      1| 823519|     3|2004-05-03|\n",
      "|      1| 893988|     3|2005-11-17|\n",
      "|      1| 124105|     4|2004-08-05|\n",
      "|      1|1248029|     3|2004-04-22|\n",
      "|      1|1842128|     4|2004-05-09|\n",
      "|      1|2238063|     3|2005-05-11|\n",
      "|      1|1503895|     4|2005-05-19|\n",
      "|      1|2207774|     5|2005-06-06|\n",
      "|      1|2590061|     3|2004-08-12|\n",
      "|      1|   2442|     3|2004-04-14|\n",
      "|      1| 543865|     4|2004-05-28|\n",
      "|      1|1209119|     4|2004-03-23|\n",
      "|      1| 804919|     4|2004-06-10|\n",
      "|      1|1086807|     3|2004-12-28|\n",
      "|      1|1711859|     4|2005-05-08|\n",
      "|      1| 372233|     5|2005-11-23|\n",
      "|      1|1080361|     3|2005-03-28|\n",
      "+-------+-------+------+----------+\n",
      "only showing top 20 rows\n",
      "\n",
      "Number of Rows 24053523\n"
     ]
    }
   ],
   "source": [
    "df_1.show()\n",
    "row = df_1.count()\n",
    "print(f'Number of Rows {row}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Combined Data 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+------+----------+\n",
      "|ID Film|ID User|Rating|   Tanggal|\n",
      "+-------+-------+------+----------+\n",
      "|   4500| 573364|     3|2005-06-20|\n",
      "|   4500|1696725|     3|2004-02-27|\n",
      "|   4500|1253431|     3|2004-03-31|\n",
      "|   4500|1265574|     2|2003-09-01|\n",
      "|   4500|1049643|     1|2003-11-15|\n",
      "|   4500|1601348|     4|2005-04-05|\n",
      "|   4500|1495289|     5|2005-07-09|\n",
      "|   4500|1254903|     3|2003-09-02|\n",
      "|   4500|2604070|     3|2005-05-15|\n",
      "|   4500|1006473|     5|2005-05-23|\n",
      "|   4500|1989892|     3|2004-04-06|\n",
      "|   4500|1517471|     4|2003-12-24|\n",
      "|   4500|1478381|     4|2005-05-21|\n",
      "|   4500| 923084|     2|2004-11-15|\n",
      "|   4500|2446292|     4|2005-10-06|\n",
      "|   4500|2554745|     3|2003-05-07|\n",
      "|   4500|1133125|     5|2004-08-10|\n",
      "|   4500| 349528|     4|2003-08-11|\n",
      "|   4500|1614895|     5|2004-08-29|\n",
      "|   4500| 424958|     4|2005-08-02|\n",
      "+-------+-------+------+----------+\n",
      "only showing top 20 rows\n",
      "\n",
      "Number of Rows 26977590\n"
     ]
    }
   ],
   "source": [
    "df_2 = spark.read.csv(\n",
    "    'Cleaned Input/dataset/c_combined_data_2.csv', \n",
    "    header=True, \n",
    "    schema=data_schema\n",
    ").cache()\n",
    "\n",
    "df_2.show()\n",
    "row = df_2.count()\n",
    "print(f'Number of Rows {row}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Combined Data 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+------+----------+\n",
      "|ID Film|ID User|Rating|   Tanggal|\n",
      "+-------+-------+------+----------+\n",
      "|   9211|2435457|     2|2005-06-01|\n",
      "|   9211|2338545|     3|2001-02-17|\n",
      "|   9211|2218269|     1|2002-12-27|\n",
      "|   9211| 441153|     4|2002-10-11|\n",
      "|   9211|1921624|     2|2005-08-31|\n",
      "|   9211|2096652|     3|2004-05-31|\n",
      "|   9211| 818736|     2|2004-02-17|\n",
      "|   9211| 284560|     3|2003-07-27|\n",
      "|   9211|1211224|     5|2004-05-08|\n",
      "|   9211|1984086|     1|2004-09-16|\n",
      "|   9211|1389539|     3|2005-06-07|\n",
      "|   9211| 454575|     2|2004-10-23|\n",
      "|   9211| 149028|     3|2003-06-19|\n",
      "|   9211|1105843|     2|2005-04-07|\n",
      "|   9211| 353813|     2|2004-08-05|\n",
      "|   9211| 903779|     3|2001-05-21|\n",
      "|   9211| 639194|     2|2003-10-19|\n",
      "|   9211| 308031|     4|2000-09-07|\n",
      "|   9211|1794933|     1|2003-02-14|\n",
      "|   9211| 625085|     3|2004-05-21|\n",
      "+-------+-------+------+----------+\n",
      "only showing top 20 rows\n",
      "\n",
      "Number of Rows 22601628\n"
     ]
    }
   ],
   "source": [
    "df_3 = spark.read.csv(\n",
    "    'Cleaned Input/dataset/c_combined_data_3.csv', \n",
    "    header=True, \n",
    "    schema=data_schema\n",
    ").cache()\n",
    "\n",
    "df_3.show()\n",
    "row = df_3.count()\n",
    "print(f'Number of Rows {row}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Combined Data 4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+------+----------+\n",
      "|ID Film|ID User|Rating|   Tanggal|\n",
      "+-------+-------+------+----------+\n",
      "|  13368| 659432|     3|2005-03-16|\n",
      "|  13368| 751812|     2|2002-12-16|\n",
      "|  13368|2625420|     2|2004-05-25|\n",
      "|  13368|1650301|     1|2005-08-30|\n",
      "|  13368|2269227|     4|2005-10-27|\n",
      "|  13368|2220672|     4|2002-08-19|\n",
      "|  13368|2500511|     4|2003-08-11|\n",
      "|  13368|1452058|     2|2005-01-29|\n",
      "|  13368|1624891|     3|2002-07-27|\n",
      "|  13368| 970031|     3|2004-04-14|\n",
      "|  13368| 345673|     4|2005-04-07|\n",
      "|  13368|1426869|     5|2005-04-21|\n",
      "|  13368|1037088|     2|2005-09-14|\n",
      "|  13368|2079559|     5|2005-10-08|\n",
      "|  13368|2175560|     5|2005-01-12|\n",
      "|  13368| 946805|     4|2005-02-09|\n",
      "|  13368| 767843|     1|2005-05-04|\n",
      "|  13368| 464031|     3|2001-02-08|\n",
      "|  13368|2473764|     4|2005-09-26|\n",
      "|  13368|2339119|     3|2005-11-27|\n",
      "+-------+-------+------+----------+\n",
      "only showing top 20 rows\n",
      "\n",
      "Number of Rows 26847421\n"
     ]
    }
   ],
   "source": [
    "df_4 = spark.read.csv(\n",
    "    'Cleaned Input/dataset/c_combined_data_4.csv', \n",
    "    header=True, \n",
    "    schema=data_schema\n",
    ").cache()\n",
    "\n",
    "df_4.show()\n",
    "row = df_4.count()\n",
    "print(f'Number of Rows {row}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_samples = df_1.union(df_2.union(df_3.union(df_4)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+------+----------+\n",
      "|ID Film|ID User|Rating|   Tanggal|\n",
      "+-------+-------+------+----------+\n",
      "|      1| 822109|     5|2005-05-13|\n",
      "|      1| 885013|     4|2005-10-19|\n",
      "|      1|  30878|     4|2005-12-26|\n",
      "|      1| 823519|     3|2004-05-03|\n",
      "|      1| 893988|     3|2005-11-17|\n",
      "|      1| 124105|     4|2004-08-05|\n",
      "|      1|1248029|     3|2004-04-22|\n",
      "|      1|1842128|     4|2004-05-09|\n",
      "|      1|2238063|     3|2005-05-11|\n",
      "|      1|1503895|     4|2005-05-19|\n",
      "|      1|2207774|     5|2005-06-06|\n",
      "|      1|2590061|     3|2004-08-12|\n",
      "|      1|   2442|     3|2004-04-14|\n",
      "|      1| 543865|     4|2004-05-28|\n",
      "|      1|1209119|     4|2004-03-23|\n",
      "|      1| 804919|     4|2004-06-10|\n",
      "|      1|1086807|     3|2004-12-28|\n",
      "|      1|1711859|     4|2005-05-08|\n",
      "|      1| 372233|     5|2005-11-23|\n",
      "|      1|1080361|     3|2005-03-28|\n",
      "+-------+-------+------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "all_samples.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "100480162"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "all_samples.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Cek data apakah ada yang kosong"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+------------------+------------------+\n",
      "|summary|          ID Film|           ID User|            Rating|\n",
      "+-------+-----------------+------------------+------------------+\n",
      "|  count|        100480162|         100480162|         100480162|\n",
      "|   mean| 9070.91703013974|1322488.5633934387|3.6042933131417523|\n",
      "| stddev|5131.887114081715| 764536.7800330574|1.0852168296442537|\n",
      "|    min|                1|                 6|                 1|\n",
      "|    25%|             4671|            661149|                 3|\n",
      "|    50%|             9051|           1318951|                 4|\n",
      "|    75%|            13636|           1984358|                 4|\n",
      "|    max|            17770|           2649429|                 5|\n",
      "+-------+-----------------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "all_samples[['ID Film', 'ID User', 'Rating', 'Tanggal']].summary().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Tidak ada data yang kosong\n",
    "#### Jumlah data ada 100.480.162 \n",
    "#### Data diambil dari 1999-11-11 hingga 2005-12-31"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import year\n",
    "added_column = all_samples.withColumn('year', year(all_samples['Tanggal']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "year_list = range(1999,2006)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_count = []\n",
    "for y in year_list:\n",
    "    user_count.append(added_column.select('ID User').where(added_column.year == y).count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[2178, 924443, 1769031, 4342871, 9985324, 30206362, 53249953]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Split Data Training dan Testing \n",
    "### Tahun 1999 - 2004 Training, 2005 Testing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "training_dataset = added_column.select('ID User','ID Film','Rating').where(added_column.year <= 2004)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "testing_dataset = added_column.select('ID User','ID Film','Rating').where(added_column.year == 2005)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "47230209"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "training_dataset.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "53249953"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "testing_dataset.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.recommendation import ALS, ALSModel\n",
    "from pyspark.mllib.evaluation import RegressionMetrics, RankingMetrics\n",
    "from pyspark.ml.evaluation import RegressionEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "als = ALS(maxIter=10, regParam=0.01, \n",
    "          userCol=\"ID User\", itemCol=\"ID Film\", ratingCol=\"Rating\",\n",
    "          coldStartStrategy=\"drop\",\n",
    "          implicitPrefs=False)\n",
    "model = als.fit(training_dataset)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Root-mean-square error = 1.0958761595388327\n"
     ]
    }
   ],
   "source": [
    "# Evaluate the model by computing the RMSE on the test data\n",
    "predictions = model.transform(testing_dataset)\n",
    "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"Rating\",\n",
    "                                predictionCol=\"prediction\")\n",
    "\n",
    "rmse = evaluator.evaluate(predictions)\n",
    "print(\"Root-mean-square error = \" + str(rmse))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Experiment\n",
    "### Dataset\n",
    "### dataset perbandingan training testing adalah 47:53\n",
    "### yaapa lek tahun 2005 nya di bagi 50:50 untuk testing, jadi utk testing hanya 1/2 dari data tahun 2005"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Fine Tuning ALS Model\n",
    "#### maxIter = 5,10,15,20,25,30"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}