Assignment 3 MapReduce Web Graph _ collab.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ByCSu-FhG5v7"
      },
      "source": [
        "# Assignment 3: MapReduce & Web Graph"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-e3cxxyFG_XW"
      },
      "source": [
        "## General guidelines\n",
        "\n",
        "This notebook contains considerable amount of code to help you complete this assignment. Your task is to implement any missing parts of the code and answer any questions (if exist) within this notebook. This will require understanding the existing code, may require reading about packages being used, reading additional resources, and maybe even going over your notes from class 😱\n",
        "\n",
        "**Evaluation and auto-grading**: Your submissions will be evaluated using both automatic and manual grading. Code parts for implementation are marked with a comment `# YOUR CODE HERE`, and usually followed by cell(s) containing automatic tests that evaluate the correctness of your answer. Staff will allow your notebook to **execute from start to finish for no more than 90 seconds**, then manually assess your submission. Any automatic tests that did not run due to your notebook timing out **will automatically receive 0 points**. The execution time excludes initial data download, which will already exist in the testing environment. The staff reserves the right to **modify any grade provided by the auto-grader** as well as to **execute additional tests not provided to you**. It is also important to note that **auto-graded cells only result in full or no credit**. In other words, you must pass all tests implemented in a test cell in order to get the credit for it, and passing some, but not all, of the tests in a test cell will not earn you any points for that cell. \n",
        "\n",
        "**Submission**: Upload a zip file **with your ID as the file name** containing the colab notebook and the GCP notebook to the assignment on Moodle. The names of the files should be  in the format:  012345678_colab.ipynb and 012345678_gcp.ipynb."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fscYA3wkHCLk"
      },
      "source": [
        "# Tasks\n",
        "\n",
        "In this assignment, we are going to use MapReduce in pySpark to scale up index creation. **The assignment consists of two parts, one in Google Colab and one in GCP**. In the first part (Colab), we will learn how to use MapReduce in PySpark, create an inverted index, and calculate PageRank. In the second part, we will use GCP to scale up index creation for the entire English Wikipedia. \n",
        "\n",
        "You will learn how to work with PySpark, write MapReduce code that executes on a cluster of machines, and experience the power of parallel processing. **All tasks must be completed through the use of PySpark's MapReduce**.\n",
        "\n",
        "To save you (and us) time, both in this assignment and in the final project, we are providing you with preprocessed files for the entire English wikipedia dump. Preprocessing was done, similar to Assignment #2, to extract article information from the XML, then remove the WikiMedia markup. The complete preprocessing code for one wiki dump can be seen [here](https://colab.research.google.com/drive/1xXc0YPU4BVd8CAjVNgICimW_d-Z7wryu?usp=sharing).\n",
        "\n",
        "\n",
        "**By the end of this assignment, you will have:**\n",
        "\n",
        "In Colab (65 Points)\n",
        "\n",
        "1. (20 Points) Processed articles' body to produce posting lists (using PySpark and MapReduce).\n",
        "2. (5 Points) Calculated document frequency (DF) in a  distributed fashion. \n",
        "3. (20 Points) Partitioned the posting lists and wrote them to disk in parallel.\n",
        "4. (20 Points) Constructed the links graph and calculated PageRank on it. \n",
        "All of these will be based on a small subset of articles from one wiki dump file.\n",
        "\n",
        "In GCP (35 Points):\n",
        "\n",
        "1. (10 Points) Successfully started a Dataproc cluster, setup cloud storage bucket, and properly initialized cluster nodes.\n",
        "2. (10 Points) Created an inverted index for all of English Wikipedia. \n",
        "3. (10 Points) Successfully computed PageRank for the entire English Wikipedia.\n",
        "4. (5 Points) Reporting: size of the data, size of index, time it took to build the index, time it took to compute PageRank, and the $ amount of credits it consumed.\n",
        "\n",
        "Bonus (10 points) if you implement PageRank in pure PySpark, i.e. without using the GraphFrames package, AND manage to complete 10 iterations of your algorithm on the entire English Wikipedia in less than an hour. \n",
        "\n",
        "\n",
        "\n",
        "\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "J0P4W63JvrT5"
      },
      "source": [
        "# Setup"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "59yMYAN1x3oh"
      },
      "source": [
        "## General imports\n",
        "\n",
        "The `inverted_index_colab` import requires the `inverted_index_colab.py` file. You should upload the file and then run this cell."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "id": "byNgReCYsmA9",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "f9beca45-059a-467b-a129-ea19bab2bfb1"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "[nltk_data] Downloading package stopwords to /root/nltk_data...\n",
            "[nltk_data]   Package stopwords is already up-to-date!\n"
          ]
        }
      ],
      "source": [
        "import sys\n",
        "from collections import Counter, OrderedDict\n",
        "import itertools\n",
        "from itertools import islice, count, groupby\n",
        "import pandas as pd\n",
        "import os\n",
        "import re\n",
        "from operator import itemgetter\n",
        "import nltk\n",
        "from nltk.stem.porter import *\n",
        "from nltk.corpus import stopwords\n",
        "from time import time\n",
        "from timeit import timeit\n",
        "from pathlib import Path\n",
        "import pickle\n",
        "import pandas as pd\n",
        "import numpy as np\n",
        "from google.cloud import storage\n",
        "\n",
        "import hashlib\n",
        "def _hash(s):\n",
        "    return hashlib.blake2b(bytes(s, encoding='utf8'), digest_size=5).hexdigest()\n",
        "\n",
        "nltk.download('stopwords')\n",
        "\n",
        "from inverted_index_colab import *"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "KRjOVgLBrfCs"
      },
      "source": [
        "## Installing, importing, and initializing PySpark\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GORmjMdasns1"
      },
      "source": [
        "The following code installs PySpark and its dependencies in Colab. In addition, we install GraphFrames, which is a PySpark package for dealing with graphs in a distributed fashion. Colab notebooks run on a single machine so we will work in local mode, i.e. there is no cluster of machines and both the master and worker processes run on a single machine. This will help us debug and iron out the code we will use in the second half of this assignment on an actual cluster mode in GCP. \n",
        "\n",
        "The installation in the next cell should take about 1 minute in a fresh environment. Don't worry about going  over the 90 seconds limit for the assignment because in our testing enviroment all of these requirements will already be met. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "id": "Xf84c2MsIUGk"
      },
      "outputs": [],
      "source": [
        ""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "kw33QgujFQez",
        "outputId": "2a627dc0-6dc7-4bbc-f368-5282070809f6"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "openjdk-8-jdk-headless is already the newest version (8u292-b10-0ubuntu1~18.04).\n",
            "0 upgraded, 0 newly installed, 0 to remove and 37 not upgraded.\n",
            "--2021-12-08 08:57:09--  https://repos.spark-packages.org/graphframes/graphframes/0.8.2-spark3.2-s_2.12/graphframes-0.8.2-spark3.2-s_2.12.jar\n",
            "Resolving repos.spark-packages.org (repos.spark-packages.org)... 13.226.124.87, 13.226.124.67, 13.226.124.24, ...\n",
            "Connecting to repos.spark-packages.org (repos.spark-packages.org)|13.226.124.87|:443... connected.\n",
            "HTTP request sent, awaiting response... 304 Not Modified\n",
            "File ‘/usr/local/lib/python3.7/dist-packages/pyspark/jars/graphframes-0.8.2-spark3.2-s_2.12.jar’ not modified on server. Omitting download.\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# These will already be installed in the testing environment so disregard the \n",
        "# amount of time (~1 minute) it takes to install. \n",
        "!pip install -q pyspark\n",
        "!pip install -U -q PyDrive\n",
        "!apt install openjdk-8-jdk-headless -qq\n",
        "!pip install -q graphframes\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "graphframes_jar = 'https://repos.spark-packages.org/graphframes/graphframes/0.8.2-spark3.2-s_2.12/graphframes-0.8.2-spark3.2-s_2.12.jar'\n",
        "spark_jars = '/usr/local/lib/python3.7/dist-packages/pyspark/jars'\n",
        "!wget -N -P $spark_jars $graphframes_jar"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {
        "id": "BkEErqJnsXyy"
      },
      "outputs": [],
      "source": [
        "import pyspark\n",
        "from pyspark.sql import *\n",
        "from pyspark.sql.functions import *\n",
        "from pyspark import SparkContext, SparkConf\n",
        "from pyspark.sql import SQLContext\n",
        "from pyspark.ml.feature import Tokenizer, RegexTokenizer\n",
        "from graphframes import *"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "id": "OgiSewKAsgKd"
      },
      "outputs": [],
      "source": [
        "# Initializing spark context\n",
        "# create a spark context and session\n",
        "conf = SparkConf().set(\"spark.ui.port\", \"4050\")\n",
        "sc = pyspark.SparkContext(conf=conf)\n",
        "sc.addPyFile(str(Path(spark_jars) / Path(graphframes_jar).name))\n",
        "spark = SparkSession.builder.getOrCreate()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "A9fjw2J-s87v"
      },
      "source": [
        "You can easily check the current version and get the link of the web interface. In the Spark UI, you can monitor the progress of your job and debug the performance bottlenecks (if your Colab is running with a **local runtime**)."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 222
        },
        "id": "PUyhQAlBs8qL",
        "outputId": "94ec94bc-7c9f-4aeb-9cfb-cb07b8ef8b28"
      },
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/html": [
              "\n",
              "            <div>\n",
              "                <p><b>SparkSession - in-memory</b></p>\n",
              "                \n",
              "        <div>\n",
              "            <p><b>SparkContext</b></p>\n",
              "\n",
              "            <p><a href=\"http://4897c14e912c:4050\">Spark UI</a></p>\n",
              "\n",
              "            <dl>\n",
              "              <dt>Version</dt>\n",
              "                <dd><code>v3.2.0</code></dd>\n",
              "              <dt>Master</dt>\n",
              "                <dd><code>local[*]</code></dd>\n",
              "              <dt>AppName</dt>\n",
              "                <dd><code>pyspark-shell</code></dd>\n",
              "            </dl>\n",
              "        </div>\n",
              "        \n",
              "            </div>\n",
              "        "
            ],
            "text/plain": [
              "<pyspark.sql.session.SparkSession at 0x7f87bb056150>"
            ]
          },
          "metadata": {},
          "execution_count": 6
        }
      ],
      "source": [
        "spark"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "19EG5AvVru1J"
      },
      "source": [
        "## Copy some wiki data"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ofyAoTHttjHE"
      },
      "source": [
        "As mentioned above, all wiki dumps were preprocessed and placed in a shared google storage bucket. To access the files in Colab, you will first have to authenticate with google storage, and then copy the data to your local environment. In GCP, the storage will already be mounted on the cluster machines and we will show you how to access it. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {
        "id": "DkdjBm5DOVWX"
      },
      "outputs": [],
      "source": [
        "# Authenticate your user\n",
        "# The authentication should be done with the email connected to your GCP account\n",
        "from google.colab import auth\n",
        "auth.authenticate_user()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "vDuikxFbN9B7",
        "outputId": "cb24a05c-2558-4361-b513-63e92c02dee7"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Updated property [core/project].\n",
            "\u001b[1;33mWARNING:\u001b[0m You do not appear to have access to project [core-period-321814] or it does not exist.\n",
            "mkdir: cannot create directory ‘wikidumps’: File exists\n",
            "Copying gs://wikidata_preprocessed/multistream1_preprocessed.parquet...\n",
            "\\ [1 files][316.7 MiB/316.7 MiB]                                                \n",
            "Operation completed over 1 objects/316.7 MiB.                                    \n"
          ]
        }
      ],
      "source": [
        "# Copy one wikidumps files \n",
        "import os\n",
        "from pathlib import Path\n",
        "from google.colab import auth\n",
        "\n",
        "project_id = 'core-period-321814'\n",
        "!gcloud config set project {project_id}\n",
        "\n",
        "data_bucket_name = 'wikidata_preprocessed'\n",
        "try:\n",
        "    if os.environ[\"wikidata_preprocessed\"] is not None:\n",
        "        pass  \n",
        "except:\n",
        "      !mkdir wikidumps\n",
        "      !gsutil cp gs://{data_bucket_name}/multistream1_preprocessed.parquet \"wikidumps/\" \n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_GU4djg6yizh"
      },
      "source": [
        "# Processing wikipedia"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "uF9ywiIC-sAX"
      },
      "source": [
        "Now that we completed the setup and have some data in our local environment, we are ready to process it using PySpark. "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tkEUicVTzDZd"
      },
      "source": [
        "## A 2-minute intro to PySpark"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6GVQMbCtzWI2"
      },
      "source": [
        "One of the fundamental data structures in spark is a **Resillient Distributed Dataset (RDD)**. It is an immutable distributed collections of objects that are partitioned across machines, and thus making parallel processing natural and easy. RDD's support two types of operations: *transformations* (e.g. selecting a subset of fields from each element or projecting each element from A->B), and *actions* (e.g. save to disk or `collect` elements into the master node's memory). Watch out for `collect` operations that send too much data to the master node, it will crash without a trace because of running out of memory. \n",
        "\n",
        "Spark employs **lazy-evaluation** (like many of us...), meaning that it accumulates things we ask it to do into an execution plan, but doesn't do any calculation until a result is actually needed. When a result is needed? when we save something to disk, ask spark to show the data, or use other actions. But make no mistakes, Spark is not a slacker. It employs lazy-evaluation because it allows optimizations that wouldn't be possible otherwise and it often saves users from unnecessary data reads, tranfers across machines, and other costly operations. As we will see in a bit, the RDD data structure lends itself to simple and efficient MapReduce operations."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "5aQwbFGxPb4D"
      },
      "source": [
        "Let's look at our data before transforming it to RDD."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 9,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "24AI2jXgOuVN",
        "outputId": "6c6d9e26-55d2-4ed5-d33e-ff2ca1ce37e4"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+---+--------------------+--------------------+--------------------+\n",
            "| id|               title|                text|         anchor_text|\n",
            "+---+--------------------+--------------------+--------------------+\n",
            "| 12|           Anarchism|'''Anarchism''' i...|[{23040, politica...|\n",
            "| 25|              Autism|'''Autism''' is a...|[{492271, Clinica...|\n",
            "| 39|              Albedo|thumb|upright=1.3...|[{679294, diffuse...|\n",
            "|290|                   A|'''A''', or '''a'...|[{290, See below}...|\n",
            "|303|             Alabama|'''Alabama''' () ...|[{351590, Yellowh...|\n",
            "|305|            Achilles|thumb|260px|Ancie...|[{1076007, potter...|\n",
            "|307|     Abraham Lincoln|'''Abraham Lincol...|[{1827174, Alexan...|\n",
            "|308|           Aristotle|'''Aristotle''' (...|[{1389981, bust},...|\n",
            "|309|An American in Paris|'''''An American ...|[{13066, George G...|\n",
            "|316|Academy Award for...|The '''Academy Aw...|[{39842, Academy ...|\n",
            "|324|      Academy Awards|The '''Academy Aw...|[{649481, film in...|\n",
            "|330|             Actrius|'''''Actresses'''...|[{5282, Catalan},...|\n",
            "|332|     Animalia (book)|'''''Animalia''''...|[{2511084, Graeme...|\n",
            "|334|International Ato...|'''International ...|[{25453985, atomi...|\n",
            "|336|            Altruism|thumb|Giving alms...|[{657573, alms}, ...|\n",
            "|339|            Ayn Rand|'''Alice O'Connor...|[{24320051, St. P...|\n",
            "|340|        Alain Connes|'''Alain Connes''...|[{1201522, Dragui...|\n",
            "|344|          Allan Dwan|'''Allan Dwan''' ...|[{64646, Toronto}...|\n",
            "|358|             Algeria|'''Algeria''', of...|[{803, Arabic}, {...|\n",
            "|359|List of Atlas Shr...|This is a list of...|[{339, Ayn Rand},...|\n",
            "+---+--------------------+--------------------+--------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "from pathlib import Path \n",
        "import os\n",
        "\n",
        "try:\n",
        "    if os.environ[\"wikidata_preprocessed\"] is not None:\n",
        "      path = os.environ[\"wikidata_preprocessed\"]+\"/wikidumps/*\"\n",
        "except:\n",
        "      path = \"wikidumps/*\"\n",
        "\n",
        "parquetFile = spark.read.parquet(path)\n",
        "parquetFile.show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 10,
      "metadata": {
        "id": "KMcsrc8RO9kn"
      },
      "outputs": [],
      "source": [
        "# take the 'text' and 'id' or the first 1000 rows and create an RDD from it\n",
        "doc_text_pairs = parquetFile.limit(1000).select(\"text\", \"id\").rdd"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "__g69rcETm4w"
      },
      "source": [
        "## Word counts"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-zUibXvjWb6Q"
      },
      "source": [
        "### Term frequency"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "UrBuHUHmA5pP"
      },
      "source": [
        "**YOUR TASK (10 POINTS):** Complete the implementation of `word_count`, which calculates the term-frequency (tf) for each word in a single document.  map function. The functions gets the text and the wiki id, calculates the tf and returns list of tuples in the form of (token, (wiki_id, tf))."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 11,
      "metadata": {
        "id": "9GmPedzexLyt"
      },
      "outputs": [],
      "source": [
        "english_stopwords = frozenset(stopwords.words('english'))\n",
        "corpus_stopwords = ['category', 'references', 'also', 'links', 'extenal', 'see', 'thumb']\n",
        "RE_WORD = re.compile(r\"\"\"[\\#\\@\\w](['\\-]?\\w){2,24}\"\"\", re.UNICODE)\n",
        "\n",
        "all_stopwords = english_stopwords.union(corpus_stopwords)\n",
        "\n",
        "def word_count(text, id):\n",
        "  ''' Count the frequency of each word in `text` (tf) that is not included in \n",
        "  `all_stopwords` and return entries that will go into our posting lists. \n",
        "  Parameters:\n",
        "  -----------\n",
        "    text: str\n",
        "      Text of one document\n",
        "    id: int\n",
        "      Document id\n",
        "  Returns:\n",
        "  --------\n",
        "    List of tuples\n",
        "      A list of (token, (doc_id, tf)) pairs \n",
        "      for example: [(\"Anarchism\", (12, 5)), ...]\n",
        "  '''\n",
        "  tokens = [token.group() for token in RE_WORD.finditer(text.lower())]\n",
        "  # YOUR CODE HERE\n",
        "  \n",
        "  \n",
        "  dic = {}\n",
        "  dictlist=[]\n",
        "  for i in tokens:\n",
        "    if i in all_stopwords:\n",
        "      continue\n",
        "    if i not in dic:\n",
        "      dic[i]=(id,1)\n",
        "    else:\n",
        "      dic[i]=(id,dic[i][1]+1)\n",
        "  return list(dic.items())\n",
        "  \n",
        "  \n",
        "  \"\"\"\n",
        "  rd = sc.parallelize(tokens)\n",
        "  rd = rd.filter(lambda x : x not in all_stopwords)\n",
        "  mapped = rd.map(lambda x:(x,1))\n",
        "  reduced = mapped.reduceByKey(lambda x,y:x+y)\n",
        "  final = reduced.mapValues(lambda x:(id,x) )\n",
        "  result = final.collect()\n",
        "  return result\n",
        "  \"\"\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 12,
      "metadata": {
        "id": "RLqHPKUvbT8R"
      },
      "outputs": [],
      "source": [
        "# test for word count\n",
        "text = \"This is a test to check the words count function actually count words\"\n",
        "id = 0\n",
        "word_count_results = word_count(text, id)\n",
        "assert word_count_results[0][0] == 'test'\n",
        "assert len(word_count_results) == 6\n",
        "assert word_count_results[2][1][1] == word_count_results[2][1][1]\n",
        "assert type(word_count_results[0][0]) == str\n",
        "assert type(word_count_results[0][1]) == tuple\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "MNC7QlO8KZL_"
      },
      "source": [
        "PySpark has a `map` trasformation that, just like we learned in class about Map in MapReduce, executes a given function on each element of an  RDD. It also has a `flatMap()` transformation that executes a given function on each element of an RDD, but afterwards flattens the results to create one big \"list\" instead multiple nested lists. For example, `flatMap` on the above `word_count` function will concatenate the tokens from multiple document into one big list rather than keeping separate list of tokens for each document like a regular `map` would produce. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 13,
      "metadata": {
        "id": "g3DUXXzxsy-B"
      },
      "outputs": [],
      "source": [
        "word_counts = doc_text_pairs.flatMap(lambda x: word_count(x[0], x[1]))\n",
        "\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "cwC3pX1ss1en"
      },
      "source": [
        "After `flatMap` is executed, the counts for a given token are likely to be scattered accross multiple machines/workers in the cluster. In order to create the posting lists for each token, we need a reduce operation that collects them into one list and sorts them by the document id (wiki_id). \n",
        "\n",
        "Spark has several ways of reducing or aggergating values, but we will use `groupByKey`, which like Reduce in MapReduce, collects all the values for a given key into an iterable list of the values associated with that key. The result of `groupByKey` is an RDD with (key, value*s*) pairs. Then, we can call `map` to execute a function on each pair in that RDD or call `mapValues` to execute a function only on the values iterable.\n",
        "\n",
        "**YOUR TASK (5 POINTS):** Complete the implementation of `reduce_word_counts`, which operates on the pairs returned by `word_count`. The functions gets a list of values (unsorted posting list) and sorts it."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 14,
      "metadata": {
        "id": "1yFhHQBZHIBb"
      },
      "outputs": [],
      "source": [
        "def reduce_word_counts(unsorted_pl):\n",
        "  ''' Returns a sorted posting list by wiki_id.\n",
        "  Parameters:\n",
        "  -----------\n",
        "    unsorted_pl: list of tuples\n",
        "      A list of (wiki_id, tf) tuples \n",
        "  Returns:\n",
        "  --------\n",
        "    list of tuples\n",
        "      A sorted posting list.\n",
        "  '''\n",
        "  # YOUR CODE HERE\n",
        " \n",
        "  sorted_pl = sorted(unsorted_pl)\n",
        "  return sorted_pl"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 15,
      "metadata": {
        "id": "-VWkWnUO_QKd"
      },
      "outputs": [],
      "source": [
        "postings = word_counts.groupByKey().mapValues(reduce_word_counts)\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 16,
      "metadata": {
        "id": "6zbBgNIM46OX"
      },
      "outputs": [],
      "source": [
        "## test reduce_word_count\n",
        "token, posting_list = postings.take(1)[0]\n",
        "assert type(token) == str\n",
        "assert type(posting_list) == list\n",
        "doc_id, tf = zip(*posting_list)\n",
        "assert type(doc_id[0]) == int\n",
        "assert type(tf[0]) == int\n",
        "assert np.diff(doc_id).min() > 0 \n",
        "assert np.min(tf) > 0 \n",
        "assert postings.count() == 153014\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "3LiIJMOjWnsB"
      },
      "source": [
        "### Document frequency and filteration"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zrWoS-kHvhmM"
      },
      "source": [
        "Next, we will filter out rare words, words that appear in 10 or fewer documents (when working on the entire corpus, we will increase this threshold to a minimum of 50 documents). "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 17,
      "metadata": {
        "id": "EnrhMaR0digd"
      },
      "outputs": [],
      "source": [
        "postings_filtered = postings.filter(lambda x: len(x[1])>10)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-bclZ2cmFRjq"
      },
      "source": [
        "Now, we would like to calculate the df for each word.\n",
        "\n",
        "**YOUR TASK (5 POINTS):** Complete the implementation of `calculate_df` that takes a posting list RDD and returns an RDD where each element is a (token, df) pair. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 18,
      "metadata": {
        "id": "J_-UoG2CGj4o"
      },
      "outputs": [],
      "source": [
        "import math\n",
        "\n",
        "def f(x):\n",
        "  print(type(x))\n",
        "  s =0\n",
        "  for i in x[1]:\n",
        "    s+=i[1]\n",
        " \n",
        "  return x\n",
        " \n",
        "def calculate_df(postings):\n",
        "  ''' Takes a posting list RDD and calculate the df for each token.\n",
        "  Parameters:\n",
        "  -----------\n",
        "    postings: RDD\n",
        "      An RDD where each element is a (token, posting_list) pair.\n",
        "  Returns:\n",
        "  --------\n",
        "    RDD\n",
        "      An RDD where each element is a (token, df) pair.\n",
        "  '''\n",
        "\n",
        "  rd = postings.mapValues(lambda x:(len(x)))\n",
        "  \n",
        "  \"\"\"\n",
        "  end = []\n",
        "  for t in postings.collect():\n",
        "    token = t[0]\n",
        "    postinglist = t[1]\n",
        "    s=0\n",
        "    for i in postinglist :\n",
        "      s+=1\n",
        "    tuple1 = (token,s)\n",
        "    end.append(tuple1)\n",
        "  rd = sc.parallelize(end)\n",
        "  \"\"\"\n",
        "  return rd\n",
        "  "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 19,
      "metadata": {
        "id": "ddVcByPLCaP_"
      },
      "outputs": [],
      "source": [
        "# global statistics\n",
        "w2df = calculate_df(postings_filtered)\n",
        "# test calculate_df types\n",
        "token, df = w2df.take(1)[0]\n",
        "assert type(token) == str\n",
        "assert type(df) == int\n",
        "# test min/max df values\n",
        "# collectAsMap collects the results to the master node's memory as a dictionary.\n",
        "# we know it's not so big so this is okay.\n",
        "w2df_dict = w2df.collectAsMap()\n",
        "assert np.min(list(w2df_dict.values())) == 11\n",
        "assert np.max(list(w2df_dict.values())) == 819\n",
        "# test select words\n",
        "assert w2df_dict['first'] == 805\n",
        "assert w2df_dict['many'] == 670\n",
        "assert w2df_dict['used'] == 648"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "5z1wMOiFTyGJ"
      },
      "source": [
        "## Partitioning and writing the index"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "q3a2sC0WytzO"
      },
      "source": [
        "Now that we have the postings information we need to write it to disk. We would like to partition the tokens and posting lists into buckets so when we get a query, we know exactly in which bucket to look for its posting list. This can also help parallelize retrieval from the index if we have multiple machines answering queries. \n",
        "\n",
        "Here, you are asked to do this partitioning/bucketing through another round of map and reduce operations, where the last step should be to write out each bucket into a separate inverted index on disk. Use the staff-provided function of  `write_a_posting_list` implemented in inverted_index_colab.py for the InvertedIndex class. Eventually, this partitioning and writing process needs to return information about where each posting list is located (file name and offset) for a token. \n",
        "\n",
        "**YOUR TASK (20 POINTS):** Complete the implementation of `partition_postings_and_write` that partitions the posting list, writes out each bucket, and returns information about the location on disk of each posting list."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 20,
      "metadata": {
        "id": "MvZDJk0bu9V9"
      },
      "outputs": [],
      "source": [
        "NUM_BUCKETS = 124\n",
        "def token2bucket_id(token):\n",
        "  return int(_hash(token),16) % NUM_BUCKETS\n",
        "\n",
        "def partition_postings_and_write(postings):\n",
        "  ''' A function that partitions the posting lists into buckets, writes out \n",
        "  all posting lists in a bucket to disk, and returns the posting locations for \n",
        "  each bucket. Partitioning should be done through the use of `token2bucket` \n",
        "  above. Writing to disk should use the function  `write_a_posting_list`, a \n",
        "  static method implemented in inverted_index_colab.py under the InvertedIndex \n",
        "  class. \n",
        "  Parameters:\n",
        "  -----------\n",
        "    postings: RDD\n",
        "      An RDD where each item is a (w, posting_list) pair.\n",
        "  Returns:\n",
        "  --------\n",
        "    RDD\n",
        "      An RDD where each item is a posting locations dictionary for a bucket. The\n",
        "      posting locations maintain a list for each word of file locations and \n",
        "      offsets its posting list was written to. See `write_a_posting_list` for \n",
        "      more details.\n",
        "  '''\n",
        "  # YOUR CODE HERE\n",
        "  \"\"\"lst = []\n",
        "  dic = {}\n",
        "  for i in range(124):\n",
        "    dic[i]=[]\n",
        "  for w,posting_list in postings.collect():\n",
        "    bucket_id = token2bucket_id(w)\n",
        "    tuple1 = (w , posting_list)\n",
        "    dic[bucket_id].append(tuple1)\n",
        "  \n",
        "  for item in dic.items():\n",
        "    dictionary = InvertedIndex.write_a_posting_list(item)\n",
        "    lst.append(dictionary)\n",
        "  return sc.parallelize(lst)\"\"\"\n",
        "  rd = sc.parallelize(postings.collect())\n",
        "  rd = rd.map(lambda x:(token2bucket_id(x[0]),[(x[0],x[1])]))\n",
        "  rd = rd.reduceByKey(lambda x,y:x+y)\n",
        "  rd = rd.map(lambda x :InvertedIndex.write_a_posting_list(x))\n",
        "  return rd\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "fohEJFfR3QiE"
      },
      "source": [
        "The next cell uses the `collect` operator to aggregate the posting location information from the multiple buckets/indices into a single dictionary. We know in this case that the posting locations info is relatively small so this is not going to crash our master node. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 21,
      "metadata": {
        "id": "vxCMgAlRDIac",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "79195237-02e7-4cd9-f71c-9ba0441d0159"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "defaultdict(<class 'list'>, {'sceptical': [('74_000.bin', 0)], 'philosopher': [('74_000.bin', 72)], 'exploitation': [('74_000.bin', 894)], 'gave': [('74_000.bin', 1044)], 'feminist': [('74_000.bin', 2832)], 'followed': [('74_000.bin', 2988)], 'aspects': [('74_000.bin', 4998)], 'expanded': [('74_000.bin', 5730)], 'signals': [('74_000.bin', 6630)], 'lead': [('74_000.bin', 6930)], 'tool': [('74_000.bin', 8316)], 'connection': [('74_000.bin', 8724)], 'unable': [('74_000.bin', 9522)], 'arguing': [('74_000.bin', 10326)], 'communication': [('74_000.bin', 10614)], 'differences': [('74_000.bin', 11202)], 'availability': [('74_000.bin', 12036)], 'hugh': [('74_000.bin', 12318)], 'societal': [('74_000.bin', 12708)], 'measurements': [('74_000.bin', 12792)], 'deciduous': [('74_000.bin', 13074)], '9th': [('74_000.bin', 13158)], 'slowly': [('74_000.bin', 13632)], 'abbreviations': [('74_000.bin', 14208)], 'ordinal': [('74_000.bin', 14304)], 'losing': [('74_000.bin', 14388)], 'popularized': [('74_000.bin', 14832)], 'amendment': [('74_000.bin', 15000)], 'suppress': [('74_000.bin', 15216)], 'influx': [('74_000.bin', 15408)], 'plant': [('74_000.bin', 15582)], 'hall': [('74_000.bin', 16284)], 'festival': [('74_000.bin', 17262)], 'championship': [('74_000.bin', 17904)], 'infrastructure': [('74_000.bin', 18138)], 'referendum': [('74_000.bin', 18540)], 'failed': [('74_000.bin', 18672)], '102': [('74_000.bin', 19752)], 'morris': [('74_000.bin', 19962)], '274': [('74_000.bin', 20250)], 'brightest': [('74_000.bin', 20340)], 'heroic': [('74_000.bin', 20430)], 'allies': [('74_000.bin', 20670)], 'carefully': [('74_000.bin', 21162)], 'samuel': [('74_000.bin', 21444)], 'cousin': [('74_000.bin', 22020)], 'realizing': [('74_000.bin', 22440)], 'defending': [('74_000.bin', 22518)], 'intact': [('74_000.bin', 22746)], 'grant': [('74_000.bin', 22962)], \"god's\": [('74_000.bin', 23496)], 'protestants': [('74_000.bin', 23760)], 'lighter': [('74_000.bin', 23922)], 'pursuing': [('74_000.bin', 24156)], 'visitor': [('74_000.bin', 24300)], 'tenor': [('74_000.bin', 24444)], 'laura': [('74_000.bin', 24564)], 'nichols': [('74_000.bin', 24714)], '45th': [('74_000.bin', 24792)], 'nemesis': [('74_000.bin', 24858)], 'longevity': [('74_000.bin', 24930)], 'lenin': [('74_000.bin', 25008)], 'reacting': [('74_000.bin', 25110)], 'ottomans': [('74_000.bin', 25182)], 'coins': [('74_000.bin', 25308)], 'reign': [('74_000.bin', 25590)], 'countryside': [('74_000.bin', 26316)], 'köppen': [('74_000.bin', 26508)], 'clarified': [('74_000.bin', 26676)], 'anymore': [('74_000.bin', 26760)], 'methodology': [('74_000.bin', 26832)], 'responses': [('74_000.bin', 26970)], 'conferences': [('74_000.bin', 27180)], 'covenant': [('74_000.bin', 27324)], 'warlord': [('74_000.bin', 27408)], 'episodes': [('74_000.bin', 27480)], 'kid': [('74_000.bin', 27654)], 'metres': [('74_000.bin', 27720)], 'thrace': [('74_000.bin', 27894)], 'trunk': [('74_000.bin', 28038)], '1780': [('74_000.bin', 28146)], 'genealogy': [('74_000.bin', 28326)], 'penn': [('74_000.bin', 28434)], 'henderson': [('74_000.bin', 28524)], 'nouns': [('74_000.bin', 28632)], 'coloured': [('74_000.bin', 28740)], 'spiral': [('74_000.bin', 28860)], 'canary': [('74_000.bin', 29010)], 'chemically': [('74_000.bin', 29100)], 'defendants': [('74_000.bin', 29226)], 'corrosion': [('74_000.bin', 29298)], 'rocky': [('74_000.bin', 29364)], 'precipitated': [('74_000.bin', 29634)], 'half-life': [('74_000.bin', 29754)], 'tips': [('74_000.bin', 29856)], 'conditional': [('74_000.bin', 29934)], 'neon': [('74_000.bin', 30048)], 'indies': [('74_000.bin', 30114)], 'thomson': [('74_000.bin', 30294)], '1649': [('74_000.bin', 30432)], 'elastic': [('74_000.bin', 30510)], 'ethel': [('74_000.bin', 30582)], 'calligraphy': [('74_000.bin', 30648)], 'jurists': [('74_000.bin', 30720)], 'canons': [('74_000.bin', 30786)], 'sicilian': [('74_000.bin', 30864)], 'voiced': [('74_000.bin', 30960)], 'sits': [('74_000.bin', 31128)], 'takeoff': [('74_000.bin', 31248)], 'exhaustion': [('74_000.bin', 31362)], '640': [('74_000.bin', 31428)], 'federico': [('74_000.bin', 31506)], '1549': [('74_000.bin', 31572)], '1619': [('74_000.bin', 31662)], 'vic': [('74_000.bin', 31746)], 'augustinian': [('74_000.bin', 31824)]})\n"
          ]
        }
      ],
      "source": [
        "# test partitioning for the different buckets (10 points)\n",
        "posting_locs_list = partition_postings_and_write(postings_filtered).collect()\n",
        "print(posting_locs_list[0])\n",
        "exit()\n",
        "assert len(posting_locs_list) == 124\n",
        "assert len(list(Path('.').glob('*.bin'))) == 124\n",
        "t = [type(pl) for pl in posting_locs_list]\n",
        "pl_lens = [len(pl) for pl in posting_locs_list]\n",
        "assert np.max(pl_lens) == 141\n",
        "assert np.min(pl_lens) == 91 \n",
        "assert int(np.median(pl_lens)) == 116"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 22,
      "metadata": {
        "id": "C-h5dgtF-TeL"
      },
      "outputs": [],
      "source": [
        "# merge the posting locations into a single dict and run more tests (5 points)\n",
        "super_posting_locs = defaultdict(list)\n",
        "for posting_loc in posting_locs_list:\n",
        "  for k, v in posting_loc.items():\n",
        "    super_posting_locs[k].extend(v)\n",
        "\n",
        "assert super_posting_locs['try'][0][0][:3] == '105'\n",
        "assert super_posting_locs['political'][0][0][:2] == '83'\n",
        "assert 'hi' not in super_posting_locs\n",
        "assert len(super_posting_locs) == 14524"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "xBpxtFW7HeAT"
      },
      "source": [
        "Putting everything together (posting locations, df) and test that the resulting index is correct. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 23,
      "metadata": {
        "id": "VKrew8cUAyTz"
      },
      "outputs": [],
      "source": [
        "# Create inverted index instance\n",
        "inverted = InvertedIndex()\n",
        "# Adding the posting locations dictionary to the inverted index\n",
        "inverted.posting_locs = super_posting_locs\n",
        "# Add the token - df dictionary to the inverted index\n",
        "inverted.df = w2df_dict\n",
        "# write the global stats out\n",
        "inverted.write_index('.', 'index')"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 24,
      "metadata": {
        "id": "j2_qt2IPkxqz"
      },
      "outputs": [],
      "source": [
        "TUPLE_SIZE = 6       \n",
        "TF_MASK = 2 ** 16 - 1 # Masking the 16 low bits of an integer\n",
        "from contextlib import closing\n",
        "\n",
        "def read_posting_list(inverted, w):\n",
        "  with closing(MultiFileReader()) as reader:\n",
        "    locs = inverted.posting_locs[w]\n",
        "    b = reader.read(locs, inverted.df[w] * TUPLE_SIZE)\n",
        "    posting_list = []\n",
        "    for i in range(inverted.df[w]):\n",
        "      doc_id = int.from_bytes(b[i*TUPLE_SIZE:i*TUPLE_SIZE+4], 'big')\n",
        "      tf = int.from_bytes(b[i*TUPLE_SIZE+4:(i+1)*TUPLE_SIZE], 'big')\n",
        "      posting_list.append((doc_id, tf))\n",
        "    return posting_list"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 25,
      "metadata": {
        "id": "VF6gfVFSIzM6"
      },
      "outputs": [],
      "source": [
        "## test the final index\n",
        "pl = read_posting_list(inverted, 'clarity')\n",
        "assert pl[-3] == (2204, 2)\n",
        "assert len(pl) == 18"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "I2oCcQwP5zS_"
      },
      "source": [
        "## PageRank"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "U9gMGQIohFOz"
      },
      "source": [
        "In this section, your task is to compute PageRank for wiki articles using the anchor text we extracted from the MediaWiki markdown. Let's select the 'id' and 'anchor_text' fields into a new RDD:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 26,
      "metadata": {
        "id": "pUUdHGbCOr3C"
      },
      "outputs": [],
      "source": [
        "pages_links = spark.read.parquet(path).limit(1000).select(\"id\", \"anchor_text\").rdd"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IVhmHraRIHec"
      },
      "source": [
        "**YOUR TASK (20 POINTS):** Complete the implementation of `generate_graph`, which generates the graph of internal wikipedia links (wiki articles pointing to other wiki articles). Use PySpark to compute the vertices (pages) and edges (links) of the graph. Multiple links from page A to page B need to be represented by a single edge (edges are not weighted). You may use the built-in function `distinct` of the RDD object to identify distinct elements, but be mindful of its impact on computation time. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 27,
      "metadata": {
        "id": "YFRogCnlHIGR"
      },
      "outputs": [],
      "source": [
        "from pyspark.sql import SparkSession, types\n",
        "\n",
        "def f (arr):\n",
        "  dic = {}\n",
        "  for i in arr :\n",
        "    dic[i[0]]=(i[0],i[1])\n",
        "  \n",
        "  return list(dic.values())\n",
        "def generate_graph(pages):\n",
        "  ''' Compute the directed graph generated by wiki links.\n",
        "  Parameters:\n",
        "  -----------\n",
        "    pages: RDD\n",
        "      An RDD where each row consists of one wikipedia articles with 'id' and \n",
        "      'anchor_text'.\n",
        "  Returns:\n",
        "  --------\n",
        "    edges: RDD\n",
        "      An RDD where each row represents an edge in the directed graph created by\n",
        "      the wikipedia links. The first entry should the source page id and the \n",
        "      second entry is the destination page id. No duplicates should be present. \n",
        "    vertices: RDD\n",
        "      An RDD where each row represents a vetrix (node) in the directed graph \n",
        "      created by the wikipedia links. No duplicates should be present. \n",
        "  '''\n",
        "  # YOUR CODE HERE\n",
        "  \n",
        "  #vertices\n",
        "  vs1 = pages.map(lambda x:x[0])\n",
        "  vs2 = pages.map(lambda x:x[1]).flatMap(lambda x:x)\n",
        "  vs2 = vs2.map(lambda x:x[0])\n",
        "  vs = vs1.union(vs2)\n",
        "  vs = vs.distinct()\n",
        "  \n",
        "  \n",
        "  #edges\n",
        "  ds1 =pages.map(lambda x:f(x[1]))\n",
        "  ds1 = ds1.flatMap(lambda x:x)\n",
        "  \n",
        "  vertices = spark.createDataFrame(vs,StringType())\n",
        "  edges = spark.createDataFrame(ds1)\n",
        "  return edges, vertices"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 28,
      "metadata": {
        "id": "ajIiUP8CCY51"
      },
      "outputs": [],
      "source": [
        "# test graph construction efficiency (10 points)\n",
        "t_start = time()\n",
        "# construct the graph for a small sample of (1000) pages\n",
        "edges, vertices = generate_graph(pages_links)\n",
        "# time the actual execution\n",
        "v_cnt, e_cnt = vertices.count(), edges.count()\n",
        "graph_const_time = time() - t_start\n",
        "# test that the implementation is efficient (<20 seconds)\n",
        "#assert graph_const_time < 20"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 29,
      "metadata": {
        "id": "tb6ainuknU6z"
      },
      "outputs": [],
      "source": [
        "# test that we get the right number of vertices (5 points)\n",
        "assert v_cnt == 114966"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 30,
      "metadata": {
        "id": "ruKfLiFXnqcY",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "6b56b4d7-4099-49ee-b416-ad22047cc72c"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "183727\n"
          ]
        }
      ],
      "source": [
        "# test that we get the right number of edges (5 points)\n",
        "print(e_cnt)\n",
        "assert e_cnt == 183727"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yAx3CmIqOcLn"
      },
      "source": [
        "Let's compute PageRank in a distributed fashion (using PySpark's GraphFrame package) and show the top 20 pages."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "XQl-tJSFNV_g",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "f6420135-8b94-4ff7-cfcd-9f9fa2f12476"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "['src', 'dst']\n"
          ]
        }
      ],
      "source": [
        "edgesDF = edges.toDF('src', 'dst').repartition(4, 'src')\n",
        "print(edgesDF.columns)\n",
        "verticesDF = vertices.toDF('id').repartition(4, 'id')\n",
        "g = GraphFrame(verticesDF, edgesDF)\n",
        "pr_results = g.pageRank(resetProbability=0.15, maxIter=10)\n",
        "pr = pr_results.vertices.select(\"id\", \"pagerank\")\n",
        "pr = pr.sort(col('pagerank').desc())\n",
        "pr.repartition(1).write.csv('pr', compression=\"gzip\")\n",
        "pr.show()"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        ""
      ],
      "metadata": {
        "id": "xon_8pDTs4bD"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "x0iXbQIstzvW"
      },
      "source": [
        "**YOUR TASK (5 POINTS)**: Complete the implmentation of `get_top_pr_page_titles` to return the titles a few of the top pages as ranked by PageRank and printed above. To get the title for each page id listed above, navigate to the url https://en.wikipedia.org/?curid=wiki_id_here by replacing the `wiki_id_here` part with the page id. For example, if you navigate to https://en.wikipedia.org/?curid=25507 for the page with id 25507 you'd see that its title is 'Roman Empire'."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vhZVTSpnxQaj"
      },
      "outputs": [],
      "source": [
        "def get_top_pr_page_titles():\n",
        "  ''' Returns the title of the first, fourth, and fifth pages as ranked about \n",
        "      by PageRank.\n",
        "  Returns:\n",
        "  --------\n",
        "    list of three strings.\n",
        "  '''\n",
        "  # YOUR CODE HERE\n",
        "  return \"Catholic Church\",\"Ancient Greek\",\"Aristotle\"\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "nli6ETKzyPn6"
      },
      "outputs": [],
      "source": [
        "assert [_hash(x.lower()) for x in get_top_pr_page_titles()] == \\\n",
        "  ['52bb7dbc02', '3e920ea944', '3d007b19ee']"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "d4V85PGkP3We"
      },
      "source": [
        "Do the top ranked pages make sense to you? Why do you think these pages received such a high scores from PageRank?"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "OkDnmqAur3Ph"
      },
      "source": [
        "# GCP\n",
        "\n",
        "The second part of the assignment involves running index creation in GCP and reporting results in this notebook. Follow [this](https://docs.google.com/document/d/1HTZVDMQ1gyeq8dlyzmomOkxjSo1Td76PsCGlFkQohps/edit?usp=sharing) step-by-step guide in order to set up your cluster environment on GCP so you can execute pyspark code on a cluster of machines. "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Tq-ilXaFLFVh"
      },
      "source": [
        "The code to the GCP notebook is in the moodle under assignment 3. You will have to copy the code you completed here in the relevant sections.\n",
        "\n",
        "In the current notebook we used MapReduce to create the index for only one wiki dump. Using GCP, we will be able to create the index on the entire english wikipedia. The GCP part of this assignment will be the first time we will work with the entire corpus. You will need to follow the notebook, complete the code as you did here (the same code you completed in this notebook can be used in the GCP notebook). \n",
        "At the end of the GCP assignment you will have questions that you will need to answer - this are not the tests we used so far, so you should follow the instructions in the notebook."
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "assignment_3.ipynb",
      "provenance": []
    },
    "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.7.6"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}