course2-big-data-analysis/c02w05_GraphBasedMusicRecommenderTask1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window\n",
    "from pyspark.sql.functions import row_number, sum, col, abs, count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "sparkSession = SparkSession.builder.enableHiveSupport().master(\"local\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = sparkSession.read.parquet(\"/data/sample264\")\n",
    "meta = sparkSession.read.parquet(\"/data/meta\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Graph based Music Recommender. Task 1\n",
    "\n",
    "Build the edges of the type “track-track”. To do it you will need to count the collaborative similarity between all the tracks: if a user has started listening to track B within 7 minutes after starting track A, then you should add 1 to the weight of the edge from vertex A to vertex B (initial weight is equal to 0).\n",
    "\n",
    "Example:\n",
    "    \n",
    "    userId artistId trackId timestamp\n",
    "    7        12        1          1534574189\n",
    "    7        13        4          1534574289 \n",
    "    5        12        1          1534574389 \n",
    "    5        13        4          1534594189 \n",
    "    6        12        1          1534574489 \n",
    "    6        13        4          1534574689 \n",
    "\n",
    "The track 1 is similar to the track 4 with the weight 2 (before normalization): the user 7 and the user 6 listened these 2 tracks together in the 7 minutes long window:\n",
    "\n",
    "  - userId 7: 1534574289 - 1534574189 = 100 seconds = 1 min 40 seconds < 7 minutes\n",
    "  - userId 6: 1534574689 - 1534574489 = 200 seconds = 3 min 20 seconds < 7 minutes\n",
    "\n",
    "Note that the track 4 is similar to the track 1 with the same weight 2.\n",
    "\n",
    "Tip: consider joining the graph to itself with the UserId and remove pairs with the same tracks.For each track choose top 50 tracks ordered by weight similar to it and normalize weights of its edges (divide the weight of each edge on a sum of weights of all edges). Use rank() to choose top 40 tracks as is done in the demo.\n",
    "\n",
    "Sort the resulting Data Frame in the descending order by the column norm_weight, and then in the ascending order this time first by “id1”, then by “id2”. Take top 40 rows, select only the columns “id1”, “id2”, and print the columns “id1”, “id2” of the resulting dataframe.\n",
    "\n",
    "Output example:\n",
    "\n",
    "    54719\t\t767867\n",
    "    54719\t\t767866\n",
    "    50787\t\t327676\n",
    "\n",
    "----\n",
    "\n",
    "For all subtasks use the same ipython notebook, each subtask should be the continuation of the previous"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def norm(df, key1, key2, field, n): \n",
    "    \n",
    "    window = Window.partitionBy(key1).orderBy(col(field).desc())\n",
    "        \n",
    "    topsDF = df.withColumn(\"row_number\", row_number().over(window)) \\\n",
    "        .filter(col(\"row_number\") <= n) \\\n",
    "        .drop(col(\"row_number\")) \n",
    "        \n",
    "    tmpDF = topsDF.groupBy(col(key1)).agg(col(key1), sum(col(field)).alias(\"sum_\" + field))\n",
    "   \n",
    "    normalizedDF = topsDF.join(tmpDF, key1, \"inner\") \\\n",
    "        .withColumn(\"norm_\" + field, col(field) / col(\"sum_\" + field)) \\\n",
    "        .cache()\n",
    "\n",
    "    return normalizedDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "data1 = data.select(\n",
    "    col('userId').alias('userId'), \n",
    "    col('trackId').alias('trackId1'), \n",
    "    col('timestamp').alias('timestamp1')\n",
    ")\n",
    "\n",
    "data2 = data.select(\n",
    "    col('userId').alias('userId'), \n",
    "    col('trackId').alias('trackId2'), \n",
    "    col('timestamp').alias('timestamp2')\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "similarity_count = (data1.join(data2, \"userId\").cache()\n",
    "                    .filter(col('trackId1') != col('trackId2'))\n",
    "                    .filter(abs(col('timestamp1') - col('timestamp2')) <= 420).cache()\n",
    "                    .groupBy(col('trackId1'), col('trackId2'))\n",
    "                    .count().alias('count')\n",
    "                   ).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "normalized = norm(similarity_count, \"trackId1\", \"trackId2\", \"count\", 40)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = (normalized\n",
    " .orderBy(col(\"norm_count\").desc(), col(\"trackId1\"), col(\"trackId2\"))\n",
    " .limit(40)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = results.select(col(\"trackId1\"), col(\"trackId2\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "798256\t923706\n",
      "798319\t837992\n",
      "798322\t876562\n",
      "798331\t827364\n",
      "798335\t840741\n",
      "798374\t816874\n",
      "798375\t810685\n",
      "798379\t812055\n",
      "798380\t840113\n",
      "798396\t817687\n",
      "798398\t926302\n",
      "798405\t867217\n",
      "798443\t905923\n",
      "798457\t918918\n",
      "798460\t891840\n",
      "798461\t940379\n",
      "798470\t840814\n",
      "798474\t963162\n",
      "798477\t883244\n",
      "798485\t955521\n",
      "798505\t905671\n",
      "798545\t949238\n",
      "798550\t936295\n",
      "798626\t845438\n",
      "798691\t818279\n",
      "798692\t898823\n",
      "798702\t811440\n",
      "798704\t937570\n",
      "798725\t933147\n",
      "798738\t894170\n",
      "798745\t799665\n",
      "798782\t956938\n",
      "798801\t950802\n",
      "798820\t890393\n",
      "798833\t916319\n",
      "798865\t962662\n",
      "798931\t893574\n",
      "798946\t946408\n",
      "799012\t809997\n",
      "799024\t935246\n"
     ]
    }
   ],
   "source": [
    "for t1, t2 in results.collect():\n",
    "    print(\"{}\\t{}\".format(t1,t2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}