friend_recommend.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/Praneeth33/Hello_world-/blob/main/friend_recommend.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import sys\n",
        "import itertools\n",
        "from pyspark import SparkConf, SparkContext\n",
        "\n",
        "\n",
        "def line_to_friend_ownership(line):\n",
        "   \n",
        "    split = line.split()\n",
        "    user_id = int(split[0])\n",
        "\n",
        "    if len(split) == 1:\n",
        "        friends = []\n",
        "    else:\n",
        "        friends = list(map(lambda x: int(x), split[1].split(',')))\n",
        "\n",
        "    return user_id, friends\n",
        "\n",
        "\n",
        "def friend_ownership_to_connection(f_o):\n",
        "    user_id = f_o[0]\n",
        "    friends = f_o[1]\n",
        "\n",
        "    connections = []\n",
        "\n",
        "    for friend_id in friends:\n",
        "        key = (user_id, friend_id)\n",
        "        if user_id > friend_id:\n",
        "            key = (friend_id, user_id)\n",
        "\n",
        "        connections.append(\n",
        "            (key, 0)\n",
        "        )\n",
        "\n",
        "    for friend_pair in itertools.combinations(friends, 2):\n",
        "        friend_0 = friend_pair[0]\n",
        "        friend_1 = friend_pair[1]\n",
        "\n",
        "        key = (friend_0, friend_1)\n",
        "        if friend_0 > friend_1:\n",
        "            key = (friend_1, friend_0)\n",
        "        connections.append(\n",
        "            (key, 1)\n",
        "        )\n",
        "\n",
        "    return connections\n",
        "\n",
        "\n",
        "def mutual_friend_count_to_recommendation(m):\n",
        "    \"\"\"\n",
        "    Maps a \"mutual friend count\" object to two distinct recommendations. The value\n",
        "    ``((0, 1), 21)`` encodes that users 0 and 1 share 21 mutual friends. This means that user 1 should be recommended\n",
        "    to user 0 AND that user 0 should be recommended to user 1. For every input to this function, two \"recommendations\"\n",
        "    will be returned in a List.\n",
        "    A \"recommendation\" has the following form::\n",
        "        (user_id_0, (recommended_user, mutual_friends_count))\n",
        "    :param m: a mutual friend count item\n",
        "    :return: List[Tuple[int, Tuple[int, int]]] two recommendation items\n",
        "    \"\"\"\n",
        "    connection = m[0]\n",
        "    count = m[1]\n",
        "\n",
        "    friend_0 = connection[0]\n",
        "    friend_1 = connection[1]\n",
        "\n",
        "    recommendation_0 = (friend_0, (friend_1, count))\n",
        "    recommendation_1 = (friend_1, (friend_0, count))\n",
        "\n",
        "    return [recommendation_0, recommendation_1]\n",
        "\n",
        "\n",
        "def recommendation_to_sorted_truncated(recs):\n",
        "    if len(recs) > 1024:\n",
        "        # Before sorting, find the highest 10 elements in recs (if log(len(recs)) > 10)\n",
        "        # This optimization runs in O(n), where n is the length of recs. This is so that sorting the best 10\n",
        "        # recommendations can run in constant time. Otherwise, sorting the whole list would run in O(n lgn). \n",
        "        # As long as n > 1024 (or, in other words, lg(n) > 10), this is faster.\n",
        "\n",
        "        max_indices = []\n",
        "\n",
        "        for current_rec_number in range(0, 10):\n",
        "            current_max_index = 0\n",
        "            for i in range(1, len(recs)):\n",
        "                rec = recs[i]\n",
        "                if rec[1] >= recs[current_max_index][1] and i not in max_indices:\n",
        "                    current_max_index = i\n",
        "\n",
        "            max_indices.append(current_max_index)\n",
        "\n",
        "        recs = [recs[i] for i in max_indices]\n",
        "\n",
        "    # Sort first by mutual friend count, then by user_id (for equal number of mutual friends between users)\n",
        "    recs.sort(key=lambda x: (-x[1], x[0]))\n",
        "\n",
        "    # Map every [(user_id, mutual_count), ...] to [user_id, ...] and truncate to 10 elements\n",
        "    return list(map(lambda x: x[0], recs))[:10]\n",
        "\n",
        "\n",
        "# ============ #\n",
        "# MAIN PROGRAM #\n",
        "# ============ #\n",
        "\n",
        "# Initialize spark configuration and context\n",
        "\n",
        "# Read from text file, split each line into \"words\" by any whitespace (i.e. empty parameters to string.split())\n",
        "lines = sc.textFile(\"FriendData.txt\")\n",
        "\n",
        "# Map each line to the form: (user_id, [friend_id_0, friend_id_1, ...])\n",
        "friend_ownership = lines.map(line_to_friend_ownership)\n",
        "\n",
        "# Map each \"friend ownership\" to multiple instances of ((user_id, friend_id), VALUE).\n",
        "# VALUE = 0 indicates that user_id and friend_id are already friends.\n",
        "# VALUE = 1 indicates that user_id and friend_id are not friends.\n",
        "friend_edges = friend_ownership.flatMap(friend_ownership_to_connection)\n",
        "friend_edges.cache()\n",
        "\n",
        "# Filter all pairs of users that are already friends, then sum all the \"1\" values to get their mutual friend count.\n",
        "mutual_friend_counts = friend_edges.groupByKey() \\\n",
        "    .filter(lambda edge: 0 not in edge[1]) \\\n",
        "    .map(lambda edge: (edge[0], sum(edge[1])))\n",
        "\n",
        "# Create the recommendation objects, group them by key, then sort and truncate the recommendations to the 10 most\n",
        "# highly recommended.\n",
        "recommendations = mutual_friend_counts.flatMap(mutual_friend_count_to_recommendation) \\\n",
        "    .groupByKey() \\\n",
        "    .map(lambda m: (m[0], recommendation_to_sorted_truncated(list(m[1]))))\n",
        "\n",
        "# Save to output directory, end context\n",
        "recommendations.saveAsTextFile(\"Result\")\n",
        "sc.stop()"
      ],
      "metadata": {
        "id": "LN0YVkMpjzMw"
      },
      "execution_count": 7,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "opKI2fgBkGLz"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install pyspark\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "2fdrZM-7kPVg",
        "outputId": "f4e805bc-24e5-468b-ae39-607fa4c10fec"
      },
      "execution_count": 4,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
            "Collecting pyspark\n",
            "  Downloading pyspark-3.4.0.tar.gz (310.8 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m310.8/310.8 MB\u001b[0m \u001b[31m3.7 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from pyspark) (0.10.9.7)\n",
            "Building wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.4.0-py2.py3-none-any.whl size=311317130 sha256=9018e32e130bc4d540968ecfacacb4889e0df17a6c9106e4e57670ba07d6edf1\n",
            "  Stored in directory: /root/.cache/pip/wheels/7b/1b/4b/3363a1d04368e7ff0d408e57ff57966fcdf00583774e761327\n",
            "Successfully built pyspark\n",
            "Installing collected packages: pyspark\n",
            "Successfully installed pyspark-3.4.0\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "gJLXgtOgkQxo"
      },
      "execution_count": null,
      "outputs": []
    }
  ],
  "metadata": {
    "colab": {
      "provenance": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}