W3-02b-Pyspark-Partitions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark Tutorial\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": "markdown",
   "metadata": {},
   "source": [
    "Spark was originally developed using Scala, although there are Python and Java interfaces as well. This tutorial covers [most of the RDD API](https://spark.apache.org/docs/latest/rdd-programming-guide.html#resilient-distributed-datasets-rdds) using Python bindings.\n",
    "\n",
    "You may want to consult the [PySpark manual](http://spark.apache.org/docs/2.1.0/api/python/pyspark.html) as well."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "import numpy as np\n",
    "import operator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that I am using an explicit declaration of the number of local processes to use with `local[3]`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "conf=SparkConf().setAppName(\"pyspark tutorial\").setMaster(\"local[3]\")\n",
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Partitions\n",
    "\n",
    "RDD's are broken into multiple partitions or slices which are the unit of work allocation (*i.e.*, more partitions gives more potential for parallelism, but too many partitions gives too much overhead). By default, the number of partitions is related to your cluster size. In this example, I uses `local[3]` to specify three worker processes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a = sc.parallelize([7, 2, 3, 1, 2, 3, 4, 5, 6, 7])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.getNumPartitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also specify the number of partitions or slices when parallelizing a data structure"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a2 = sc.parallelize([7, 2, 3, 1, 2, 3, 4, 5, 6, 7], numSlices=2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a2.getNumPartitions()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## When are partitions visible?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In general, you don't need to be aware of partitions / slices but some code uses the underlying partition information when implementing other operations. We'll look at one example, **fold**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**fold** takes a \"identity value\" and a function and then repeatedly performs a reduction on the RDD using the identity value and function and then again when the values have be collected at the host. `fold` is a general version of `reduce` that handles the case of data across multiple partitions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For example, assume we want to create a list from elements of an RDD using list addition. There are numerous reasons why this is a bad idea, but it helps us illustrate the impact of partitions.\n",
    "\n",
    "The problem is that you can't use \"list addition\" until you have a list. For example, you need to execute:\n",
    "` [] + [1] ` before you can create a list using `[] + [1] + [2]`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "[] + [1] + [2]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**fold** only combines elements in a partition -- **reduce** basically does a **fold** and then combines the results from the individual partitions. Let's apply **fold** to a 3-parition structure:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "a.fold([], lambda x,y: x + [y])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Underneath the hood of **reduce** there's a set of tools that handle operations within a partition and then across partitions. We're going to look at the general **aggregate** method."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregate -- generalization of reduce and fold\n",
    "\n",
    "In order to understand the order of operations, we need a function that will illustrate that order for an RDD. The **showAdd** function will show the order of operations using parenthesis."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def showAdd(x,y):\n",
    "    return f\"({x} + {y})\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "showAdd( showAdd(1,2), 3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "oneslice = sc.parallelize([2,3,4,5,6],1)\n",
    "oneslice.reduce(showAdd)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If we partition the same data into two slices, we see that one partition contains `(2+3)` and the other contains `(4+5)+6`. The reduce combines these two together."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "twoslice = sc.parallelize([2,3,4,5,6],2)\n",
    "twoslice.reduce(showAdd)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, lets see the semantics of **fold**:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "oneslice.fold(1,showAdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "twoslice.fold(1,showAdd)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In other words, the identity element is added to the first element of each partition Like **reduce**, the **fold** operation really only works well for commutative-associative operators because it's applied to each slice of an RDD independently.\n",
    "\n",
    "Recall that we explicitly specified that `twoslice`should have two slices."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "twoslice.reduce(showAdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "twoslice.fold(1, showAdd)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The **aggregate* function performs an operation like`fold` on each RDD partition and then uses a __combine function__ to join partitions.\n",
    "\n",
    "For example, assume we have data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "twoPart = sc.parallelize([1,2,3,4], numSlices=2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This data will (likely) be divided into `[1,2]` and `[3,4]`. Now assume we want to reduce two values -- the first is the sum of the data (10) and the second is the length of largest partition (likely 2).\n",
    "\n",
    "We'll have two distinct functions -- `seqOp` will define operations within a partition and `combOp` will define op how partitions are combined."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def seqOp( x, y):\n",
    "    return f\"({x} + S + {y})\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def combOp( x, y ):\n",
    "    return f\"({x} + C + {y})\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As with `fold`, we need a \"zero-value\" to start folding"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "oneslice.aggregate( 0, seqOp, combOp )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Recall that `oneslice` has a single partition. The `seqOp` operation is applied to the elements in the single RDD and combined with the identity value (0). That RDD is then combined using  `combOp` the identity value (0) and the result from the single RDD.\n",
    "\n",
    "Now, lets see what this is like for two slices."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "twoslice.aggregate( 0, seqOp, combOp )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this case, the two values in each of the two RDD's are combined using `seqOp` and the identity element.\n",
    "\n",
    "The result from the two sequences are then combined using `compOp` in a *left to right* oerdering.\n",
    "\n",
    "**aggregate** is a the basis of many of the other operations in Spark. You can use it to build additional extensions, but many of the common operations we need are built in using **aggregate**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}