course2-big-data-analysis/c02w04-Mutal-Friends.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Counting number of the mutual friends"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For each user having ID in the column userId count the amount of his / her common friends with each other user having ID in the column userId.\n",
    "\n",
    "Print 49 pairs of the users having the largest amount of common friends, ordered in descending order first by the common friends count , then by id of user1 and finally by id of user 2. The format is following: \"count user1 user2\"7\n",
    "\n",
    "Example:\n",
    "\n",
    "    234 54719 767867\n",
    "\n",
    "    120 54719 767866\n",
    "\n",
    "    97 50787 327676\n",
    "\n",
    "To solve this task use the algorithm described in the last video of lesson 1. The overall plan could look like this:\n",
    "\n",
    "1. Create a new column “friend” by exploding of column “friends” (like in the demo iPython notebook)\n",
    "1. group the resulting dataframe by the column “friend” (like in the demo iPython notebook)\n",
    "1. create a column “users” by collecting all users with the same id in the column “friend” together (like in the demo iPython notebook)\n",
    "1. sort the elements in the column “users” by the function sort_array\n",
    "1. filter only the rows which have more than 1 element in the column “users”\n",
    "1. for each row emit all possible ordered pairs of users from the column “users” (tip: write a user defined function for this)\n",
    "1. count the number of times each pair has appeared \n",
    "1. with the help of the window function (like in the demo python notebook) select 49 pairs of users who have the biggest amount of common friends\n",
    "\n",
    "The sample dataset is located at /data/graphDFSample.\n",
    "\n",
    "The part of the result on the sample dataset:\n",
    "\n",
    "    ...\n",
    "    3044 21864412 51640390\n",
    "    3021 17139850 51640390\n",
    "    3010 14985079 51640390\n",
    "    2970 17139850 21864412\n",
    "    2913 20158643 27967558\n",
    "    ...\n",
    "\n",
    "Notes\n",
    "\n",
    "The system grades standard output and error streams from the last non-empty cell.\n",
    "If you have clicked \"Open tool\" and received \"404\" or \"405\" error, please reload the page from the Coursera interface.\n",
    "\n",
    "If you want to deploy the environment on your own machine, please use [bigdatateam/spark-course2](https://hub.docker.com/r/bigdatateam/spark-course2/) Docker container."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "sparkSession = SparkSession.builder.enableHiveSupport().master(\"local\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import explode, collect_list, size, col, row_number, sort_array,udf, count, desc\n",
    "from pyspark.sql import Window, types\n",
    "from itertools import combinations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "graphPath = \"/data/graphDFSample\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "def emit(users):\n",
    "    return combinations(sorted(users), 2)\n",
    "\n",
    "pair_schema = types.StructType([\n",
    "    types.StructField(\"u1\", types.IntegerType(), False),\n",
    "    types.StructField(\"u2\", types.IntegerType(), False)\n",
    "])\n",
    "\n",
    "udf_emit = udf(emit, types.ArrayType(pair_schema))\n",
    "\n",
    "reversedGraph = (sparkSession.read.parquet(graphPath)\n",
    "                 .withColumn(\"friend\", explode('friends'))\n",
    "                 .groupBy(\"friend\")\n",
    "                 .agg(collect_list(\"user\").alias(\"users\"))\n",
    "                 .withColumn(\"users\", sort_array('users'))\n",
    "                 .where(size('users') > '1')\n",
    "                )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pairs = (reversedGraph.withColumn(\"pairs\", udf_emit('users'))\n",
    "         .withColumn(\"pair\", explode('pairs'))\n",
    "         .groupBy(\"pair\")\n",
    "         .agg(count(\"pair\").alias(\"pair_count\"))\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = (pairs.select(col(\"pair_count\"), \"pair.*\")\n",
    "           .orderBy(desc(\"pair_count\"), desc(\"u1\"), desc(\"u2\"))\n",
    "           .limit(49)\n",
    "          ).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3206 27967558 42973992\n",
      "3130 20158643 42973992\n",
      "3066 22582764 42973992\n",
      "3044 21864412 51640390\n",
      "3021 17139850 51640390\n",
      "3010 14985079 51640390\n",
      "2970 17139850 21864412\n",
      "2913 20158643 27967558\n",
      "2903 22280814 51151280\n",
      "2870 23848749 51640390\n",
      "2855 20158643 22582764\n",
      "2849 20158643 44996025\n",
      "2846 22280814 42973992\n",
      "2784 21864412 23848749\n",
      "2779 31964081 51640390\n",
      "2776 39205988 51640390\n",
      "2754 17139850 23848749\n",
      "2749 22582764 27967558\n",
      "2728 50561859 51640390\n",
      "2724 15485897 51640390\n",
      "2700 28135661 42973992\n",
      "2655 22280814 27967558\n",
      "2653 42973992 43548989\n",
      "2639 26755857 51640390\n",
      "2621 14635589 51640390\n",
      "2608 15485897 17139850\n",
      "2606 17139850 26755857\n",
      "2601 21864412 39205988\n",
      "2600 8406745 51640390\n",
      "2599 37735419 51640390\n",
      "2597 20158643 28135661\n",
      "2585 40003405 42973992\n",
      "2585 21864412 31964081\n",
      "2581 27967558 43548989\n",
      "2579 23848749 31964081\n",
      "2578 27967558 28135661\n",
      "2578 15485897 21864412\n",
      "2577 42973992 64755069\n",
      "2574 51151280 57077210\n",
      "2573 20158643 43548989\n",
      "2566 21864412 26755857\n",
      "2564 22280814 64755069\n",
      "2561 42973992 44996025\n",
      "2556 17139850 39205988\n",
      "2543 23848749 39205988\n",
      "2521 17139850 31964081\n",
      "2515 27967558 44996025\n",
      "2506 41629539 51640390\n",
      "2505 51151280 64755069\n"
     ]
    }
   ],
   "source": [
    "for pair, u1, u2 in results:\n",
    "    print(\"{} {} {}\".format(pair, u1, u2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}