hw1/q1.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "RWSJpsyKqHjH",
        "outputId": "b26df1aa-a311-4987-bf20-195ce66e0a12"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Downloading...\n",
            "From: https://drive.google.com/uc?id=1-OCBGBtKoY_PadKHcXDyWxHQ2BS8Nulo\n",
            "To: /content/bigdata_hw1_files.zip\n",
            "100% 38.9M/38.9M [00:00<00:00, 144MB/s]\n"
          ]
        }
      ],
      "source": [
        "!pip install --upgrade --no-cache-dir gdown >/dev/null\n",
        "!gdown 1-OCBGBtKoY_PadKHcXDyWxHQ2BS8Nulo"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "17mGoDg2sxWC",
        "outputId": "ce96221a-4b9a-4118-e864-428255a6f87c"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Archive:  bigdata_hw1_files.zip\n",
            "   creating: hw1-files/\n",
            "   creating: hw1-files/q3/\n",
            "  inflating: hw1-files/q3/patches.csv  \n",
            "  inflating: hw1-files/q3/lsh.py     \n",
            "   creating: hw1-files/q1/\n",
            "  inflating: hw1-files/q1/dataset1.txt  \n",
            "   creating: hw1-files/q1/.ipynb_checkpoints/\n",
            "   creating: hw1-files/q2/\n",
            "  inflating: hw1-files/q2/games_library.txt  \n",
            "   creating: hw1-files/.ipynb_checkpoints/\n"
          ]
        }
      ],
      "source": [
        "!unzip bigdata_hw1_files.zip"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ahIn1Qrf29uF",
        "outputId": "bd7dc590-0bfa-4703-8367-4ef4a847450a"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting pyspark\n",
            "  Downloading pyspark-3.2.1.tar.gz (281.4 MB)\n",
            "\u001b[K     |████████████████████████████████| 281.4 MB 26 kB/s \n",
            "\u001b[?25hCollecting py4j==0.10.9.3\n",
            "  Downloading py4j-0.10.9.3-py2.py3-none-any.whl (198 kB)\n",
            "\u001b[K     |████████████████████████████████| 198 kB 52.0 MB/s \n",
            "\u001b[?25hBuilding wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.2.1-py2.py3-none-any.whl size=281853642 sha256=ee3057050fa4fdc6fb0a68a4af3f046a5524f912f32928a7571587f6507d8c4e\n",
            "  Stored in directory: /root/.cache/pip/wheels/9f/f5/07/7cd8017084dce4e93e84e92efd1e1d5334db05f2e83bcef74f\n",
            "Successfully built pyspark\n",
            "Installing collected packages: py4j, pyspark\n",
            "Successfully installed py4j-0.10.9.3 pyspark-3.2.1\n"
          ]
        }
      ],
      "source": [
        "!pip install pyspark"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 14,
      "metadata": {
        "id": "hZSzvYQ92hD3"
      },
      "outputs": [],
      "source": [
        "from pyspark import SparkContext\n",
        "\n",
        "N = 10\n",
        "INPUT_USERS = (98, 135, 117, 911, 8804)\n",
        "INPUT_FILE = 'hw1-files/q1/dataset1.txt'"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 16,
      "metadata": {
        "id": "jfIJuEGvVHpp"
      },
      "outputs": [],
      "source": [
        "def map_func(line):\n",
        "    # Split user id and friend ids\n",
        "    tokens = line.split('\\t')\n",
        "\n",
        "    # Skip bad lines\n",
        "    if len(tokens) < 2:\n",
        "        return []\n",
        "\n",
        "    # Parse line\n",
        "    user = int(tokens[0])\n",
        "    friends = [int(f) for f in tokens[1].split(',') if f.strip() != '']\n",
        "    pairs = []\n",
        "\n",
        "    # Mark already friendships with 0\n",
        "    for f in friends:\n",
        "        if user in INPUT_USERS:\n",
        "            pairs.append(((user, f), 0))\n",
        "        if f in INPUT_USERS:\n",
        "            pairs.append(((f, user), 0))\n",
        "\n",
        "    # Mark mutual friendships with 1\n",
        "    for i in range(0, len(friends) - 1):\n",
        "        for j in range(i + 1, len(friends)):\n",
        "            a = friends[i]\n",
        "            b = friends[j]\n",
        "            if a in INPUT_USERS:\n",
        "                pairs.append(((a, b), 1))\n",
        "            if b in INPUT_USERS:\n",
        "                pairs.append(((b, a), 1))\n",
        "\n",
        "    return pairs"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 18,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "AXFaEAcEBdUb",
        "outputId": "a5a65298-3880-4261-d474-ccea89b09c81"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "98 \t 18560,16324,30691,2554,30134,16350,13654,168,5924,5052\n",
            "117 \t 34164,13793,34169,15314,23510,12519,34207,23507,34140,34220\n",
            "135 \t 33060,13792,34151,25256,34164,629,19217,34441,45054,5490\n",
            "911 \t 24456,40560,39540,30995,37875,30984,30996,41352,30993,33333\n",
            "8804 \t 34332,34179,3230,34174,13182,29745,8677,34233,13872,11400\n"
          ]
        }
      ],
      "source": [
        "# Create context and read input file\n",
        "sc = SparkContext.getOrCreate()\n",
        "rdd = sc.textFile(INPUT_FILE)\n",
        "\n",
        "# Mapping step\n",
        "map_result = rdd.flatMap(map_func)\n",
        "\n",
        "# Fetch already friendship pairs.\n",
        "# Marked them with 0 value in the mapping step.\n",
        "already_friends = map_result.filter(lambda x: x[1] == 0)\n",
        "\n",
        "# Reducing step\n",
        "# Remove already friendship pairs through subtraction to avoid absurd recommendations!\n",
        "# Simply reduce by summing values for each key.\n",
        "# Group by has been considered by spark in reduceByKey function.\n",
        "reduce_result = map_result \\\n",
        "    .subtractByKey(already_friends) \\\n",
        "    .reduceByKey(lambda a, b: a + b)\n",
        "\n",
        "# Transform mapping format\n",
        "# Consider user id as key and a list of pairs of recommended friend id with its mutual count as value.\n",
        "# Sort by mutual count in ascending order.\n",
        "# Group by user id.\n",
        "# Sort by user id to generate the output in descending order.\n",
        "# Include only N top recommendations for each user.\n",
        "recoms = reduce_result \\\n",
        "    .map(lambda x: (x[0][0], (x[0][1], x[1]))) \\\n",
        "    .sortBy(lambda x: -x[1][1]) \\\n",
        "    .groupByKey() \\\n",
        "    .sortByKey() \\\n",
        "    .mapValues(list) \\\n",
        "    .map(lambda x: (x[0], x[1][:N]))\n",
        "\n",
        "# Print recommendations for each input user\n",
        "for r in recoms.collect():\n",
        "    recom_list = [str(p[0]) for p in r[1]]\n",
        "    print(r[0], '\\t', ','.join(recom_list))"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "q1.ipynb",
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}