SparkContext_Workers_Lazy_Evaluations.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SparkContext - number of workers and lazy evaluation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Checking the impact of number of workers\n",
    "While initializing the `SparkContext`, we can specify number of worker nodes. Generally, it is recommended to have one worker per core of the machine. But it can be smaller or larger. In the following code, we will examine the impact of number of worker cores on some parallelized operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from time import time\n",
    "from pyspark import SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1 executors, time = 4.308391571044922\n",
      "2 executors, time = 2.318211793899536\n",
      "3 executors, time = 2.5603320598602295\n",
      "4 executors, time = 2.663661003112793\n"
     ]
    }
   ],
   "source": [
    "for j in range(1,5):\n",
    "    sc= SparkContext(master = \"local[%d]\"%(j))\n",
    "    t0=time()\n",
    "    for i in range(10):\n",
    "        sc.parallelize([1,2]*10000).reduce(lambda x,y:x+y)\n",
    "    print(f\"{j} executors, time = {time()-t0}\")\n",
    "    sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### We observe that it takes almost double time for 1 worker, and after that time reduces to a flat level for 2,3,4 workers etc. This is because this code run on a Linux virtual box using only 2 cores from the host machine. If you run this code on a machine with 4 cores, you will see benefit upto 4 cores and then the flattening out of the time taken. It also become clear that using more than one worker per core is not beneficial as it just does context-switching in that case and does not speed up the parallel computation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Showing the essence of _lazy_ evaluation\n",
    "![](https://qph.fs.quoracdn.net/main-qimg-d49dcf35ecb7eecfc6e5b39493a0e086-c)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext(master=\"local[2]\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Make a RDD with 1 million elements"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 316 µs, sys: 5.13 ms, total: 5.45 ms\n",
      "Wall time: 24.6 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "rdd1 = sc.parallelize(range(1000000))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Some computing function - `taketime`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "from math import cos\n",
    "def taketime(x):\n",
    "    [cos(j) for j in range(100)]\n",
    "    return cos(x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check how much time is taken by `taketime` function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 21 µs, sys: 7 µs, total: 28 µs\n",
      "Wall time: 31.5 µs\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "-0.4161468365471424"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "taketime(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Now do the `map` operation on the function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 23 µs, sys: 8 µs, total: 31 µs\n",
      "Wall time: 34.8 µs\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "interim = rdd1.map(lambda x: taketime(x))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### How come each taketime function takes 45.8 us but the map operation with a 10000 element RDD also took similar time?<br><br>Because of _lazy_ evaluation i.e. nothing was computed in the previous step, just a plan of execution was made. The variable `interim` does not point to a data structure, instead it points to a plan of execution, expressed as a dependency graph. The dependency graph defines how RDDs are computed from each other."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let's see the \"Dependency Graph\" using `toDebugString` method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) PythonRDD[10] at RDD at PythonRDD.scala:49 []\n",
      " |  ParallelCollectionRDD[7] at parallelize at PythonRDD.scala:184 []\n"
     ]
    }
   ],
   "source": [
    "print(interim.toDebugString().decode())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![](https://raw.githubusercontent.com/tirthajyoti/Spark-with-Python/master/Images/RDD_dependency_graph.PNG)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The actual execution by `reduce` method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "output = -0.28870546796843666\n",
      "CPU times: user 11.6 ms, sys: 5.56 ms, total: 17.2 ms\n",
      "Wall time: 15.6 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "print('output =',interim.reduce(lambda x,y:x+y))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "31.0"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "1000000*31e-6"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### It is less than what we would have expected considering 1 million operations with the `taketime` function. This is the result of parallel operation of 2 cores."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Now, we have not saved (materialized) any intermediate results in `interim`, so another simple operation (e.g. counting elements > 0) will take almost same time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "500000\n",
      "CPU times: user 10.6 ms, sys: 8.55 ms, total: 19.2 ms\n",
      "Wall time: 12.1 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "print(interim.filter(lambda x:x>0).count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Caching to reduce computation time on similar operation (spending memory)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Run the same computation as before with `cache` method to tell the dependency graph to plan for caching"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 7.22 ms, sys: 4.29 ms, total: 11.5 ms\n",
      "Wall time: 63 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "interim = rdd1.map(lambda x: taketime(x)).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(2) PythonRDD[14] at RDD at PythonRDD.scala:49 [Memory Serialized 1x Replicated]\n",
      " |  ParallelCollectionRDD[7] at parallelize at PythonRDD.scala:184 [Memory Serialized 1x Replicated]\n"
     ]
    }
   ],
   "source": [
    "print(interim.toDebugString().decode())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "output = -0.28870546796843666\n",
      "CPU times: user 16.4 ms, sys: 2.24 ms, total: 18.7 ms\n",
      "Wall time: 15.3 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "print('output =',interim.reduce(lambda x,y:x+y))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Now run the same `filter` method with the help of cached result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "500000\n",
      "CPU times: user 14.2 ms, sys: 3.27 ms, total: 17.4 ms\n",
      "Wall time: 811 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "print(interim.filter(lambda x:x>0).count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### This time it took much shorter time due to cached result, which it could use to compare to 0 and count easily."
   ]
  }
 ],
 "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
}