Ejercicios_4_Spark.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "Considerando el pagelink_sample.csv, usando una representación de grafos realizar una función genérica que nos permita calcular los contenidos que se encuentran a un grado de separación de cualquier identificador de contenido de la wikipedia. Mostrar el funcionamiento de la implementación con algún contenido incluido en el set de datos (⭐⭐⭐)"
      ],
      "metadata": {
        "id": "4ptW6IFXZZeY"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cPCem4uvZVaX",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "511de104-b52e-4162-f315-2f1d7db79b4a"
      },
      "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[31m2.8 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=311317145 sha256=7422fb3ab6b3c46f9b15e48bcbed0318445bb96951011471ab3402b08f9dfbf4\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",
            "Get:1 http://security.ubuntu.com/ubuntu focal-security InRelease [114 kB]\n",
            "Hit:2 http://ppa.launchpad.net/c2d4u.team/c2d4u4.0+/ubuntu focal InRelease\n",
            "Hit:3 http://ppa.launchpad.net/cran/libgit2/ubuntu focal InRelease\n",
            "Hit:4 http://ppa.launchpad.net/deadsnakes/ppa/ubuntu focal InRelease\n",
            "Get:5 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2004/x86_64  InRelease [1,581 B]\n",
            "Hit:6 http://ppa.launchpad.net/graphics-drivers/ppa/ubuntu focal InRelease\n",
            "Hit:7 http://ppa.launchpad.net/ubuntugis/ppa/ubuntu focal InRelease\n",
            "Hit:8 http://archive.ubuntu.com/ubuntu focal InRelease\n",
            "Get:9 https://cloud.r-project.org/bin/linux/ubuntu focal-cran40/ InRelease [3,622 B]\n",
            "Get:10 http://archive.ubuntu.com/ubuntu focal-updates InRelease [114 kB]\n",
            "Get:11 http://archive.ubuntu.com/ubuntu focal-backports InRelease [108 kB]\n",
            "Get:12 http://security.ubuntu.com/ubuntu focal-security/universe amd64 Packages [1,045 kB]\n",
            "Get:13 http://security.ubuntu.com/ubuntu focal-security/main amd64 Packages [2,674 kB]\n",
            "Get:14 https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2004/x86_64  Packages [1,009 kB]\n",
            "Get:15 http://archive.ubuntu.com/ubuntu focal-updates/main amd64 Packages [3,157 kB]\n",
            "Get:16 http://archive.ubuntu.com/ubuntu focal-updates/universe amd64 Packages [1,341 kB]\n",
            "Fetched 9,567 kB in 4s (2,556 kB/s)\n",
            "Reading package lists... Done\n",
            "Building dependency tree       \n",
            "Reading state information... Done\n",
            "27 packages can be upgraded. Run 'apt list --upgradable' to see them.\n",
            "The following additional packages will be installed:\n",
            "  libxtst6 openjdk-8-jre-headless\n",
            "Suggested packages:\n",
            "  openjdk-8-demo openjdk-8-source libnss-mdns fonts-dejavu-extra\n",
            "  fonts-ipafont-gothic fonts-ipafont-mincho fonts-wqy-microhei\n",
            "  fonts-wqy-zenhei fonts-indic\n",
            "The following NEW packages will be installed:\n",
            "  libxtst6 openjdk-8-jdk-headless openjdk-8-jre-headless\n",
            "0 upgraded, 3 newly installed, 0 to remove and 27 not upgraded.\n",
            "Need to get 36.5 MB of archives.\n",
            "After this operation, 144 MB of additional disk space will be used.\n",
            "Selecting previously unselected package libxtst6:amd64.\n",
            "(Reading database ... 122518 files and directories currently installed.)\n",
            "Preparing to unpack .../libxtst6_2%3a1.2.3-1_amd64.deb ...\n",
            "Unpacking libxtst6:amd64 (2:1.2.3-1) ...\n",
            "Selecting previously unselected package openjdk-8-jre-headless:amd64.\n",
            "Preparing to unpack .../openjdk-8-jre-headless_8u362-ga-0ubuntu1~20.04.1_amd64.deb ...\n",
            "Unpacking openjdk-8-jre-headless:amd64 (8u362-ga-0ubuntu1~20.04.1) ...\n",
            "Selecting previously unselected package openjdk-8-jdk-headless:amd64.\n",
            "Preparing to unpack .../openjdk-8-jdk-headless_8u362-ga-0ubuntu1~20.04.1_amd64.deb ...\n",
            "Unpacking openjdk-8-jdk-headless:amd64 (8u362-ga-0ubuntu1~20.04.1) ...\n",
            "Setting up libxtst6:amd64 (2:1.2.3-1) ...\n",
            "Setting up openjdk-8-jre-headless:amd64 (8u362-ga-0ubuntu1~20.04.1) ...\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/orbd to provide /usr/bin/orbd (orbd) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/servertool to provide /usr/bin/servertool (servertool) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/jre/bin/tnameserv to provide /usr/bin/tnameserv (tnameserv) in auto mode\n",
            "Setting up openjdk-8-jdk-headless:amd64 (8u362-ga-0ubuntu1~20.04.1) ...\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/idlj to provide /usr/bin/idlj (idlj) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/wsimport to provide /usr/bin/wsimport (wsimport) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/jsadebugd to provide /usr/bin/jsadebugd (jsadebugd) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/native2ascii to provide /usr/bin/native2ascii (native2ascii) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/javah to provide /usr/bin/javah (javah) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/hsdb to provide /usr/bin/hsdb (hsdb) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/clhsdb to provide /usr/bin/clhsdb (clhsdb) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/xjc to provide /usr/bin/xjc (xjc) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/schemagen to provide /usr/bin/schemagen (schemagen) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/extcheck to provide /usr/bin/extcheck (extcheck) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/jhat to provide /usr/bin/jhat (jhat) in auto mode\n",
            "update-alternatives: using /usr/lib/jvm/java-8-openjdk-amd64/bin/wsgen to provide /usr/bin/wsgen (wsgen) in auto mode\n",
            "Processing triggers for libc-bin (2.31-0ubuntu9.9) ...\n"
          ]
        }
      ],
      "source": [
        "!pip install pyspark\n",
        "!pip install -U -q PyDrive\n",
        "!apt update\n",
        "!apt install openjdk-8-jdk-headless -qq\n",
        "#!apt install default-jre\n",
        "#!apt install default-jdk+\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\""
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "from pydrive.auth import GoogleAuth\n",
        "from pydrive.drive import GoogleDrive\n",
        "from google.colab import auth\n",
        "from oauth2client.client import GoogleCredentials\n",
        "from pyspark.sql import *\n",
        "from pyspark.sql.functions import *\n",
        "from pyspark import SparkContext\n",
        "from pyspark.sql import SQLContext"
      ],
      "metadata": {
        "id": "UmpSfmoqZdH4"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "auth.authenticate_user()\n",
        "gauth = GoogleAuth()\n",
        "gauth.credentials = GoogleCredentials.get_application_default()\n",
        "drive = GoogleDrive(gauth)"
      ],
      "metadata": {
        "id": "HgL6Nl3sZfK1"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "spark = SparkSession.builder.getOrCreate()\n",
        "sc = spark.sparkContext"
      ],
      "metadata": {
        "id": "Ht9nxpMXZhme"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "sqlContext = SQLContext(sc)\n",
        "df_page = sqlContext.read.csv('/content/drive/MyDrive/pagelinks_sample.csv', header=True, inferSchema=True, multiLine=True, escape='\"')\n",
        "rdd_page= df_page.rdd"
      ],
      "metadata": {
        "id": "kNUEXYEmZktH",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "a43d7b4c-85e8-4097-b8cc-c5dfa7ed857e"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stderr",
          "text": [
            "/usr/local/lib/python3.10/dist-packages/pyspark/sql/context.py:112: FutureWarning: Deprecated in 3.0.0. Use SparkSession.builder.getOrCreate() instead.\n",
            "  warnings.warn(\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "sqlContext2 = SQLContext(sc)\n",
        "df_contents = sqlContext2.read.csv('/content/drive/MyDrive/contents.csv', header=True, inferSchema=True, multiLine=True, escape='\"')\n",
        "rdd_contents= df_contents.rdd"
      ],
      "metadata": {
        "id": "bgsZ05XJyGTO"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Uso otro formato para los dos rdds"
      ],
      "metadata": {
        "id": "SzcNoxWmkBGw"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_page_filtrado = rdd_page.map(lambda x: (x.pl_title, x.pl_from))"
      ],
      "metadata": {
        "id": "ee2nA5Io7mcU"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_contents_filtrado = rdd_contents.map(lambda x: (x.title, x.id))"
      ],
      "metadata": {
        "id": "EG9dOKg90dT5"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Necesito trabajar con los artiuclos cuyo titulo figura en contens_csv, ya que para armar el grafo necesito trabajar con las ids asociadas de cada artiuclo. Para esto hago join"
      ],
      "metadata": {
        "id": "T59MaVeX1alc"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_articulos_ids = rdd_page_filtrado.join(rdd_contents_filtrado).cache()"
      ],
      "metadata": {
        "id": "MMHaxwB81TBw"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Agrupo los datos por ids, en donde forma una lista de adyacencia la cual representa a un grafo."
      ],
      "metadata": {
        "id": "YC-_JO81m7GH"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_articulos_ids = rdd_articulos_ids.map(lambda x: (x[1][0], [x[1][1]]))"
      ],
      "metadata": {
        "id": "s84RaqBj5Bb9"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_agrupados_id = rdd_articulos_ids.reduceByKey(lambda x, y: x + y).cache()"
      ],
      "metadata": {
        "id": "dABXzDqz9rb0"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Una vez armado el grafo, defino la funcion que me va a retornar una lista de todos los nodos adyacentes que son adyacentes a la clave."
      ],
      "metadata": {
        "id": "cgb3XD31mqlj"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def devolver_lista_grado_1_separacion(clave):\n",
        "\n",
        "  #Creo una lista de listas (un elemento) con los nodos adyacentes del nodo clave.\n",
        "  rdd_nodos_adyacentes = rdd_agrupados_id.filter(lambda x: x[0] == clave).map(lambda x: x[1])\n",
        "  #Creo la lista de nodos adyacente a la clave en un formato trabajable.\n",
        "  lista_adyacentes = rdd_nodos_adyacentes.take(1)[0]\n",
        "\n",
        "  #Creo un rdd en donde estan los nodos adyacentes de los nodos adyacentes a la clave, luego lo llevo a una lista para poder trabajarlo.\n",
        "  rdd_adyacentes_de_adyacentes = rdd_agrupados_id.filter(lambda x: x[0] in lista_adyacentes).map(lambda x: (1, x[1]))\n",
        "  #Agrupo todas las listas de los nodos de los adyacentes de los adyacentes de la clave.\n",
        "  rdd_adyacentes_de_adyacentes = rdd_adyacentes_de_adyacentes.reduceByKey(lambda x, y: x + y).map(lambda x: x[1]).cache()\n",
        "  rdd_adyacentes_de_adyacentes = sc.parallelize(rdd_adyacentes_de_adyacentes.take(1)[0], 1)\n",
        "  #Elimino nodos repetidos y creo una lista de los adyacentes de los adyacentes del nodo clave.\n",
        "  Lista_adyacentes_de_adyacentes = rdd_adyacentes_de_adyacentes.distinct().collect()\n",
        "  #Si esta en la lista de adyacentes de adyacente lo remuevo.\n",
        "  #Creo la Lista que voy a devolver, ya que necesito fijarme los nodos que no son adyacentes al nodo clave y que no sean\n",
        "  #la clave (Puede estar en lista_adyacentes_adyacntes ya que no es la version final de la misma)\n",
        "  Lista_devolver = rdd_agrupados_id.filter(lambda x: (x[0] in Lista_adyacentes_de_adyacentes) and (clave not in x[1]) and (x[0] not in lista_adyacentes) and (clave != x[0])).map(lambda x: x[0]).collect()\n",
        "  Lista_auxiliar = []\n",
        "\n",
        "  #Agrego aquellos nodos que no aparecen en la lista de rdd_agrupados_id ya que estos nodos no tiene adyacentes, probocando\n",
        "  #que sean de grado 1 con respecto al nodo clave, siempre y cuando no sean adyacentes a la clave o sea la clave.\n",
        "  for i in Lista_adyacentes_de_adyacentes:\n",
        "    if((i not in lista_adyacentes) and (i not in Lista_devolver) and (i != clave)):\n",
        "        Lista_auxiliar.append(i)\n",
        "\n",
        "  Lista_devolver.extend(Lista_auxiliar)\n",
        "  return Lista_devolver\n",
        "\n",
        ""
      ],
      "metadata": {
        "id": "Zis3ELBxJ657"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Funcionamiento del codigo"
      ],
      "metadata": {
        "id": "m3NClxhuThQD"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "Lista_grado_1_separacion_ej1 = devolver_lista_grado_1_separacion(2389124)"
      ],
      "metadata": {
        "id": "Or9FQKewRT4H"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "Lista_grado_1_separacion_ej2 = devolver_lista_grado_1_separacion(300)"
      ],
      "metadata": {
        "id": "ew6skY7eTw_Z"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "Lista_grado_1_separacion_ej3 = devolver_lista_grado_1_separacion(434367)"
      ],
      "metadata": {
        "id": "F55ALKiiT17J"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Aclaracion: La clave elegida tiene que tener adyacentes para que funcione el algoritmo, sino no va a estar represebtado en la lista de adyacencia. Elegi esta implentacion del grafo porque los nodos que no tienen adyacentes no me sirven para realizar el ejercicio porque necesito solo los que tienen un grado de separacion y van a estar ocupando un espacio innecesario."
      ],
      "metadata": {
        "id": "j3nRZHoEVNWl"
      }
    }
  ]
}