collaborative_good_map.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "from pyspark.sql import SQLContext\n",
    "sc._jsc.hadoopConfiguration().set(\"fs.s3a.access.key\", ACCESS_KEY)\n",
    "sc._jsc.hadoopConfiguration().set(\"fs.s3a.secret.key\", SECRET_KEY)\n",
    "sc._jsc.hadoopConfiguration().set(\"fs.s3a.endpoint\", \"s3.amazonaws.com\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# Read Triplets\n",
    "from pyspark.sql.types import IntegerType, StringType, StructField, StructType\n",
    "rddTriplets = sc.textFile(\"file:///home/hadoop/train_triplets.txt\").map(lambda x: x.split(\"\\t\")).map(lambda p: (p[0], p[1], int(p[2])))\n",
    "schemaT = StructType([StructField(\"user_id\", StringType(), True), StructField(\"song_id\", StringType(), True), StructField(\"playcount\", IntegerType(), True)])\n",
    "sqlContext = SQLContext(sc)\n",
    "triplets = sqlContext.createDataFrame(rddTriplets, schema=schemaT)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "from pyspark.sql import functions as F\n",
    "# read mismatches\n",
    "mismatches = (\n",
    "    spark.read.text('file:///home/hadoop/sid_mismatches.txt')\n",
    "    .select(\n",
    "        F.trim(F.col('value').substr(9, 18)).alias('song_id').cast(StringType()),\n",
    "        F.trim(F.col('value').substr(28, 18)).alias('track_id').cast(StringType())\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# remove mismatches song from triplets\n",
    "triplets = (\n",
    "    triplets\n",
    "    .join(\n",
    "        mismatches\n",
    "        .select('song_id'),\n",
    "        on='song_id',\n",
    "        how='left_anti'\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# count unique user_id\n",
    "triplets.select('user_id').dropDuplicates().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# count unique song_id\n",
    "triplets.select('song_id').dropDuplicates().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# most active users\n",
    "active = (\n",
    "    triplets\n",
    "    .groupBy('user_id')\n",
    "    .agg(\n",
    "        F.sum('playcount').alias('playcount'),\n",
    "        F.collect_list('song_id').alias('songs')\n",
    "    )\n",
    "    .orderBy('playcount', ascending=False)\n",
    "    .limit(1)\n",
    "    .rdd.take(1)[0]\n",
    ")\n",
    "active.__getattr__('user_id'), len(active.__getattr__('songs')) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# song popularity\n",
    "song_popularity = (\n",
    "    triplets\n",
    "    .groupBy('song_id')\n",
    "    .agg(\n",
    "        F.sum('playcount').alias('playcount')\n",
    "    )\n",
    ")\n",
    "\n",
    "n = song_popularity.approxQuantile('playcount', [0.0, 0.25, 0.5, 0.75, 1.0], 0.0)\n",
    "  # [1.0, 8.0, 31.0, 130.0, 726885.0]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# user popularity\n",
    "user_activity = (\n",
    "    triplets\n",
    "    .groupBy('user_id')\n",
    "    .agg(\n",
    "        F.count('song_id').alias('songcount')\n",
    "    )\n",
    ")\n",
    "\n",
    "m = user_activity.approxQuantile('songcount', [0.0, 0.25, 0.5, 0.75, 1.0], 0.0)\n",
    "  # [3.0, 15.0, 26.0, 53.0, 4316.0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# inactive songs\n",
    "inactive_song = (\n",
    "    song_popularity\n",
    "    .filter(song_popularity.playcount < n[1])\n",
    "    .select('song_id')\n",
    ")\n",
    "\n",
    "# inactive users\n",
    "inactive_user = (\n",
    "    user_activity\n",
    "    .filter(user_activity.songcount < m[1])\n",
    "    .select('user_id')\n",
    ")\n",
    "\n",
    "# remove inactives from triplets\n",
    "triplets = (\n",
    "    triplets\n",
    "    .join(inactive_song, on='song_id', how='left_anti')\n",
    "    .join(inactive_user, on='user_id', how='left_anti')\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# count\n",
    "triplets.count()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import StringIndexer, QuantileDiscretizer\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.mllib.evaluation import RankingMetrics\n",
    "\n",
    "# ALS Recomm\n",
    "# convert user and song into an integer index\n",
    "indexer_user = StringIndexer(inputCol='user_id', outputCol='user')\n",
    "indexer_song = StringIndexer(inputCol='song_id', outputCol='item')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# pipeline\n",
    "pipeline = Pipeline(stages=[indexer_user, indexer_song]) \n",
    "pipelineModel = pipeline.fit(triplets)\n",
    "dataset = pipelineModel.transform(triplets)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# convert playcount into double (as rating)\n",
    "dataset = (\n",
    "    dataset\n",
    "    .withColumn('user', F.col('user').cast(IntegerType()))\n",
    "    .withColumn('item', F.col('item').cast(IntegerType()))\n",
    "    .withColumn('rating', F.col('playcount').cast(IntegerType()))\n",
    "    .select(['user', 'item', 'rating'])\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# create dictionary of fraction 30% for each user\n",
    "f = (\n",
    "    dataset\n",
    "    .select('user')\n",
    "    .dropDuplicates()\n",
    "    .withColumn('temp', F.lit(0))\n",
    "    .groupBy('temp')\n",
    "    .agg(F.collect_list('user').alias('list'))\n",
    "    .select('list')\n",
    "    .rdd.take(1)[0].__getattr__('list')\n",
    ")\n",
    "\n",
    "fractions = dict(\n",
    "    (user, 0.3) for user in f\n",
    ")\n",
    "\n",
    "# sample test set using fractions\n",
    "test = dataset.sampleBy('user', fractions, seed=1)\n",
    "\n",
    "# get training set by remove test set from full dataset\n",
    "training = (\n",
    "    dataset\n",
    "    .join(\n",
    "        test\n",
    "        .select(['user', 'item']),\n",
    "        on=['user', 'item'],\n",
    "        how='left_anti'\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# select three users\n",
    "test.orderBy('rating', ascending=False).show(10, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# create als model / Set the seed. \n",
    "als = ALS(seed=1)\n",
    "\n",
    "# fit training set and tranform test set\n",
    "model = als.fit(training)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "from pyspark.sql.types import ArrayType\n",
    "u = [525941, 519321, 208949]\n",
    "users = sqlContext.createDataFrame(\n",
    "    [(525941,), (519321,), (208949,)],\n",
    "    StructType([StructField('user', IntegerType())])\n",
    ")\n",
    "\n",
    "# recommend for users\n",
    "recommends = model.recommendForUserSubset(users, 10)\n",
    "\n",
    "# get prediction and labels\n",
    "def recommend(recommendations):\n",
    "    items = []\n",
    "    for item, rating in recommendations:\n",
    "        items.append(item)\n",
    "    return items\n",
    "\n",
    "udf_recommend = F.udf(lambda recommendations: recommend(recommendations), ArrayType(IntegerType()))\n",
    "\n",
    "recommends = (\n",
    "    recommends\n",
    "    .withColumn('recommends', udf_recommend(recommends.recommendations))\n",
    "    .select(\n",
    "        F.col('user').cast(IntegerType()),\n",
    "        F.col('recommends')\n",
    "    )\n",
    ")\n",
    "\n",
    "recommends.show(3, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# get labels\n",
    "ground_truths = (\n",
    "    test\n",
    "    .filter(F.col('user').isin(u))\n",
    "    .orderBy('rating', ascending=False)\n",
    "    .groupBy('user')\n",
    "    .agg(F.collect_list('item').alias('ground_truths'))\n",
    ")\n",
    "\n",
    "ground_truths.show(3, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "compare = recommends.join(ground_truths, on='user', how='left')\n",
    "compare = [(r.__getattr__('recommends'), r.__getattr__('ground_truths')) for r in compare.collect()]\n",
    "compare = sc.parallelize(compare)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "autoscroll": "auto"
   },
   "outputs": [],
   "source": [
    "%spark.pyspark\n",
    "# print metrics\n",
    "metrics = RankingMetrics(compare)\n",
    "print(metrics.precisionAt(10))\n",
    "\n",
    "print(metrics.ndcgAt(10))\n",
    "\n",
    "print(metrics.meanAveragePrecision)\n",
    "\n",
    "\n",
    "# predict test and rmse\n",
    "predict = model.transform(test)\n",
    "predict = predict.filter(F.col('prediction') != float('nan'))\n",
    "reg_eval = RegressionEvaluator(predictionCol='prediction', labelCol='rating', metricName='rmse')\n",
    "reg_eval.evaluate(predict)\n"
   ]
  }
 ],
 "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}