Partioning and Gloming.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Partitioning\n",
    "\n",
    "When an RDD is created, you can specify the number of partitions.\n",
    "<br>The default is the number of workers defined when you setu th `SparkContext`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating `SparkContext` with 2 workers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext(master=\"local[2]\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "A = sc.parallelize(range(1000000))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Use `getNumPartition` to retrive the number of partitions created"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2\n"
     ]
    }
   ],
   "source": [
    "print(A.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We can repartition _A_ in any number of partitions we want"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "D = A.repartition(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10\n"
     ]
    }
   ],
   "source": [
    "print(D.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### We can also set the number of partitions while creating the RDD with `numSlices` argument "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "A = sc.parallelize(range(1000000),numSlices=8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8\n"
     ]
    }
   ],
   "source": [
    "print(A.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Why partitions are important?\n",
    "\n",
    "* They define the unit the executor works on\n",
    "* You should have at least as many partitions as the number of worker nodes\n",
    "* Smaller partitions may allow more parallelization"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Repartitioning for Load Balancing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Suppose we start with 10 partitions, all with exactly the same number of elements"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000]\n"
     ]
    }
   ],
   "source": [
    "A=sc.parallelize(range(1000000)).map(lambda x:(x,x)).partitionBy(10)\n",
    "print(A.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Suppose we want to use **`filter()`** to select some of the elements in A.<br>\n",
    "Some partitions might have more elements remaining than others."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[100000, 0, 0, 0, 0, 100000, 0, 0, 0, 0]\n"
     ]
    }
   ],
   "source": [
    "#select 10% of the entries\n",
    "# A bad filter for numbers divisable by 5\n",
    "B=A.filter(lambda pair: pair[0]%5==0)\n",
    "# get no. of partitions\n",
    "print(B.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Future operations on B will use only two workers.<br>\n",
    "The other workers will do nothing, because their partitions are empty."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### To fix the situation we need to repartition the unbalanced RDD. <br>One way to do that is to repartition using a new key using the method `partitionBy()`\n",
    "\n",
    "* The method **`.partitionBy(k)`** expects to get a **`(key,value)`** RDD where keys are integers.\n",
    "* Partitions the RDD into **`k`** partitions.\n",
    "* The element **`(key,value)`** is placed into partition no. **`key % k`**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000]\n"
     ]
    }
   ],
   "source": [
    "C=B.map(lambda pair:(pair[1]/10,pair[1])).partitionBy(10) \n",
    "print(C.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note, how **`C`** consists of only 200,000 elements from the unbalanced **`B`** partition but redistributes them in equal partitions of 20,000 elements each."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Another approach is to use random partitioning using **`repartition(k)`**\n",
    "* An **advantage** of random partitioning is that it does not require defining a key.\n",
    "* A **disadvantage** of random partitioning is that you have no control on the partitioning i.e. which elements go to which partition."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000]\n"
     ]
    }
   ],
   "source": [
    "C=B.repartition(10)\n",
    "print(C.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## `Glom()`\n",
    "* In general, spark does not allow the worker to refer to specific elements of the RDD.\n",
    "* Keeps the language clean, but can be a major limitation.\n",
    "\n",
    "#### `glom()` transforms each partition into a tuple (immutabe list) of elements.<br> Creates an RDD of tules. One tuple per partition. <br>Workers can refer to elements of the partition by index but you cannot assign values to the elements, the RDD is still immutable.\n",
    "\n",
    "* Consider **the command used above to count the number of elements in each partition.**: `print(C.glom().map(len).collect())`\n",
    "* We used `glom()` to make each partition into a tuple.\n",
    "* We used `len` on each partition to get the length of the tuple - size of the partition.\n",
    "* We `collect`ed the results to print them out."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### A more elaborate example\n",
    "There are many things that you can do using `glom()`.\n",
    "<br>\n",
    "For example, suppose we want to get the first element, the number of elements, and the sum of the elements of the unbalanced partitions we made from `A` into `B`. Of the partition is empty we just return `None`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(0, 100000, 999990), None, None, None, None, (5, 100000, 999990), None, None, None, None]\n"
     ]
    }
   ],
   "source": [
    "def getPartitionInfo(G):\n",
    "    d=0\n",
    "    if len(G)>1: \n",
    "        for i in range(len(G)-1):\n",
    "            d+=abs(G[i+1][1]-G[i][1]) # access the glomed RDD that is now a tuple (immutable list)\n",
    "        return (G[0][0],len(G),d)\n",
    "    else:\n",
    "        return(None)\n",
    "\n",
    "output=B.glom().map(lambda B: getPartitionInfo(B)).collect()\n",
    "print(output)"
   ]
  }
 ],
 "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.6.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}