PageRank in Spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf = pyspark.SparkConf()\n",
    "conf.setAppName(\"Twitter page rank\")\n",
    "conf.setMaster(\"local[*]\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = pyspark.SparkContext(conf = conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "twitterLines = sc.textFile(\"twitter_closed.jsonl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "followers = twitterLines.map(lambda line: json.loads(line))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "links = followers\\\n",
    "    .keyBy(lambda d: d[\"user_id\"])\\\n",
    "    .mapValues(lambda d: d[\"following_id\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "N = links.count()\n",
    "d = 0.85"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ranks = links\\\n",
    "    .keyBy(lambda usuario: usuario[0])\\\n",
    "    .map(lambda usuario: (usuario[0], 1 / N))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "iterations = 10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def calcula_contribuciones(registro):\n",
    "    \"\"\"\n",
    "    Calcula las contribuciones de un `usuario` hacia las personas a las que sigue\n",
    "    \n",
    "    args:\n",
    "        registro: una tupla de tuplas, de la siguiente forma: (usuario, (seguidos, page_rank))\n",
    "    \n",
    "    returns:\n",
    "        una lista de tuplas de la forma (usuario_seguido, contribución pr)\n",
    "    \"\"\"\n",
    "    \n",
    "    resultados = list()\n",
    "    (usuario, (seguidos, page_rank_usuario)) = registro\n",
    "\n",
    "    for seguido in seguidos:\n",
    "        resultados.append((seguido, page_rank_usuario / len(seguidos)))\n",
    "    resultados.append((usuario, 0))\n",
    "    \n",
    "    return resultados"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for i in range(iterations):\n",
    "    \n",
    "    contributions = links.join(ranks)\\\n",
    "        .flatMap(calcula_contribuciones)\n",
    "    \n",
    "    ranks = contributions\\\n",
    "        .reduceByKey(lambda a, b: a + b)\\\n",
    "        .mapValues(lambda v: (1 - d) / N + d * v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = dict(ranks.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results[2582486138]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results[1306760288243462147]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  }
 ],
 "metadata": {
  "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.8.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}