W3-04-Pyspark-Applications.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark Tutorial - Applications\n",
    "<div>\n",
    " <h2> CSCI 4283 / 5253 \n",
    "  <IMG SRC=\"https://www.colorado.edu/cs/profiles/express/themes/cuspirit/logo.png\" WIDTH=50 ALIGN=\"right\"/> </h2>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "import numpy as np\n",
    "import operator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf=SparkConf().setAppName(\"pyspark tutorial\").setMaster(\"local[*]\")\n",
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## WordCount"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines=sc.textFile(\"hamlet.txt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['THE TRAGEDY OF HAMLET, PRINCE OF DENMARK', '', '']"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "counts = lines.flatMap(lambda line: line.split())\\\n",
    "              .map(lambda word: (word,1))\\\n",
    "              .reduceByKey(operator.add)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('TRAGEDY', 1), ('OF', 2), ('PRINCE', 1), ('Shakespeare', 1), ('Dramatis', 1)]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "counts.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('the', 1083), ('and', 939), ('to', 727), ('of', 670), ('a', 540)]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lines.flatMap(lambda line: line.split())\\\n",
    "              .map(lambda word: word.lower())\\\n",
    "              .map(lambda word: (word,1))\\\n",
    "              .reduceByKey(operator.add)\\\n",
    "              .sortBy(lambda x: x[1], ascending=False)\\\n",
    "              .take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Page Rank\n",
    "\n",
    "We represent our graph as a simple vertex-edge-list with the edges stored as tuples. Because each node is a Key-Value, we can directly parallelize the graph and then operate on it using K-V operation.s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "graph = sc.parallelize([\n",
    "    ('A', ('B')),\n",
    "    ('B', ('A', 'C')),\n",
    "    ('C', ('A', 'D')),\n",
    "    ('D', ('A'))\n",
    "])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The current page rank is represented as pairs of the node name the current value. We initialize the page rank to 1.0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "ranks = graph.map( lambda node: (node[0], 1.0))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('A', 1.0), ('B', 1.0), ('C', 1.0), ('D', 1.0)]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ranks.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The current page will contribute its current rank divided by the number of out edges to each node. Because the edge list indicates the destination node, this will produce pairs of values indicating the node and the contribution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def computeContrib(edges, rank):\n",
    "    return ( (e, rank/len(edges)) for e in edges )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('A', 0.5), ('D', 0.5)]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "list(computeContrib(('A','D'), 1.0))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We need to use both the graph and the current rank information -- we do this using a `join`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('B', (('A', 'C'), 1.0)),\n",
       " ('D', ('A', 1.0)),\n",
       " ('C', (('A', 'D'), 1.0)),\n",
       " ('A', ('B', 1.0))]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "graph.join(ranks).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, to compute the contribution of each link for each node, we use use `computeContrib` for that nodes information (edge list & rank). Here's an example of that happening in a single step:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('A', 0.5), ('C', 0.5), ('A', 1.0), ('A', 0.5), ('D', 0.5), ('B', 1.0)]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "graph.join(ranks).flatMap(lambda node: computeContrib(node[1][0],node[1][1])).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we reduce the values by the key and sum up the contributions. For example,"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('B', 1.0), ('D', 0.5), ('C', 0.5), ('A', 2.0)]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "graph.join(ranks).flatMap(lambda node: computeContrib(node[1][0],node[1][1]))\\\n",
    "    .reduceByKey(operator.add).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These contributions are used to calculate the final page rank.\n",
    "\n",
    "We can then perform the rank update operation multiple times until we converge to an answer. In our case, we're going to just run the code 5 times."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "=== Iteration 0 ====\n",
      "Contribs are [('A', 0.5), ('C', 0.5), ('A', 1.0), ('A', 0.5), ('D', 0.5), ('B', 1.0)]\n",
      "Ranks are [('B', 1.0), ('D', 0.575), ('C', 0.575), ('A', 1.8499999999999999)]\n",
      "=== Iteration 1 ====\n",
      "Contribs are [('A', 0.5), ('C', 0.5), ('B', 1.8499999999999999), ('A', 0.575), ('A', 0.2875), ('D', 0.2875)]\n",
      "Ranks are [('B', 1.7224999999999997), ('A', 1.3081249999999998), ('D', 0.394375), ('C', 0.575)]\n",
      "=== Iteration 2 ====\n",
      "Contribs are [('A', 0.8612499999999998), ('C', 0.8612499999999998), ('A', 0.394375), ('A', 0.2875), ('D', 0.2875), ('B', 1.3081249999999998)]\n",
      "Ranks are [('B', 1.2619062499999998), ('D', 0.394375), ('C', 0.8820624999999999), ('A', 1.4616562499999997)]\n",
      "=== Iteration 3 ====\n",
      "Contribs are [('A', 0.394375), ('A', 0.6309531249999999), ('C', 0.6309531249999999), ('A', 0.44103124999999993), ('D', 0.44103124999999993), ('B', 1.4616562499999997)]\n",
      "Ranks are [('D', 0.5248765624999999), ('B', 1.3924078124999997), ('C', 0.6863101562499999), ('A', 1.3964054687499998)]\n",
      "=== Iteration 4 ====\n",
      "Contribs are [('B', 1.3964054687499998), ('A', 0.6962039062499998), ('C', 0.6962039062499998), ('A', 0.5248765624999999), ('A', 0.34315507812499996), ('D', 0.34315507812499996)]\n",
      "Ranks are [('A', 1.4796002148437497), ('B', 1.3369446484374998), ('D', 0.4416818164062499), ('C', 0.7417733203124999)]\n",
      "====\n",
      "Final rank: [('A', 1.4796002148437497), ('B', 1.3369446484374998), ('D', 0.4416818164062499), ('C', 0.7417733203124999)]\n"
     ]
    }
   ],
   "source": [
    "for itr in range(5):\n",
    "    print(\"=== Iteration {} ====\".format(itr))\n",
    "    contribs = graph.join(ranks).\\\n",
    "       flatMap(lambda node: computeContrib(node[1][0], node[1][1]))\n",
    "    print(\"Contribs are\", contribs.collect())\n",
    "    ranks = contribs.reduceByKey(operator.add).\\\n",
    "                     mapValues(lambda rank: rank * 0.85 + 0.15)\n",
    "    print(\"Ranks are\", ranks.collect())\n",
    "print(\"====\")\n",
    "print(\"Final rank:\", ranks.collect())"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": []
  }
 ],
 "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.10.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}