course2-big-data-analysis/c02w05_GraphBasedMusicRecommenderTask4.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sparkSession = SparkSession.builder.enableHiveSupport().master(\"local\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window\n",
    "from pyspark.sql.functions import row_number, sum, col, abs, count, desc, asc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "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 4\n",
    "\n",
    "Build the edges of the type “artist-track”. Take the amount of times the track HAS BEEN listened by all users as the weight of the edge from the artist’s vertex to the track’s vertex. \n",
    "\n",
    "**Tip**: group the dataframe by the columns userId and trackId and use the function “count” of DF API. For each user take top-100 artists and normalize weights.\n",
    "\n",
    "Sort the resulting Data Frame in descending order by the column norm_weight, and then in 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",
    "The part of the result on the sample dataset:\n",
    "\n",
    "    ...\n",
    "    968017 859321\n",
    "    968022 852786\n",
    "    968034 807671\n",
    "    968038 964150\n",
    "    968042 835935\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": 5,
   "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": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "weights = (data\n",
    "           .groupBy(col(\"artistId\"), col(\"trackId\"))\n",
    "           .count().alias(\"count\")\n",
    "          ).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "normalized = (norm(weights, \"artistId\", \"trackId\", \"count\", 100)\n",
    "              .orderBy(col(\"norm_count\").desc(), col(\"artistId\").asc(), col(\"trackId\").desc())\n",
    "              .limit(40)\n",
    "             )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "967993 869415\n",
      "967998 947428\n",
      "968004 927380\n",
      "968017 859321\n",
      "968022 852786\n",
      "968034 807671\n",
      "968038 964150\n",
      "968042 835935\n",
      "968043 913568\n",
      "968046 935077\n",
      "968047 806127\n",
      "968065 907906\n",
      "968073 964586\n",
      "968086 813446\n",
      "968092 837129\n",
      "968118 914441\n",
      "968125 821410\n",
      "968140 953008\n",
      "968148 877445\n",
      "968161 809793\n",
      "968163 803065\n",
      "968168 876119\n",
      "968189 858639\n",
      "968221 896937\n",
      "968224 892880\n",
      "968232 825536\n",
      "968237 932845\n",
      "968238 939177\n",
      "968241 879045\n",
      "968242 911250\n",
      "968248 953554\n",
      "968255 808494\n",
      "968259 880230\n",
      "968265 950148\n",
      "968266 824437\n",
      "968269 913243\n",
      "968272 816049\n",
      "968278 946743\n",
      "968285 847460\n",
      "968286 940006\n"
     ]
    }
   ],
   "source": [
    "results = normalized.select(col(\"artistId\"), col(\"trackId\"))\n",
    "for u, a in results.collect():\n",
    "    print(\"{} {}\".format(u, a))"
   ]
  },
  {
   "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
}