course2-big-data-analysis/c02w05_GraphBasedMusicRecommenderTask6.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Graph based Music Recommender. Task 6"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For the user with Id 776748 print top-40 recommended tracks. Build music recommendations with the algorithm described in the lesson 3 of the fifth week. Initialize coordinates of vector x_0 corresponding to the user’s vertex and all the vertices from the task 5 with ones and all other coordinates with zeros. Do 5 iterations:\n",
    "\n",
    "Take alpha = 0.15. and the next balancing functions:\n",
    "\n",
    "- beta(user, user → artist) = 0.5\n",
    "- beta(user, user → track) = 0.5\n",
    "- beta(track, track → track) = 1\n",
    "- beta(artist, artist → track) = 1\n",
    "\n",
    "You should receive a table with 3 columns: “name”, “artist” and “rank”. Sort the resulting dataframe in descending order by “rank”, select top 40 recommended tracks, select only the columns “name”, “artist” and “rank”, leave 5 digits after the decimal point in “rank” and print the resulting dataframe.\n",
    "\n",
    "The part of the result on the sample dataset:\n",
    "\n",
    "    ...\n",
    "    Prayer Of The Refugee Artist: Rise Against 1.35278102029\n",
    "    Eagle Artist: Gotthard 1.21412311013\n",
    "    21 Guns Artist: Green Day 1.17301653219\n",
    "    Wait And Bleed Artist: Slipknot 0.921552328559\n",
    "    Beautiful disaster Artist: 311 0.921552328559\n",
    "    ...\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": 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, when, expr, lit, rank, round as rnd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = sparkSession.read.parquet(\"/data/sample264\")\n",
    "meta = sparkSession.read.parquet(\"/data/meta\")"
   ]
  },
  {
   "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": "markdown",
   "metadata": {},
   "source": [
    "--------------"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "user = 776748\n",
    "\n",
    "alpha = 0.15\n",
    "beta_user_artist = 0.5\n",
    "beta_user_track = 0.5\n",
    "beta_track_track = 1\n",
    "beta_artist_track = 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Graph based Music Recommender. Task 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "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",
    ")\n",
    "\n",
    "weights = (data1.join(data2, \"userId\")\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()\n",
    "\n",
    "norm_weights = (norm(weights, \"trackId1\", \"trackId2\", \"count\", 1000)\n",
    "                .withColumn(\"nxt_value\", col(\"norm_count\") * beta_track_track)\n",
    "               )\n",
    "track_track = norm_weights.select(\n",
    "    col(\"trackId1\").alias(\"source\"),\n",
    "    col(\"trackId2\").alias(\"target\"),\n",
    "    col(\"nxt_value\"),\n",
    ").cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Graph based Music Recommender. Task 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "weights = (data\n",
    "              .groupBy(col(\"userId\"), col(\"trackId\"))\n",
    "              .count().alias(\"count\")\n",
    "             ).cache()\n",
    "\n",
    "norm_weights = (norm(weights, \"userId\", \"trackId\", \"count\", 1000)\n",
    "                .withColumn(\"nxt_value\", col(\"norm_count\") * beta_user_track)\n",
    "               )\n",
    "\n",
    "user_track = norm_weights.select(\n",
    "    col(\"userId\").alias(\"source\"),\n",
    "    col(\"trackId\").alias(\"target\"),\n",
    "    col(\"nxt_value\"),\n",
    ").cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Graph based Music Recommender. Task 3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "weights = (data\n",
    "           .groupBy(col(\"userId\"), col(\"artistId\"))\n",
    "           .count().alias(\"count\")\n",
    "          ).cache()\n",
    "\n",
    "norm_weights = (norm(weights, \"userId\", \"artistId\", \"count\", 100)\n",
    "                .withColumn(\"nxt_value\", col(\"norm_count\") * beta_user_artist)\n",
    "               )\n",
    "\n",
    "user_artist = norm_weights.select(\n",
    "    col(\"userId\").alias(\"source\"),\n",
    "    col(\"artistId\").alias(\"target\"),\n",
    "    col(\"nxt_value\"),\n",
    ").cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Graph based Music Recommender. Task 4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "weights = (data\n",
    "           .groupBy(col(\"artistId\"), col(\"trackId\"))\n",
    "           .count().alias(\"count\")\n",
    "          ).cache()\n",
    "\n",
    "norm_weights = (norm(weights, \"artistId\", \"trackId\", \"count\", 100)\n",
    "                .withColumn(\"nxt_value\", col(\"norm_count\") * beta_artist_track)\n",
    "               )\n",
    "\n",
    "artist_track = norm_weights.select(\n",
    "    col(\"artistId\").alias(\"source\"),\n",
    "    col(\"trackId\").alias(\"target\"),\n",
    "    col(\"nxt_value\"),\n",
    ").cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Graph based Music Recommender. Task 6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "edges = (track_track\n",
    "         .union(user_track)\n",
    "         .union(user_artist)\n",
    "         .union(artist_track)\n",
    "        ).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "----------"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_data = data.filter(col(\"userId\") == user)\n",
    "\n",
    "users = (data\n",
    "         .select(col(\"userId\").alias(\"id\"))\n",
    "         .distinct()\n",
    "         .withColumn(\"p\", when(col(\"id\") == user, 1.0).otherwise(0.0))\n",
    "        )\n",
    "\n",
    "tracks = (data\n",
    "          .select(col(\"trackId\").alias(\"id\"))\n",
    "          .distinct()\n",
    "          .join(user_data.select(col(\"trackId\").alias(\"id\"), lit(1).alias(\"tmp\")).distinct(), on=\"id\", how=\"left\")\n",
    "          .withColumn(\"p\", when(~col(\"tmp\").isNull(), 1.0).otherwise(0.0))\n",
    "          .select(col(\"id\"), col(\"p\"))\n",
    "         )\n",
    "\n",
    "artists = (data\n",
    "           .select(col(\"artistId\").alias(\"id\"))\n",
    "           .distinct()\n",
    "           .join(user_data.select(col(\"artistId\").alias(\"id\"), lit(1).alias(\"tmp\")).distinct(), on=\"id\", how=\"left\")\n",
    "           .withColumn(\"p\", when(~col(\"tmp\").isNull(), 1.0).otherwise(0.0))\n",
    "           .select(col(\"id\"), col(\"p\"))\n",
    "         )\n",
    "\n",
    "x = (users\n",
    "     .union(artists)\n",
    "     .union(tracks)\n",
    "    ).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "u = (x\n",
    "     .withColumn(\"u_prob\", when(col(\"id\") == user, 1.0).otherwise(0.0))\n",
    "     .select(\"id\", \"u_prob\") \n",
    "    ).cache()\n",
    "\n",
    "\n",
    "for _ in range(5):\n",
    "    sigma = (x\n",
    "             .join(edges, on=expr(\"id = source\"), how=\"left\")\n",
    "             .na.fill(0.0, [\"nxt_value\"])\n",
    "             .withColumn(\"acc\", col(\"p\") * col(\"nxt_value\"))\n",
    "             .groupBy(\"target\")\n",
    "             .agg(sum(\"acc\").alias(\"sigma\"))\n",
    "            )\n",
    "    x = (u\n",
    "         .join(sigma, on=expr(\"id = target\"), how=\"left\")\n",
    "         .na.fill(0.0, [\"sigma\"])\n",
    "         .withColumn(\"nxt_value\", alpha*col(\"u_prob\") + (1-alpha) * col(\"sigma\"))\n",
    "         .select(col(\"id\"), col(\"nxt_value\").alias(\"p\"))\n",
    "        ).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = (x\n",
    "          .where(\"id != \" + str(user))\n",
    "          .join(meta, on=\"id\")\n",
    "          .orderBy(col(\"p\").desc())\n",
    "          .select(col(\"Name\"), col(\"Artist\"), rnd(col(\"p\"), 5).alias(\"p\"))\n",
    "         ).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Kill The DJ Artist: Green Day 1.42809\n",
      "Come Out and Play Artist: The Offspring 1.37473\n",
      "I Hate Everything About You Artist: Three Days Grace 1.37362\n",
      "Prayer Of The Refugee Artist: Rise Against 1.35278\n",
      "Eagle Artist: Gotthard 1.21412\n",
      "21 Guns Artist: Green Day 1.17302\n",
      "Beautiful disaster Artist: 311 0.92155\n",
      "Wait And Bleed Artist: Slipknot 0.92155\n",
      "Here To Stay Artist: Korn 0.91653\n",
      "Hard Rock Hallelujah Artist: Lordi 0.91653\n",
      "Nothing Going On Artist: Clawfinger 0.80983\n",
      "In The End Artist: Linkin Park 0.80292\n",
      "Numb Artist: Linkin Park 0.80292\n",
      "Sky is Over Artist: Serj Tankian 0.68799\n",
      "Kryptonite Artist: 3 Doors Down 0.68799\n",
      "Take It Out On Me Artist: Thousand Foot Krutch 0.47024\n",
      "Girls and Boys Artist: Blur 0.40245\n",
      "Cocaine Artist: Nomy 0.20893\n",
      "Getting Away With Murder Artist: Papa Roach 0.20648\n",
      "Artist: Green Day Artist: Green Day 0.01181\n",
      "Artist: Clawfinger Artist: Clawfinger 0.00472\n",
      "Artist: The Offspring Artist: The Offspring 0.00472\n",
      "Artist: Linkin Park Artist: Linkin Park 0.00472\n",
      "The Vengeful One Artist: Disturbed 0.00437\n",
      "She Keeps Me Up Artist: Nickelback 0.00437\n",
      "Sunday Artist: Iggy Pop 0.00437\n",
      "Artist: Iggy Pop Artist: Iggy Pop 0.00236\n",
      "Artist: Gotthard Artist: Gotthard 0.00236\n",
      "Artist: Slipknot Artist: Slipknot 0.00236\n",
      "Artist: Thousand Foot Krutch Artist: Thousand Foot Krutch 0.00236\n",
      "Artist: Papa Roach Artist: Papa Roach 0.00236\n",
      "Artist: Blur Artist: Blur 0.00236\n",
      "Artist: Rise Against Artist: Rise Against 0.00236\n",
      "Artist: Nickelback Artist: Nickelback 0.00236\n",
      "Artist: Serj Tankian Artist: Serj Tankian 0.00236\n",
      "Artist: Three Days Grace Artist: Three Days Grace 0.00236\n",
      "Artist: Disturbed Artist: Disturbed 0.00236\n",
      "Artist: 311 Artist: 311 0.00236\n",
      "Artist: Korn Artist: Korn 0.00236\n",
      "Artist: Nomy Artist: Nomy 0.00236\n"
     ]
    }
   ],
   "source": [
    "for name, artist, p in results.limit(40).collect():\n",
    "    print(\"{} {} {}\".format(name, artist, p))"
   ]
  },
  {
   "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
}