Degrees of Separation with Breadth-first Search Algorithm.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "a5cdb163",
   "metadata": {},
   "source": [
    "# Determining Degrees of Separation between Marvel Superheroes with Breadth-first Search in PySpark RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44a9dab8",
   "metadata": {},
   "source": [
    "In this project, the **Breath-first Search (BFS)** algorithm was implemented in **Spark RDD** to find the degrees of separation between two given [Marvel Superheros](https://www.marvel.com/comics/characters)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "9d99694b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf, SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "d63c0ffc",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a new SparkConf object to configure the Spark application\n",
    "conf = SparkConf().setMaster(\"local\").setAppName(\"DegreesOfSeparation\")\n",
    "# Create a new SparkContext object with the specified configuration (conf)\n",
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "042fd8ba",
   "metadata": {},
   "source": [
    "### About the Datasets"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2f64f883",
   "metadata": {},
   "source": [
    "The **Marvel-names.txt** dataset consists of two columns: **heroID** and **superhero names**. The heroID uniquely identifies each individual superhero in the dataset. The superhero names represent the names of the corresponding superheroes. The dataset provides a mapping between the heroIDs and their corresponding superhero names, allowing easy identification and reference to specific superheroes within the Marvel universe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "4eae5735",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1 \"24-HOUR MAN/EMMANUEL\"\n",
      "2 \"3-D MAN/CHARLES CHAN\"\n",
      "3 \"4-D MAN/MERCURIO\"\n",
      "4 \"8-BALL/\"\n",
      "5 \"A\"\n",
      "6 \"A'YIN\"\n",
      "7 \"ABBOTT, JACK\"\n",
      "8 \"ABCISSA\"\n",
      "9 \"ABEL\"\n",
      "10 \"ABOMINATION/EMIL BLO\"\n"
     ]
    }
   ],
   "source": [
    "!head -n 10 ./data/Marvel-names.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4960ea0c",
   "metadata": {},
   "source": [
    "The **Marvel-graph.txt** dataset represents the network of Marvel superheroes based on their appearances together in the same comics. It is structured as a graph with each row containing a superhero's ID (heroID) followed by the IDs of other superheroes with whom they have appeared together in comics. The dataset helps establish connections between superheroes who have shared comic book appearances, reflecting their interactions and collaborations within the Marvel universe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "6c316388",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5988 748 1722 3752 4655 5743 1872 3413 5527 6368 6085 4319 4728 1636 2397 3364 4001 1614 1819 1585 732 2660 3952 2507 3891 2070 2239 2602 612 1352 5447 4548 1596 5488 1605 5517 11 479 2554 2043 17 865 4292 6312 473 534 1479 6375 4456 \n",
      "5989 4080 4264 4446 3779 2430 2297 6169 3530 3272 4282 6432 2548 4140 185 105 3878 2429 1334 4595 2767 3956 3877 4776 4946 3407 128 269 5775 5121 481 5516 4758 4053 1044 1602 3889 1535 6038 533 3986 \n",
      "5982 217 595 1194 3308 2940 1815 794 1503 5197 859 5096 6039 2664 651 2244 528 284 1449 1097 1172 1092 108 3405 5204 387 4607 4545 3705 4930 1805 4712 4404 247 4754 4427 1845 536 5795 5978 533 3984 6056 \n",
      "5983 1165 3836 4361 1282 716 4289 4646 6300 5084 2397 4454 1913 5861 5485 \n",
      "5980 2731 3712 1587 6084 2472 2546 6313 875 859 323 2664 1469 522 2506 2919 2423 3624 5736 5046 1787 5776 3245 3840 2399 \n"
     ]
    }
   ],
   "source": [
    "!head -n 5 ./data/Marvel-graph.txt"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "05a7185b",
   "metadata": {},
   "source": [
    "### Define the functions for Breadth-first Search"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "671f8441",
   "metadata": {},
   "source": [
    "In this project, the **Breadth-first search (BFS)** algorithm is implemented through a set of user-defined functions to determine the degrees of separation between two Marvel superheros. The concept can be illustrated through the simple example shown in the image below."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b8e21853",
   "metadata": {},
   "source": [
    "![spark-BFS](https://github.com/MaxineXiong/Degrees-of-Separation-with-Breadth-first-Search/assets/55864839/ebf8fa3d-3726-47db-a716-e84baf19b509)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "87e39ad1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to convert each input record into the format (heroID, (connections, distance, color))\n",
    "def convertToBFS(x):\n",
    "    # Extract the hero ID from the input record and convert it to an integer\n",
    "    heroID = int(x.split()[0])\n",
    "    # Extract the list of connections from the input record and convert them to integers\n",
    "    connections = [int(connection) for connection in x.split()[1:]]\n",
    "    # Remove duplicate connections\n",
    "    connections = list(set(connections))\n",
    "    # Initialize the distance to a large value (effectively infinity)\n",
    "    distance = 9999\n",
    "    # Initialize the color of the hero to 'WHITE', indicating it has not been visited yet\n",
    "    color = \"WHITE\"\n",
    "\n",
    "    # Check if the heroID is the source hero\n",
    "    if heroID == sourceHeroID:\n",
    "        # If the hero is the source hero, set the distance to 0 as it is the starting point\n",
    "        distance = 0\n",
    "        # Also set the color to 'GRAY', indicating that it has been discovered but not fully explored\n",
    "        color = \"GRAY\"\n",
    "\n",
    "    return (heroID, (connections, distance, color))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "ebfeb1a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to create the initial RDD from the input file\n",
    "def createStartingRdd(sc, file_path):\n",
    "    # Read the input file containing the Marvel graph data and create an RDD\n",
    "    RDD_lines = sc.textFile(file_path)\n",
    "    # Map each line of the RDD to a new format (heroID, (connections, distance, color)) using the convertToBFS function\n",
    "    return RDD_lines.map(convertToBFS)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "1eb9b8f9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to perform the map operation for the RDD\n",
    "def bfsMap(x):\n",
    "    # Extract the hero ID, connections, distance, and color from the input record\n",
    "    heroID = x[0]\n",
    "    connections = x[1][0]\n",
    "    distance = x[1][1]\n",
    "    color = x[1][2]\n",
    "\n",
    "    # Initialize an empty list to store the results of the BFS map phase\n",
    "    result = []\n",
    "\n",
    "    # Exploring the hero that's in 'GRAY' state in the current BFS iteration\n",
    "    if color == \"GRAY\":\n",
    "        # Change the color of the current hero to 'BLACK', indicating it has been fully explored in this iteration\n",
    "        color = \"BLACK\"\n",
    "\n",
    "        # Iterate through the connections of the current hero\n",
    "        for connection in connections:\n",
    "            # Create a new entry for each connection with an updated distance and the color set to 'GRAY'\n",
    "            # This represents the exploration of the next level in the BFS\n",
    "            newEntry = (connection, ([], distance + 1, \"GRAY\"))\n",
    "            # Append the new entry to the result list\n",
    "            result.append(newEntry)\n",
    "            # If the connection is the target hero, increment the hitCounter indicating that the target is reached\n",
    "            if connection == targetHeroID:\n",
    "                hitCounter.add(1)\n",
    "\n",
    "    # Append the current hero to the result list as well\n",
    "    result.append((heroID, (connections, distance, color)))\n",
    "\n",
    "    # Return the result list containing the updated hero information and its connected nodes for the next BFS iteration\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "73a52641",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to perform the reduce operation for the RDD\n",
    "def bfsReduce(x, y):\n",
    "    # Extract the connections, distance, and color from the input records x and y that share the same key (i.e. heroID)\n",
    "    connections_1 = x[0]\n",
    "    connections_2 = y[0]\n",
    "    distance_1 = x[1]\n",
    "    distance_2 = y[1]\n",
    "    color_1 = x[2]\n",
    "    color_2 = y[2]\n",
    "\n",
    "    # Merge the connections from both records and remove duplicates\n",
    "    connections = connections_1 + connections_2\n",
    "    connections = list(set(connections))\n",
    "\n",
    "    # Determine the minimum distance from the source hero to this hero\n",
    "    if distance_1 <= distance_2:\n",
    "        distance = distance_1\n",
    "    else:\n",
    "        distance = distance_2\n",
    "\n",
    "    # Determine the final color for this hero\n",
    "    if (color_1 == \"WHITE\" and color_2 == \"GRAY\") or (\n",
    "        color_1 == \"GRAY\" and color_2 == \"WHITE\"\n",
    "    ):\n",
    "        color = \"GRAY\"\n",
    "    elif (color_1 == \"BLACK\" and color_2 == \"GRAY\") or (\n",
    "        color_1 == \"GRAY\" and color_2 == \"BLACK\"\n",
    "    ):\n",
    "        color = \"BLACK\"\n",
    "    else:\n",
    "        color = color_1\n",
    "\n",
    "    # Return a new tuple as value representing the combined information for this hero\n",
    "    return (connections, distance, color)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f032ddc4",
   "metadata": {},
   "source": [
    "Read **Marvel-names.txt** data into a RDD for superhero queries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "f92c6919",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, '24-HOUR MAN/EMMANUEL'),\n",
       " (2, '3-D MAN/CHARLES CHAN'),\n",
       " (3, '4-D MAN/MERCURIO'),\n",
       " (4, '8-BALL/'),\n",
       " (5, 'A'),\n",
       " (6, \"A'YIN\"),\n",
       " (7, 'ABBOTT, JACK'),\n",
       " (8, 'ABCISSA'),\n",
       " (9, 'ABEL'),\n",
       " (10, 'ABOMINATION/EMIL BLO')]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Read the input file 'Marvel-names.txt' containing hero names and IDs and create an RDD\n",
    "RDD_heros = sc.textFile(\"./data/Marvel-names.txt\")\n",
    "# Map each line of the RDD to a new format (heroID, heroName)\n",
    "RDD_heros = RDD_heros.map(\n",
    "    lambda x: (int(x.split()[0]), \" \".join(x.split()[1:])[1:-1])\n",
    ")\n",
    "\n",
    "# Preview the first 10 records\n",
    "RDD_heros.collect()[:10]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "d7fb6402",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to perform a search for a superhero's ID based on their name or vice versa\n",
    "def searchHero(heroName=None, heroID=None):\n",
    "    # Search for a superhero's ID based on name\n",
    "    if heroName:\n",
    "        return RDD_heros.filter(lambda x: heroName.upper() in x[1]).collect()\n",
    "\n",
    "    # Search for a superhero's name based on ID\n",
    "    if heroID:\n",
    "        return RDD_heros.filter(lambda x: x[0] == heroID).collect()[0][1]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "4b95506e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Main logic function for executing the BFS algorithm to find degrees of separations between two given superheros\n",
    "def bfsMain(sc, sourceHeroID, targetHeroID):\n",
    "    # Print the names of the source hero and target hero based on their IDs\n",
    "    sourceHeroName = searchHero(heroID=sourceHeroID)\n",
    "    targetHeroName = searchHero(heroID=targetHeroID)\n",
    "    print(\"Source hero: \" + sourceHeroName)\n",
    "    print(\"Target hero: {}\\n\".format(targetHeroName))\n",
    "\n",
    "    # Initialize the hitCounter to 0 as no target hero has been hit yet\n",
    "    global hitCounter\n",
    "    hitCounter = sc.accumulator(0)\n",
    "    # Initialize the iteration variable to 1, which represents the index for current iteration\n",
    "    iteration = 1\n",
    "\n",
    "    # Create the initial RDD to start the BFS algorithm\n",
    "    network_file_path = \"./data/Marvel-graph.txt\"\n",
    "    iterationRDD = createStartingRdd(sc, network_file_path)\n",
    "\n",
    "    # Continue BFS iterations until the target hero is found or all connected heros have been explored\n",
    "    while True:\n",
    "        print(\"Running BFS iteration {}\".format(iteration))\n",
    "\n",
    "        # Perform the map operation to explore the connections of each 'GRAY' hero\n",
    "        mapRDD = iterationRDD.flatMap(bfsMap)\n",
    "\n",
    "        # Count the number of records in the mapRDD,\n",
    "        # forcing the RDD to be evaluated and updating the hitCounter\n",
    "        print(\"Processing {} values.\".format(mapRDD.count()))\n",
    "\n",
    "        # Check if the target hero has been hit (i.e., reached during exploration)\n",
    "        if hitCounter.value > 0:\n",
    "            print(\n",
    "                \"Target character successfully hit! From {} different direction(s).\".format(\n",
    "                    hitCounter.value\n",
    "                )\n",
    "            )\n",
    "            print(\n",
    "                \"\\n{} is {} degree(s) of separation away from {}\".format(\n",
    "                    targetHeroName, iteration, sourceHeroName\n",
    "                )\n",
    "            )\n",
    "            break\n",
    "        else:\n",
    "            # If the target hero has not been hit, prepare for the next BFS iteration\n",
    "            # Increment the index for iteration\n",
    "            iteration += 1\n",
    "            # Combine the records with same heroID into a single record in the updated iterationRDD\n",
    "            iterationRDD = mapRDD.reduceByKey(bfsReduce)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b5e14f9a",
   "metadata": {},
   "source": [
    "### Test to find the degrees of separation between two given superheros"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "714f91da",
   "metadata": {},
   "source": [
    "Find the degrees of separation between two random superheros"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "b1090651",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Source hero: SPERZEL, ANTON\n",
      "Target hero: AGAMEMNON\n",
      "\n",
      "Running BFS iteration 1\n",
      "Processing 6594 values.\n",
      "Running BFS iteration 2\n",
      "Processing 10683 values.\n",
      "Running BFS iteration 3\n",
      "Processing 248780 values.\n",
      "Running BFS iteration 4\n",
      "Processing 96187 values.\n",
      "Target character successfully hit! From 2 different direction(s).\n",
      "\n",
      "AGAMEMNON is 4 degree(s) of separation away from SPERZEL, ANTON\n"
     ]
    }
   ],
   "source": [
    "# Define the ID of the source hero and the target hero\n",
    "sourceHeroID = 5301\n",
    "targetHeroID = 41\n",
    "\n",
    "# Call the bfsMain function to perform BFS\n",
    "bfsMain(sc, sourceHeroID, targetHeroID)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8b402b41",
   "metadata": {},
   "source": [
    "Find the degrees of separation between **_Deadpool_** and **_Quicksilver_**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "90fdc45f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1397, 'DEADPOOL/JACK/WADE W')]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Search for Deadpool's ID\n",
    "searchHero(heroName=\"deadpool\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "25ca8ff5",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(4454, 'QUICKSILVER/PIETRO M'),\n",
       " (4455, 'QUICKSILVER DOPPELGA'),\n",
       " (4456, 'QUICKSILVER | MUTANT')]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Search for Quicksilver's ID\n",
    "searchHero(heroName=\"quicksilver\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "34736b43",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Source hero: DEADPOOL/JACK/WADE W\n",
      "Target hero: QUICKSILVER/PIETRO M\n",
      "\n",
      "Running BFS iteration 1\n",
      "Processing 6797 values.\n",
      "Running BFS iteration 2\n",
      "Processing 49756 values.\n",
      "Target character successfully hit! From 74 different direction(s).\n",
      "\n",
      "QUICKSILVER/PIETRO M is 2 degree(s) of separation away from DEADPOOL/JACK/WADE W\n"
     ]
    }
   ],
   "source": [
    "# Define the ID of the source hero and the target hero\n",
    "sourceHeroID = 1397\n",
    "targetHeroID = 4454\n",
    "\n",
    "# Call the bfsMain function to perform BFS\n",
    "bfsMain(sc, sourceHeroID, targetHeroID)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.17"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}