Week5-Graph-based-music-recommender-task-5.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "#execfile(os.path.join(os.environ[\"SPARK_HOME\"], 'python/pyspark/shell.py'))"
   ]
  },
  {
   "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": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userId: integer (nullable = true)\n",
      " |-- trackId: integer (nullable = true)\n",
      " |-- artistId: integer (nullable = true)\n",
      " |-- timestamp: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "playHistoryPath = '/data/sample264'\n",
    "playHistoryGraph = sparkSession.read.parquet(playHistoryPath)\n",
    "playHistoryGraph.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- type: string (nullable = true)\n",
      " |-- Name: string (nullable = true)\n",
      " |-- Artist: string (nullable = true)\n",
      " |-- Id: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "metaDataPath = '/data/meta'\n",
    "metaDataGraph = sparkSession.read.parquet(metaDataPath)\n",
    "metaDataGraph.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "playHistoryGraph.createTempView(\"history1\")\n",
    "playHistoryGraph.createTempView(\"history2\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Task 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import count, col\n",
    "\n",
    "consecutiveTracksForUser = sparkSession.sql( \\\n",
    "    \"select h1.trackId as track1, h2.trackId as track2, h1.userId as user \" \\\n",
    "    \"from history1 h1, history2 h2 \" \\\n",
    "    \"where h1.userId = h2.userId \" \\\n",
    "    \"and h1.trackId != h2.trackId \" \\\n",
    "    \"and abs(h2.timestamp - h1.timestamp) <= 420 \" \\\n",
    ").groupBy(col(\"track1\"), col(\"track2\")) \\\n",
    ".count().alias(\"count\") \\\n",
    ".orderBy(col(\"track1\"), col(\"track2\")) \\\n",
    ".cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#consecutiveTracksForUser.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window\n",
    "from pyspark.sql.functions import col, row_number, sum\n",
    "\n",
    "window = Window.partitionBy(\"track1\").orderBy(col(\"count\").desc())\n",
    "       \n",
    "topsDF = consecutiveTracksForUser.withColumn(\"row_number\", row_number().over(window)) \\\n",
    "        .filter(col(\"row_number\") <= 40) \\\n",
    "        .drop(col(\"row_number\")) \\\n",
    "        .orderBy(col(\"track1\"), col(\"track2\")) \\\n",
    "        .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#topsDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "sumsDF = topsDF.groupBy(col(\"track1\")) \\\n",
    ".agg(sum(col(\"count\")).alias(\"sum_weights\")) \\\n",
    ".orderBy(\"track1\") \\\n",
    ".cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#sumsDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "normalized_count = topsDF.join(sumsDF, \"track1\", \"inner\") \\\n",
    "    .withColumn(\"weight\", col(\"count\") / col(\"sum_weights\")) \\\n",
    "    .cache()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#normalized_count.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = normalized_count.orderBy(col(\"weight\").desc(), col(\"track1\"), col(\"track2\")).limit(40)\n",
    "#results.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = results.select(col(\"track1\"), col(\"track2\"))\n",
    "for t1, t2 in results.collect():\n",
    "    print(\"{}\\t{}\".format(t1,t2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Task 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import count, col\n",
    "\n",
    "tracksPerUser = sparkSession.sql( \\\n",
    "    \"select userId as user, trackId as track \" \\\n",
    "    \"from history1 \"\n",
    ").groupBy(col(\"user\"), col(\"track\")) \\\n",
    ".count().alias(\"count\") \\\n",
    ".orderBy(col(\"count\").desc(), col(\"user\"), col(\"track\")) \\\n",
    ".cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tracksPerUser.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window\n",
    "from pyspark.sql.functions import col, row_number, sum\n",
    "\n",
    "window = Window.partitionBy(\"user\").orderBy(col(\"count\").desc())\n",
    "       \n",
    "topsTracksPerUser = tracksPerUser.withColumn(\"row_number\", row_number().over(window)) \\\n",
    "        .filter(col(\"row_number\") <= 1000) \\\n",
    "        .drop(col(\"row_number\")) \\\n",
    "        .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sumsTopsTracksPerUser = topsTracksPerUser.groupBy(col(\"user\")) \\\n",
    ".agg(sum(col(\"count\")).alias(\"sum_weights\")) \\\n",
    ".orderBy(\"user\") \\\n",
    ".cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "normalized_topsTracksPerUser = topsTracksPerUser.join(sumsTopsTracksPerUser, \"user\", \"inner\") \\\n",
    "    .withColumn(\"norm_weight\", col(\"count\") / col(\"sum_weights\")) \\\n",
    "    .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = normalized_topsTracksPerUser.orderBy(col(\"norm_weight\").desc(), col(\"user\"), col(\"track\")) \\\n",
    "    .limit(40) \\\n",
    "    .select(col(\"user\"),col(\"track\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for u, t in results.collect():\n",
    "    print(\"{} {}\".format(u, t))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Task 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import count, col\n",
    "\n",
    "artistsPerUser = sparkSession.sql( \\\n",
    "    \"select userId as user, artistId as artist \" \\\n",
    "    \"from history1 \"\n",
    ").groupBy(col(\"user\"), col(\"artist\")) \\\n",
    ".count().alias(\"count\") \\\n",
    ".orderBy(col(\"count\").desc(), col(\"user\"), col(\"artist\")) \\\n",
    ".cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "artistsPerUser.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window\n",
    "from pyspark.sql.functions import col, row_number, sum\n",
    "\n",
    "window = Window.partitionBy(\"user\").orderBy(col(\"count\").desc())\n",
    "       \n",
    "topsArtistsPerUser = artistsPerUser.withColumn(\"row_number\", row_number().over(window)) \\\n",
    "        .filter(col(\"row_number\") <= 100) \\\n",
    "        .drop(col(\"row_number\")) \\\n",
    "        .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sumsTopsArtistsPerUser = topsArtistsPerUser.groupBy(col(\"user\")) \\\n",
    ".agg(sum(col(\"count\")).alias(\"sum_weights\")) \\\n",
    ".orderBy(\"user\") \\\n",
    ".cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "normalized_topsArtistsPerUser = topsArtistsPerUser.join(sumsTopsArtistsPerUser, \"user\", \"inner\") \\\n",
    "    .withColumn(\"norm_weight\", col(\"count\") / col(\"sum_weights\")) \\\n",
    "    .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = normalized_topsArtistsPerUser.orderBy(col(\"norm_weight\").desc(), col(\"user\"), col(\"artist\")) \\\n",
    "    .limit(40) \\\n",
    "    .select(col(\"user\"),col(\"artist\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for u, t in results.collect():\n",
    "    print(\"{} {}\".format(u, t))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Task 4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import count, col\n",
    "\n",
    "tracksPerArtist = sparkSession.sql( \\\n",
    "    \"select artistId as artist, trackId as track \" \\\n",
    "    \"from history1 \"\n",
    ").groupBy(col(\"artist\"), col(\"track\")) \\\n",
    ".count().alias(\"count\") \\\n",
    ".orderBy(col(\"count\").desc(), col(\"artist\"), col(\"track\")) \\\n",
    ".cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tracksPerArtist.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window\n",
    "from pyspark.sql.functions import col, row_number, sum\n",
    "\n",
    "window = Window.partitionBy(\"artist\").orderBy(col(\"count\").desc())\n",
    "       \n",
    "topsTracksPerArtist = tracksPerArtist.withColumn(\"row_number\", row_number().over(window)) \\\n",
    "        .filter(col(\"row_number\") <= 100) \\\n",
    "        .drop(col(\"row_number\")) \\\n",
    "        .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sumsTracksPerArtist = topsTracksPerArtist.groupBy(col(\"artist\")) \\\n",
    ".agg(sum(col(\"count\")).alias(\"sum_weights\")) \\\n",
    ".orderBy(\"artist\") \\\n",
    ".cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "normalized_topsTracksPerArtist = topsTracksPerArtist.join(sumsTracksPerArtist, \"artist\", \"inner\") \\\n",
    "    .withColumn(\"norm_weight\", col(\"count\") / col(\"sum_weights\")) \\\n",
    "    .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = normalized_topsTracksPerArtist.orderBy(col(\"norm_weight\").desc(), col(\"artist\"), col(\"track\")) \\\n",
    "    .limit(40) \\\n",
    "    .select(col(\"artist\"),col(\"track\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for u, t in results.collect():\n",
    "    print(\"{} {}\".format(u, t))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Task 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col\n",
    "\n",
    "user = playHistoryGraph.filter(\"userId == 776748\").cache()\n",
    "\n",
    "tracks = user.select(col(\"trackId\").alias(\"Id\")).distinct()\n",
    "artists = user.select(col(\"artistId\").alias(\"Id\")).distinct()\n",
    "\n",
    "result = tracks.union(artists) \\\n",
    "    .join(metaDataGraph, on=\"Id\") \\\n",
    "    .orderBy(col(\"Artist\"), col(\"Name\")) \\\n",
    "    .select (col(\"Artist\"), col(\"Name\")) \\\n",
    "    .limit(40) \\\n",
    "    .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Artist: 3 Doors Down Artist: 3 Doors Down\n",
      "Artist: 3 Doors Down Kryptonite\n",
      "Artist: 311 Artist: 311\n",
      "Artist: 311 Beautiful disaster\n",
      "Artist: Blur Artist: Blur\n",
      "Artist: Blur Girls and Boys\n",
      "Artist: Clawfinger Artist: Clawfinger\n",
      "Artist: Clawfinger Nothing Going On\n",
      "Artist: Disturbed Artist: Disturbed\n",
      "Artist: Disturbed The Vengeful One\n",
      "Artist: Gotthard Artist: Gotthard\n",
      "Artist: Gotthard Eagle\n",
      "Artist: Green Day 21 Guns\n",
      "Artist: Green Day Artist: Green Day\n",
      "Artist: Green Day Kill The DJ\n",
      "Artist: Iggy Pop Artist: Iggy Pop\n",
      "Artist: Iggy Pop Sunday\n",
      "Artist: Korn Artist: Korn\n",
      "Artist: Korn Here To Stay\n",
      "Artist: Linkin Park Artist: Linkin Park\n",
      "Artist: Linkin Park In The End\n",
      "Artist: Linkin Park Numb\n",
      "Artist: Lordi Artist: Lordi\n",
      "Artist: Lordi Hard Rock Hallelujah\n",
      "Artist: Nickelback Artist: Nickelback\n",
      "Artist: Nickelback She Keeps Me Up\n",
      "Artist: Nomy Artist: Nomy\n",
      "Artist: Nomy Cocaine\n",
      "Artist: Papa Roach Artist: Papa Roach\n",
      "Artist: Papa Roach Getting Away With Murder\n",
      "Artist: Rise Against Artist: Rise Against\n",
      "Artist: Rise Against Prayer Of The Refugee\n",
      "Artist: Serj Tankian Artist: Serj Tankian\n",
      "Artist: Serj Tankian Sky is Over\n",
      "Artist: Slipknot Artist: Slipknot\n",
      "Artist: Slipknot Wait And Bleed\n",
      "Artist: The Offspring Artist: The Offspring\n",
      "Artist: The Offspring Come Out and Play\n",
      "Artist: Thousand Foot Krutch Artist: Thousand Foot Krutch\n",
      "Artist: Thousand Foot Krutch Take It Out On Me\n"
     ]
    }
   ],
   "source": [
    "for a, n in result.collect():\n",
    "    print(\"{} {}\".format(a,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.6.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}