ch11_Parallelization.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Chapter 11: Parallelization\n",
    "\n",
    "This notebook shows IPthon parallel capabilities. We will see the commands that allow\n",
    "us to parallelize tasks, see section Introduction. If properly combined, they can lead to \n",
    "interting parallel algorithms, see section A complete example. \n",
    "\n",
    "\n",
    "## Introduction\n",
    "\n",
    "In order to start IPython's parallel capabilities, the simplest way of proceeding \n",
    "it to click on the Clusters tab of the notebook dashboard, and press \n",
    "Start with the desired number of cores. \n",
    "This will automatically run the necessary commands to start the\n",
    "IPython cluster. We will now be able to send different tasks to\n",
    "the engines using the web interface.\n",
    "\n",
    "The next commands allows you to connect to the cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from IPython import parallel\n",
    "engines = parallel.Client()\n",
    "engines.block = True\n",
    "print engines.ids"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These commands connect to the cluster and output the number of engines in it. \n",
    "If an error is shown when running the\n",
    "commands, the cluster has not been correctly created.\n",
    "\n",
    "### Direct view of engines\n",
    "\n",
    "The following commands,\n",
    "executed on the notebook (i.e., the IPython interpreter), send commands\n",
    "to the first engine:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "engines[0].execute('a = 2') \n",
    "engines[0].execute('b = 10') \n",
    "engines[0].execute('c = a + b')  \n",
    "engines[0].pull('c')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Observe that we do not have direct access to the command line of the first engine. Rather, we send commands to it through the client. The result is retrieved by means of the `pull` command.\n",
    "\n",
    "Since each engine is\n",
    "an independent process, the operating system may schedule each engine\n",
    "in a different core and thus execution may be performed in parallel. Take a look at the following example."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "engines[0].execute('a = 2') \n",
    "engines[0].execute('b = 10') \n",
    "engines[1].execute('a = 9') \n",
    "engines[1].execute('b = 7') \n",
    "engines[0:2].execute('c = a + b')   \n",
    "engines[0:2].pull('c')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Observe that the operation `c = a + b` is executed on each egine. The latter calculation and does not show the power of parallization. For that purpose we are next perform more computational intenstive tasks. Let us now show that we are really doing computations in parallel. Let us try with something bigger!\n",
    "\n",
    "In order to simplify the code, let us define the following variable that references the first two engines (even if there are more active engines)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "dview2 = engines[0:2]    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We are next going to focus on matrix multiplication, for instance. We begin by doing serialized computations on the notebook and compute the total processing time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import time\n",
    "import numpy as np\n",
    "\n",
    "# Create four 1000x1000 matrix\n",
    "A0 = np.random.rand(1000,1000)\n",
    "B0 = np.random.rand(1000,1000)\n",
    "A1 = np.random.rand(1000,1000)\n",
    "B1 = np.random.rand(1000,1000)\n",
    "\n",
    "t0 = time.time() \n",
    "\n",
    "C0 = np.dot(A0, B0)\n",
    "C1 = np.dot(A1, B1)\n",
    "    \n",
    "print \"Time in seconds (Computations): \", time.time() - t0 "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And now we will do computations in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "dview2.execute('import numpy as np')       # We import numpy on both engines!\n",
    "\n",
    "t0 = time.time()\n",
    "engines[0].push(dict(A=A0, B=B0))    # We send A0 and B0 to engine 0 \n",
    "engines[1].push(dict(A=A1, B=B1))    # We send A1 and B1 to engine 1 \n",
    "\n",
    "t0_computations = time.time()\n",
    "\n",
    "dview2.execute('C = np.dot(A,B)')\n",
    "    \n",
    "print \"Computations: \", time.time() - t0_computations\n",
    "\n",
    "[C0, C1] = dview2.pull('C')\n",
    "print \"Time in seconds: \", time.time() - t0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The total computing time should decrease thanks to the divison of the computation in two tasks. Each task is then manually executed in two different engines and each engine is scheduled by the operating system on two different processors (if the computer has at least two processors).\n",
    "\n",
    "The previous commands show us how to execute commands on engines as if we were typing \n",
    "them directly on the command-line. Indeed, we\n",
    "have manually sent, executed and retrieved the results of computations.\n",
    "This procedure may be useful in some cases but in many cases there\n",
    "will be no need to do so. The `map` function may be used to that purpose"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def mul(A, B):\n",
    "    import numpy as np\n",
    "    C = np.dot(A, B)\n",
    "    return C\n",
    "\n",
    "[C0, C1] = dview2.map(mul,[A0, A1],[B0, B1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These commands, executed on the client, perform a remote call.\n",
    "The function `mul` is defined locally. There is no need to use the `push`\n",
    "and `pull` functions explicitly to send and retrieve the results; it is done\n",
    "implicitly. Note the `import numpy as np`\n",
    "inside the `mul` function. This is a common model, to ensure that the appropriate\n",
    "toolboxes are imported to where the task is run. \n",
    "\n",
    "The `map` call splits the tasks between the engines associated\n",
    "with `dview2`. In the previous example, the task `mul(A0,B0)`\n",
    "is executed on one engine and `mul(A1, B1)` is executed on\n",
    "the one. Which command is executed on each engine? What happens if\n",
    "the list of arguments to map includes three or more matrices? We may\n",
    "see this with the following example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "engines[0].execute('my_id = \"engineA\"') \n",
    "engines[1].execute('my_id = \"engineB\"')\n",
    "\n",
    "def sleep_and_return_id(sec):     \n",
    "    import time     \n",
    "    time.sleep(sec)                      \n",
    "    return my_id,sec\n",
    "\n",
    "dview2.map(sleep_and_return_id, [3,3,3,1,1,1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Execute the previous code and observe the returned result that indicates us which engine executed the function. You may repeat this experment as many times as you wish, but the result will always be the same. The tasks are distributed in a uniform way among the\n",
    "engines before executing them no matter which is the delay we pass\n",
    "as argument to the function `sleep_and_return_id`. This is in fact a characteristic of the direct view interface: the tasks are distributed among the engines before executing them. \n",
    "\n",
    "This a good way to proceed if you expect each task to take\n",
    "the same amount of time. But if not, as is the case in the previous\n",
    "example, computation time is wasted and so we recommend to use the \n",
    "load-balanced view instead.\n",
    "\n",
    "### Load-balanced view of engines\n",
    "\n",
    "This interface is simpler and more powerful than the direct interface. We would like to point out, however, that with this interface the user has no direct access to individual engines. It is the IPython scheduler that assignes work to each engine. To create a load-balanced view we may use the following command: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "engines.block = True\n",
    "lview2 = engines.load_balanced_view(targets=[0,1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We use the blocking mode since it simplifies the code. The `lview2` is a variable\n",
    "that references the first two engines.\n",
    "\n",
    "Our example here will be centered on the `sleep_and_return_id`\n",
    "function we have seen in the previous subsection:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "lview2.map(sleep_and_return_id, [3,3,3,1,1,1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Observe that rather than using the direct\n",
    "view interface (`dview2` variable) of the `map` function, we use the associated load-balanced view interface (`lview2` variable).\n",
    "\n",
    "In this case, the tasks are assigned to the engines in a dynamic way.\n",
    "The `map` function of the load-balanced view begins by assigning\n",
    "one task to each engine in the order given by the parameters of the\n",
    "`map` function. By default, the load-balanced view scheduler\n",
    "then assigns a new task to an engine when it becomes free"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## A complete example: the New York taxi trips database\n",
    "\n",
    "We next present a real application of the parallel capabilities\n",
    "of IPython and the discussion of several approaches to it. The dataset\n",
    "is a database of taxi trips in New York and it has been obtained through\n",
    "a Freedom of Information Law (FOIL) request from the New York City\n",
    "Taxi & Limousine Commission (NYCT&L) by University of Illinois at\n",
    "Urbana-Champaign (http://publish.illinois.edu/dbwork/open-data/).\n",
    "The dataset consists in $12\\times2$GBytes Comma Separated Files (CSV)\n",
    "files. Each file has approximately $14$ million entries (lines) and\n",
    "is already cleaned. Thus no special preprocessing is needed to be\n",
    "able to process it. For our purposes we are interested only in the\n",
    "following information from each entry: \n",
    "\n",
    "+ `pickup_datetime`: start time of the trip, mm-dd-yyyy hh24:mm:ss\n",
    "EDT. \n",
    "+ `pickup_longitude` and `pickup_latitude`: GPS coordinates\n",
    "at the start of the trip. \n",
    "\n",
    "Our objective is to perform an analysis of this data in order to answer\n",
    "the following questions: for each district, how many pickups are performed\n",
    "during week days and how many during weekends? And how many pickups\n",
    "are performed in the morning? For that issue the New York city is\n",
    "arbitrarily divided into nine districts: ChinaTown, WTC, Soho, Harlem,\n",
    "UpperTown, MidTown, DownTown, UpperEastSide, UpperWestSide and Financial. \n",
    "\n",
    "Implementing the previous classification is rather simple since it\n",
    "only requires checking, for each entry, the GPS coordinates of the\n",
    "start of the trip and the pickup datetime. Performing this task in\n",
    "a sequential may take a rather large amount of time since the number\n",
    "of entries, for a single CSV file, is rather large. In addition, special\n",
    "care has to be taken when reading the file since a 2GByte file may\n",
    "not fully fit into the computer's memory. \n",
    "\n",
    "We may take advantage of the parallelization capabilities in order\n",
    "to reduce the processing time. The idea is to divide the input data\n",
    "into chunks so that each engine takes care of classifying the entries\n",
    "of their corresponding chunks. We propose here an approach which \n",
    "is based on implementing a producer-consumer paradigm\n",
    "in order to distribute the tasks. The producer, associated to the\n",
    "client, reads the chunks from disc and distributes them among the\n",
    "engines using a round robin technique. No explicit `map` function\n",
    "is used in this case. Rather, we simulate the behavior of the `map`\n",
    "function in order to have fine control of the parallel problem. Recall\n",
    "that each engine is an independent process. Since we assign different\n",
    "tasks to each engine, the operating system will try to execute each\n",
    "engine on a different process.\n",
    "\n",
    "For further deails, please see corresponding chapter in the book.\n",
    "\n",
    "## The source code\n",
    "\n",
    "We begin by initializing the engines:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%reset -f\n",
    "\n",
    "from IPython import parallel\n",
    "from itertools import islice\n",
    "from itertools import cycle\n",
    "from collections import Counter\n",
    "import sys\n",
    "import time\n",
    "\n",
    "#Connect to the Ipython cluster    \n",
    "engines = parallel.Client()\n",
    "\n",
    "#Create a DirectView to all engines\n",
    "dview = engines.direct_view()\n",
    "\n",
    "print \"The number of engines in the cluster is: \" + str(len(engines.ids))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We next declare the functions that will be executed on the engines. We do this thanks to the `%%px` parallel magic command."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%px\n",
    "\n",
    "# The %%px magic executes the code of this cell on each engine.\n",
    "\n",
    "from datetime import datetime\n",
    "from collections import Counter\n",
    "\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "# A Counter object to store engine's local result\n",
    "local_total = Counter();\n",
    "\n",
    "def dist(p0, p1):\n",
    "    \"Returns the distance**2 between two points\"\n",
    "    # We compute the squared distance. Since we only want to compare\n",
    "    # distances there is no need to compute the square root (sqrt) \n",
    "    return (p0[0] - p1[0])**2 + (p0[1] - p1[1])**2\n",
    "\n",
    "# Coordinates (latitude, longitude) of diferent points of the island\n",
    "district_dict = { \n",
    "    'Financial': [40.724863, -73.994718], \n",
    "    'Midtown': [40.755905, -73.984997],\n",
    "    'Chinatown': [40.716224, -73.995925],\n",
    "    'WTC': [40.711724, -74.012888],\n",
    "    'Harlem': [40.810469, -73.943318],\n",
    "    'Uppertown': [40.826381, -73.943964],\n",
    "    'Soho': [40.723783, -74.001237],\n",
    "    'UpperEastSide': [40.773861, -73.956329],\n",
    "    'UpperWestSide': [40.787347, -73.975267]\n",
    "    }\n",
    "\n",
    "# Computes the distance to each district center and obtains the one that\n",
    "# gives minimum distance\n",
    "def get_district(coors):\n",
    "    \"Given a coordinate inn latitude and longitude, returns the district in Manhatan\"   \n",
    "    #If dist^2 is bigger than 0.0005, the district is 'None'.\n",
    "    dist_min = 0.0005\n",
    "    district = None\n",
    "    for key in district_dict.iterkeys():\n",
    "        d = dist(coors, district_dict[key])\n",
    "        if dist_min > d:\n",
    "            dist_min = d\n",
    "            district = key\n",
    "    return district\n",
    "\n",
    "def is_morning(d):\n",
    "    \"Given a datetime, returns if it was on morning or not\"\n",
    "    h = datetime.strptime(d, \"%Y-%m-%d %H:%M:%S\").hour\n",
    "    return 0 <= h and h < 12\n",
    "\n",
    "def is_weekend(d):\n",
    "    \"Given a datetime, returns if it was on weekend or not\"\n",
    "    wday = datetime.strptime(d, \"%Y-%m-%d %H:%M:%S\").weekday() #strptime transforms str to date\n",
    "    return 4 < wday <= 6\n",
    "\n",
    "#Function that classifies a single data\n",
    "def classify(x):\n",
    "    \"Given a tuple with a datetime, latitude and longitude, returns the group where it fits\"\n",
    "    date, lat, lon = x\n",
    "    latitude = float(lat)\n",
    "    longitude = float(lon)\n",
    "    return is_weekend(date), is_morning(date), get_district([latitude, longitude])\n",
    "\n",
    "# Function that given a dictionary (data), applies classify function on each element\n",
    "# and returns an histogram in a Counter object\n",
    "def process(b):\n",
    "    #Recives a block (list of strings) and updates result in global var local_total()\n",
    "    global local_total\n",
    "    \n",
    "    #Create an empty df. Preallocate the space we need by providing the index (number of rows)\n",
    "    df = pd.DataFrame(index=np.arange(0,len(b)), columns=('datetime','latitude','longitude'))\n",
    "    \n",
    "    # Data is a list of lines, containing datetime at col 5 and latitude at row 11.\n",
    "    # Allocate in the dataFrame the datetime and latitude and longitude dor each line in data\n",
    "    count = 0\n",
    "    for line in b:\n",
    "        elements = line.split(\",\")\n",
    "        df.loc[count] = elements[5], elements[11], elements[10]\n",
    "        count += 1\n",
    "        \n",
    "    #Delete NaN values from de DF\n",
    "    df.dropna(thresh=(len(df.columns) - 1), axis=0)\n",
    "    \n",
    "    #Apply classify function to the dataFrame\n",
    "    cdf = df.apply(classify, axis=1)\n",
    "    \n",
    "    #Increment the global variable local_total\n",
    "    local_total += Counter(cdf.value_counts().to_dict())\n",
    "\n",
    "# Initialization function\n",
    "def init():\n",
    "    #Reset total var\n",
    "    global local_total\n",
    "    local_total = Counter()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next we show the code executed by the client. The next code performs the next task\n",
    "\n",
    "+ It reads a chunk of `lines_per_block` lines form the file. The chunk is assigned to an engine which performs the classification. The result of the classification is updated on a local variable on each engine. This process is repeated until all chunks have been processed by the engines.\n",
    "+ Once finished, the client retrieves the local variable of each engine and computes the overall result.\n",
    "\n",
    "This is the principle of the **MapReduce** programming model: a MapReduce program is composed of a Map() procedure that performs filtering and sorting (such as counting the number of times each word appears in a file) and a Reduce() procedure that performs a summary operation (that is, taking each of the results and computing the overall result)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# This is the main code executed on the client\n",
    "t0 = time.time() \n",
    "\n",
    "#File to be processed\n",
    "filename = 'trip_data.csv'\n",
    "\n",
    "def get_chunk(f,N):\n",
    "    \"\"\" Returns blocks of nl lines from the file descriptor fd\"\"\"\n",
    "    #Deletes first line on first chunk (header line)\n",
    "    first = 1\n",
    "    while True:\n",
    "        new_chunk = list(islice(f, first, N))\n",
    "        if not new_chunk:\n",
    "            break\n",
    "        first = 0\n",
    "        yield new_chunk\n",
    "\n",
    "# A simple counter to verify execution\n",
    "chunk_n = 0\n",
    "\n",
    "# Number of lines to be sent to each engine at a time. Use carefully!\n",
    "lines_per_block = 20\n",
    "\n",
    "# Create an emty list of async tasks. One element for each engine\n",
    "async_tasks = [None] * len(engines.ids)\n",
    "\n",
    "# Cycle Object to get an infinite iterator over the list of engines\n",
    "c_engines = cycle(engines.ids)\n",
    "\n",
    "# Initialize each engine. Observe that the execute is performed\n",
    "# in a non-blocking fashion.\n",
    "for i in engines.ids:\n",
    "    async_tasks[i] = engines[i].execute('init()', block=False)\n",
    "\n",
    "# The variable to store results\n",
    "global_result = Counter()\n",
    "\n",
    "# Open the file in ReadOnly mode\n",
    "try:\n",
    "    f = open(filename, 'r') #iterable\n",
    "except IOError:\n",
    "    sys.exit(\"Could not open input file!\")\n",
    "\n",
    "# Used to show the progress\n",
    "print \"Beginning to send chunks\"\n",
    "sys.stdout.flush()\n",
    "\n",
    "# While the generator returns new chunk, sent them to the engines\n",
    "for new_chunk in get_chunk(f,lines_per_block):\n",
    "    \n",
    "    #After the first loop, first_chunk is False. \n",
    "    first_chunk = False\n",
    "    \n",
    "    #Decide the engine to be used to classify the new chunk\n",
    "    run_engine = c_engines.next()\n",
    "    \n",
    "    # Wait until the engine is ready\n",
    "    while ( not async_tasks[run_engine].ready() ):\n",
    "        time.sleep(1)\n",
    "    \n",
    "    #Send data to the assigned engine.\n",
    "    mydict = dict(data = new_chunk)\n",
    "    \n",
    "    # The data is sent to the engine in blocking mode. The push function does not return\n",
    "    # until the engine has received the data. \n",
    "    engines[run_engine].push(mydict,block=True)\n",
    "\n",
    "    # We execute the classification task on the engine. Observe that the task is executed\n",
    "    # in non-blocking mode. Thus the execute function reurns immediately. \n",
    "    async_tasks[run_engine] = engines[run_engine].execute('process(data)', block=False)\n",
    "    \n",
    "    # Increase the counter    \n",
    "    chunk_n += 1\n",
    "\n",
    "    # Update the progress\n",
    "    if chunk_n % 1000 == 0:\n",
    "        print \"Chunks sent until this moment: \" + str(chunk_n)\n",
    "        sys.stdout.flush()\n",
    "\n",
    "print \"All chunks have been sent\"\n",
    "sys.stdout.flush()\n",
    "# Get the results from each engine and accumulate in global_result\n",
    "for engine in engines.ids:\n",
    "    # Be sure that all async tasks are finished\n",
    "    while ( not async_tasks[engine].ready() ):\n",
    "        time.sleep(1)\n",
    "    global_result += engines[engine].pull('local_total', block=True)\n",
    "\n",
    "#Close the file\n",
    "f.close()\n",
    "\n",
    "print \"Total number of chunks processed: \" + str(chunk_n)\n",
    "print \"---------------------------------------------\"\n",
    "print \"Agregated dictionary\"\n",
    "print \"---------------------------------------------\"\n",
    "print dict(global_result)\n",
    "\n",
    "print \"Time in seconds: \", time.time() - t0\n",
    "sys.stdout.flush()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The results of the experiments performed with this code can be seen in the corresponding chapter of the book."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}