6_recommendation_system.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##  Collaborative Filtering Recommendation Systems \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "from pyspark.mllib.linalg import Vectors\n",
    "from pyspark.mllib.linalg import DenseVector\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.feature import StringIndexer, VectorIndexer\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "\n",
    " \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as F"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We use matplotlib for plotting\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "# This statement allow to display plot without asking to\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load data into Spark DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# load play log data \n",
    "df_play = spark.read.csv('../data/play_ds.csv',header=True)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+------+-------+----------+---------+-----------+\n",
      "|      uid|device|song_id|      date|play_time|song_length|\n",
      "+---------+------+-------+----------+---------+-----------+\n",
      "|168535490|    ar|6616004|2017-03-30|      283|        283|\n",
      "|168540348|    ar|  77260|2017-03-30|    64528|          0|\n",
      "|168550572|    ar|4297299|2017-03-30|       11|        272|\n",
      "|168548493|    ip|6661513|2017-03-30|       63|        243|\n",
      "|168550571|    ar|      0|2017-03-30|       24|        251|\n",
      "+---------+------+-------+----------+---------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_play.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10912714"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_play.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Cleaning"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Note: bot users are already removed before the downsampling step"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Clean irregular play time and song_length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# filter play time and songs <=0 \n",
    "# change type to double, so that the approxquntile method can work \n",
    "df_play = df_play \\\n",
    "        .withColumn('play_time',F.col('play_time').cast(LongType())) \\\n",
    "        .filter((F.col('song_length')>0) & (F.col('play_time')>0))  \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Remove records that are possibly misclicks or data error\n",
    "# or song length extremely short  \n",
    "\n",
    "top_length_threshold = df_play.approxQuantile('play_time',[0.93],0.05)\n",
    "\n",
    "bot_length_threshold = df_play.approxQuantile('play_time',[0.01],0.05)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "301.0"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "top_length_threshold[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.0"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "bot_length_threshold[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_play = df_play \\\n",
    "        .filter((F.col('play_time')>bot_length_threshold[0]) & (F.col('play_time')<top_length_threshold[0]))  \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "7353847"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_play.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exclude song id =0, i.e. missing song id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_play = df_play \\\n",
    "        .withColumn('song_id',F.col('song_id').cast(LongType())) \\\n",
    "        .filter(F.col('song_id')>0)\n",
    "\n",
    "\n",
    "          \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6749472"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_play.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+------+--------+----------+---------+-----------+\n",
      "|      uid|device| song_id|      date|play_time|song_length|\n",
      "+---------+------+--------+----------+---------+-----------+\n",
      "|168535490|    ar| 6616004|2017-03-30|      283|        283|\n",
      "|168550572|    ar| 4297299|2017-03-30|       11|        272|\n",
      "|168548493|    ip| 6661513|2017-03-30|       63|        243|\n",
      "|168548714|    ar|23492572|2017-03-30|        2|        359|\n",
      "|168535909|    ar| 7145935|2017-03-30|      130|        130|\n",
      "+---------+------+--------+----------+---------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_play.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exclude inactive users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check how many songs each user has played\n",
    "\n",
    "user_song_count = df_play \\\n",
    "                 .groupBy('uid').agg(F.count(F.col('song_id')).alias('songs_played'))      \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+------------+\n",
      "|      uid|songs_played|\n",
      "+---------+------------+\n",
      "|168550986|           1|\n",
      "|168554377|         138|\n",
      "|168554339|         158|\n",
      "|168559162|         154|\n",
      "|168576985|          11|\n",
      "+---------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_song_count.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "54310"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# count the total number of users\n",
    "user_song_count.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10619"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# count the number of inactive users\n",
    "inactive_user = user_song_count.filter('songs_played < 5')\n",
    "inactive_user.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# exclude inactive users \n",
    "df_play_join = df_play.join(user_song_count,\"uid\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[uid: string, device: string, song_id: bigint, date: string, play_time: bigint, song_length: string, songs_played: bigint]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_play_join"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+------+--------+----------+---------+-----------+------------+\n",
      "|      uid|device| song_id|      date|play_time|song_length|songs_played|\n",
      "+---------+------+--------+----------+---------+-----------+------------+\n",
      "|117677098|    ar| 2216557|2017-04-02|      247|        246|         141|\n",
      "|117677098|    ar| 3392015|2017-04-02|       69|        218|         141|\n",
      "|117677098|    ar| 9845432|2017-04-02|       37|        271|         141|\n",
      "|117677098|    ar|15171705|2017-04-02|       56|        243|         141|\n",
      "|117677098|    ar|15171705|2017-04-02|        2|        243|         141|\n",
      "+---------+------+--------+----------+---------+-----------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_play_join.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_play_active = df_play_join \\\n",
    "                 .filter(F.col('songs_played')>5)            "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6722650"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_play_active.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[uid: string, device: string, song_id: bigint, date: string, play_time: bigint, song_length: string, songs_played: bigint]"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# calculate the most popular songs\n",
    "df_play_active "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Proxy for rating: playtime/song_length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "# generate play percentage as a proxy for rating\n",
    "df_play_active = df_play_active \\\n",
    "        .withColumn(\"percent\",F.col('play_time')/F.col('song_length'))\n",
    "                                        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##  Aggregation by uid and song id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_aggregate = df_play_active \\\n",
    "        .groupBy('uid','song_id').agg(F.mean(F.col('percent')).alias('rating'))      \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+-------------------+\n",
      "|      uid| song_id|             rating|\n",
      "+---------+--------+-------------------+\n",
      "|117677098| 2216557|0.41147423823123913|\n",
      "|117677098| 3392015| 0.6559633027522935|\n",
      "|117677098| 9845432|0.36070110701107017|\n",
      "|117677098|15171705|  0.401920438957476|\n",
      "|117677098|   69873|          0.3640625|\n",
      "+---------+--------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_aggregate.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Transformation\n",
    "\n",
    "ALS requires the song id to be numeric"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_index = df_aggregate \\\n",
    "          .withColumn('uid',F.col('uid').cast(IntegerType())) \\\n",
    "          .withColumn('song_id',F.col('song_id').cast(IntegerType())) \\\n",
    "          .select('uid','song_id',F.round(F.col('rating'), 3).alias('rate'))\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[uid: int, song_id: int, rate: double]"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+-----+\n",
      "|      uid| song_id| rate|\n",
      "+---------+--------+-----+\n",
      "|117677098| 2216557|0.411|\n",
      "|117677098| 3392015|0.656|\n",
      "|117677098| 9845432|0.361|\n",
      "|117677098|15171705|0.402|\n",
      "|117677098|   69873|0.364|\n",
      "+---------+--------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_index.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train Recommender "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "• Collaborative Filtering\n",
    "\n",
    "* Item-Based \n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "(training, test) = df_index.randomSplit([0.7, 0.3])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Build the recommendation model using ALS on the training data\n",
    "# Note we set cold start strategy to 'drop' to ensure we don't get NaN evaluation metrics\n",
    "als = ALS(maxIter=10, regParam=0.01, userCol=\"uid\", itemCol=\"song_id\", ratingCol=\"rate\",\n",
    "          coldStartStrategy=\"drop\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = als.fit(training)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Performance Evaluation \n",
    "• Mask Data Validation \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Root-mean-square error = 0.5673382119470404\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "\n",
    "\n",
    "# Evaluate the model by computing the RMSE on the test data\n",
    "predictions = model.transform(test)\n",
    "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rate\",\n",
    "                                predictionCol=\"prediction\")\n",
    "rmse = evaluator.evaluate(predictions)\n",
    "print(\"Root-mean-square error = \" + str(rmse))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Performance benchmark:\n",
    "for 5 level rating: RMSE~1.2-1.5, or <2 \n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Make recommendations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# Generate top 10 songs recommendations for each user\n",
    "userRecs = model.recommendForAllUsers(10)\n",
    "# Generate top 10 user recommendations for each song\n",
    "songRecs = model.recommendForAllItems(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+\n",
      "|      uid|     recommendations|\n",
      "+---------+--------------------+\n",
      "| 35556737|[[6762450, 103.46...|\n",
      "| 72854913|[[6762450, 70.460...|\n",
      "|149954144|[[6762450, 64.763...|\n",
      "|165428768|[[6762450, 75.927...|\n",
      "|167415658|[[6762450, 54.702...|\n",
      "|167570315|[[6762450, 125.99...|\n",
      "|167575737|[[6762450, 121.32...|\n",
      "|167579413|[[3288014, 52.417...|\n",
      "|167580826|[[6762450, 41.008...|\n",
      "|167584025|[[6762450, 130.17...|\n",
      "|167587488|[[6762450, 105.33...|\n",
      "|167588969|[[6762450, 70.224...|\n",
      "|167590565|[[6762450, 154.22...|\n",
      "|167595909|[[3288014, 63.964...|\n",
      "|167597788|[[3288065, 121.96...|\n",
      "|167605713|[[3288065, 60.772...|\n",
      "|167608347|[[6762450, 70.228...|\n",
      "|167627248|[[6762450, 104.45...|\n",
      "|167645953|[[6762450, 72.958...|\n",
      "|167655696|[[6762450, 52.226...|\n",
      "+---------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "userRecs.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "songRecs.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "• Matrix Factorization \n",
    "\t* truncated SVD\n",
    "\t\n",
    "\t• GraphLab \n",
    "\t• Spark ALS (DataBricks, Google(credit), AWS(small $),IBM(free)) \n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.decomposition import TruncatedSVD\n",
    "\n",
    "def fit_uvd(M,k):\n",
    "    # use TruncatedSVD to realize UVD\n",
    "    svd = TruncatedSVD(n_components=k, n_iter=7, random_state=0)\n",
    "    svd.fit(M)\n",
    "\n",
    "    V = svd.components_\n",
    "    U = svd.transform(M) # effectively, it's doing: U = M.dot(V.T)\n",
    "    # we can ignore svd.singular_values_ for our purpose\n",
    "    \n",
    "    # why we can do this?\n",
    "    # recall: \n",
    "    # SVD start from u*s*v=M => u*s=M*v.T, where M*v.T is our transformation above to get U in UVD\n",
    "    # so the above U is effectively u*s in SVD\n",
    "    # that's why U*V = u*s*v = M our original matrix\n",
    "    # there are many ways to understand it!\n",
    "    # here we by-passed singular values.\n",
    "    \n",
    "    return U,V, svd\n",
    "\n",
    "# decompose\n",
    "\n"
   ]
  },
  {
   "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.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}