5_High_Performance_Python.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# High-Performance Python\n",
    "Once in awhile, you will hear that \"oh Python is so slow or not performant enough\" or \"Python is not for <i>enterprise</i> use cases since it is not fast enough\" or \"Python is a language for script kiddies and at best \"glue\" code.\"  \n",
    "ZOMG! Scala's functional API is awesome and stands on the shoulders of the JVM. Go(lang) has goroutines that beats the pants off of Python's multithreading. Rust--I don't know what it does but it sounds trendy. 🙄 \n",
    "\n",
    "Spare me the faux concern ~~you pretentious C++ developer~~ ... I mean good question! Python supports rapid speed of development and experimentation, expressiveness and conciseness of syntax, and high-performance frameworks for speed. At a high-level, speed up Python performance on 1 machine using vectorization, multi-threading, multi-processing, and asynchronous programming. On multiple machines, you have cluster/distributed computing frameworks: Apache Spark, Dask, Apache Beam, Ray, and others. On cloud services, you have containers and scalable, serverless computing. In this tutorial, I will show you various ways to make Python fast and scalable, so performance is not your bottleneck.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Debunking the \"Slow\" Python Myth\n",
    "What do you mean by slow? Do you mean that Python runs slow, so it would cost more money when renting EC2s?  \n",
    "Remember that compute time is cheap and developer time is relatively expensive. Suppose a standard laptop has 4 CPUs and 16 GB of RAM. The equivalent EC2 instance on AWS rents for just \\\\$0.154 per hour; an engineer costs \\\\$50 per hour. So for each hour of developer time you waste, you could have rented a machine for 300 hours--or you can rent 300 machines for 1 hour. A 40 hour week for on EC2 would be \\$6--cheaper than your daily lunch!  \n",
    "Some \"faster\" languages are notoriously verbose, so you end up with pages and pages of code. Python's ease of use and expressivenes gives you a super power: rapid speed of development, which gives you a second super power: the ability to beat your competition to the market. Way of the Pro! Rapid experimentation allows you to out-maneuver your competition with better and newer features. Python's readability offers maintainability since developers come and go. Basically, can you write your code faster than somebody else: can you do something in 1 hour that somebody else takes 5?  \n",
    "\n",
    "If you need speed, Guido recommends you delegate the performance critical part to Cython:  \n",
    "```\"At some point, you end up with one little piece of your system, as a whole, where you end up spending all your time. If you write that just as a sort of simple-minded Python loop, at some point you will see that that is the bottleneck in your system. It is usually much more effective to take that one piece and replace that one function or module with a little bit of code you wrote in C or C++ rather than rewriting your entire system in a faster language, because for most of what you're doing, the speed of the language is irrelevant.\" --Guido van Rossum```  \n",
    "I won't mention Cython in the tutorial since it is rarely ever used--I have never personally met anybody who wrote their custom Cython code. Things that need to be performant in Python are already written in Cython: numpy, XGBoost, SpaCy, etc. The high level idea for Cython is that Python is actually written on top of C, so for speed you can write directly in C (or C++) and import it to Python.  \n",
    "\n",
    "If you are concerned about speed, do you know what is faster than Python? Fortran.  Then you can brush up on your Fortran to get hired by ... nobody (or worse, mainframe specialists 🤢).  \n",
    "\n",
    "Speed of the language primarily helps with CPU bound problems and not IO bound problems--in IO bound problems, any programming language will perform roughly the same since the majority of the time, you are waiting for a network response.  \n",
    "\n",
    "Finally, you don't really care about the raw speed of a language. You care about the expressioness and functionality of the language: its ecosystem, user-base, and mindshare. Whatever problem you can think of, Python probably has a library for that. Which you can pip install. So you can write minimal code. With which you solve your problem. So you can call it a day. Like Newton, when you use Python, you stand on the shoulders of giants. Things just work! Imagine if you couldn't just import sklearn and run Random Forest but had to write your own random forest algorithm. Actually, don't...  \n",
    "Here are some libraries just for scientific computing.  \n",
    "<p align=\"center\"><img src=\"images/scientific-ecosystem.png\" width=600></p><br>  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import antigravity  # a cute Easter egg; things in Python are so easy. Python has \"batteries included\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The Big O, Not Just Japanese Batman\n",
    "<p align=\"center\"><img src=\"images/The-Big-O.jpg\" width=300></p>\n",
    "\n",
    "Data structures and algorithms (ie your functions) have different performance characteristics that can be measured by how much memory (space) it takes to do something and how many steps (time) it takes to do something. Fancy people use the terms <i>space and (run)time complexity</i>, respectively. Don't be intimidated by the fancy names; I assure you it's a simple idea.  \n",
    "<b>Big O notation</b>: is used to denote the worst case scenario of how resource utilization (memory or number of steps) will grow with respect to the number of elements in your input data.  \n",
    "Note: O(n) is pronounced oh-enn.\n",
    "\n",
    "### Time Complexity of Data Structures\n",
    "Let's walk through some simple examples.  \n",
    "* list:\n",
    "    * to append to the end of a list, it is an (amoritized) O(1) operation. It is a fast operation.\n",
    "    * to append/insert to the beginning of a list, it takes O(n) operations because a new list is created, new element is inserted, and all the elements from the original list is copied over. It is slow in comparison to append to the end of the list.\n",
    "    * to check if an element is in a list (ie test for membership), it takes O(n) operations since the worst case is that  element is not in the list, so you have to check every element of the list.\n",
    "    * to get an element at a known, specific ith index, it is an O(1) operation. Fast\n",
    "* dict:\n",
    "    * to check if a key is in a dictionary: O(1) operation because dict keys are hashable (and unique).\n",
    "    * to check if a value is in a dictinoary: O(n) operations because dict values are not hashable (and do not have to be unique).\n",
    "\n",
    "Ideally all your code takes as little memory and as few steps as possible (ie O(1) for instantenous results). In practice, code uses non-trivial amounts of memory and computation. Sometimes, you can consider a tradeoff between space and time requirements. For example, if you are using a list and often have to append to the front and the end of the list, then you are better off with a data structure called double-ended queue (AKA deque). A deque have rapid `append()` and `appendleft()` methods as well as `pop()` and `popleft()`, since these operations are all O(1). However, a deque has slow access to the ith element, which is now a O(n) operation. You have to figure out which data structure gives you the best benefit given tradeoffs/constraints."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Time Complexity of Algorithms\n",
    "As with data structures, algorithms also have space and time complexity. An algorithm is just a fancy word for a set of steps to get you to the right answer: sorting algorithm, sudoku solver algorithm, determining the winner in a tournament algorithm.\n",
    "\n",
    "When an engineer says something is *(computationally) cheap*, that is fancy lingo for if the calculation uses little resources (low space complexity) and is fast (low time complexity). If a calculation is *(computationally) expensive*, it means it takes a lot of memory or is slow. Ideally you want your algorithm to be cheap.\n",
    "\n",
    "There are orders of magnitudes for complexity of algorithms (and data structures):\n",
    "* `O(1)` (AKA constant time): the best (that you can possibly do)!\n",
    "* `O(log(n))` (AKA logarithmic time): good\n",
    "* `O(n)` (AKA linear time): still good\n",
    "* `O(n*log(n))`: fair, use sparingly\n",
    "* `O(n^2)` (AKA quadratic time): bad, use very sparingly. If the exponent is any number other than 2 (ie 3, 4, ...), then it is called polynomial time\n",
    "* `O(2^n)` (AKA exponential time) and `O(n!)`: horrible, ideally you never have use an algorithm (or data structure) that has these complexities because it will be slow and resource intensive.\n",
    "    \n",
    "Basically, you want your algorithm (or data structure) to *scale* well with the size of your input data. To visualize how important the scaling factor, Big O, is to your algorithm, here's a chart. Stay in the green.  \n",
    "The chart and a more elaborate explanation of Big O is available at [Understanding time complexity with Python examples](https://towardsdatascience.com/understanding-time-complexity-with-python-examples-2bda6e8158a7)\n",
    "<p align=\"center\"><img src=\"images/Big_O_magnitudes.png\" width=800></p><br>\n",
    "\n",
    "Here is simple algorithm that concatenates strings together: it exhibits quadratic, O(n^2) time complexity."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ\n",
      "['i', 'h', 'b', 's', 'v', 'm', 'U', 'q', 's', 'J']\n"
     ]
    }
   ],
   "source": [
    "from string import ascii_letters\n",
    "from random import choices\n",
    "\n",
    "print(ascii_letters)\n",
    "lots_of_letters = choices(ascii_letters, k=100000)\n",
    "print(lots_of_letters[:10])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Wall time: 171 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# functional style: quadratic time\n",
    "from functools import reduce\n",
    "\n",
    "result_FP = reduce(lambda accumulated_string, char: accumulated_string + char, lots_of_letters)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Wall time: 31 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# procedural style: quadratic time\n",
    "result_procedural = \"\"\n",
    "for char in lots_of_letters:\n",
    "    result_procedural += char"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here's a smarter option that gives you linear, O(n) time complexity."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Wall time: 1 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "result_OOP = \"\".join(lots_of_letters)  # OOP is much faster since runtime is O(n)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result_FP == result_procedural == result_OOP  # proof that the results are the same"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\"Smart data structures and dumb code works a lot better than the other way around.\" -from *The Cathedral and the Bazaar* by Eric S. Raymond\n",
    "\n",
    "When possible, pick a good data structure that fits your problem--because there are a lot of data structures that do cool and useful things. Using the right data structure is the low-hanging fruit that requires minimum effort. If your logic is fundamentally complicated, then you will have to get creative and write (and optimize) a smart algorithm--that is more work. Of course, the quote is simply a rule of thumb--right most of the time, but not all the time. Use good judgment.\n",
    "\n",
    "Let me give you a practical example. Often in technical interview questions, the interviewer will ask you to write/whiteboard an algorithm. This is an interview question I received many years ago when I was a beginner: \"Given a list of words, find which words are anagrams of each other.\" Here is my very crude solution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'rats': {'arts', 'star'}, 'arts': {'star'}, 'god': {'dog'}}"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from collections import Counter\n",
    "\n",
    "def find_anagrams(list_of_words):\n",
    "    anagrams = {}\n",
    "    for i, current_word in enumerate(list_of_words):\n",
    "        for following_word in list_of_words[i+1:]:\n",
    "            if Counter(current_word) == Counter(following_word):\n",
    "                if current_word in anagrams:\n",
    "                    anagrams[current_word].add(following_word)\n",
    "                else:\n",
    "                    anagrams[current_word] = {following_word}\n",
    "    return anagrams\n",
    "\n",
    "find_anagrams([\"rats\", \"arts\", \"star\", \"god\", \"dog\", \"apple\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "My algorithm is very inefficient since it has O(n^2) time complexity: it has a double for-loop. And it is duplicating the letter counts (ie calling *Counter()*) multiple times for the same word. Finally, the output isn't great.\n",
    "\n",
    "The correct way (Way of the Pro) is to find some way to transform the data into what I call a \"canonical form\" or \"canonical representation\" of the data that is invariant to differences that don't matter. For example, in this anagram problem, it does not matter what order the letters are in the word. Hence, to get the \"canonical form\", you sort the letters in the word."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['rats', 'arts', 'star'], ['god', 'dog']]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from collections import defaultdict\n",
    "\n",
    "def find_anagrams_optimized(list_of_words):\n",
    "    anagrams = defaultdict(list)\n",
    "    for word in list_of_words:\n",
    "        word_with_letters_sorted = \"\".join(sorted(word))\n",
    "        anagrams[word_with_letters_sorted].append(word)\n",
    "    return [val for val in anagrams.values() if len(val) > 1]\n",
    "\n",
    "find_anagrams_optimized([\"rats\", \"arts\", \"star\", \"god\", \"dog\", \"apple\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that I used both:\n",
    "* a better data structure: default dict (with *list* as default) allows easy appending vs a regular dict\n",
    "* a better algorithm: that has linear O(n) time complexity instead of O(n^2). No redundant, repeated computations\n",
    "\n",
    "Moreover, the new function uses half the lines of code to get better results, so it is more readable and maintainable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Vectorized Calculations with SIMD\n",
    "vectorization with numpy and pandas: SIMD  \n",
    "like R's apply. pd.DataFrame.apply is not vectorized but without the speed performance. It's a convenience function.  \n",
    "CPU vs GPU: ALU vs FPU  \n",
    "order of magnitude of speed: CPU, RAM, storage, network: have image  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Multi-threading vs Multi-processing\n",
    "Multithreading and multiprocessing is when you hope to speed things up rather than wait for things to execute serially (ie a for loop). Multithreading is when you have multiple threads in 1 process, and thus the threads have access to the same variables and objects in memory--this is called *concurrency*. Multiprocessing is when you have multiple processes where each Python process/interpreter/session has its own memory and is running independently of each other--this is called true *parallelism*. Multithreading is faster to start up (<100 ms), but in most cases the right thing to do is use multiprocessing. Multithreading is used for *IO bound* tasks whereas multiprocessing is used for *CPU bound* tasks.\n",
    "<table><tr>\n",
    "    <td> <img src=\"images/concurrency.jpg\" alt=\"Concurrency\" style=\"width: 500px;\"/> </td>\n",
    "    <td> <img src=\"images/parallelism.jpg\" alt=\"Parallelism\" style=\"width: 500px;\"/> </td>\n",
    "</tr></table>\n",
    "Beautiful visual diagrams from https://medium.com/rungo/achieving-concurrency-in-go-3f84cbf870ca"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Multithreading\n",
    "Within 1 Python process, you can create multiple threads. Python has a Global Interpreter Lock (aptly shortened to GIL) that only allows 1 thread to be run at any given time, so you are switching between threads--this is called context switching. So why would you use multithreading instead of regular ol' serial execution (ie for loop)?\n",
    "<img src=\"images/multithreaded_process.png\" alt=\"Context Switching Overhead\" style=\"width: 300px;\"/>\n",
    "Multithreading is good for *IO bound* tasks. The typical example of IO bound task is scraping/downloading websites.\n",
    "Suppose you have 100 urls you want to scrap and each url takes 1 second to scrap. Serially scraping the urls will take 100 seconds. However, if you multithread, IO tasks release the GIL, so you can scrap multiple urls \"at the same time.\" Really, the bulk of the time is waiting for the response of data from the url (as any CPU time is virtually instantaneous), so multithreading can shrink that runtime significantly. Remember, you are still just using 1 CPU. Scraping websites does not take a lot of CPU power. Heavy number crunching is called *CPU bound* tasks, and multithreading does not help--and in fact, will be slower than serial execution.\n",
    "\n",
    "A Python process uses 1 CPU, so each thread uses a *time slice* of the CPU. The thread that acquires the GIL gets to run for a few milliseconds, then is paused, and then releases the GIL. Another thread acquires the GIL, gets to run for a few milliseconds, then is paused, and then releases the GIL, etc. In practice, because only 1 thread is executed at any 1 time, multithreading is in some sense running \"serially\"--2 things are happening after each other, 2 things are NOT happening at exactly the same time. Imagine if you are the CPU and are multi-tasking multiple threads, you can be making a phone call, responding to an email, watching TV. You are switching between those threads very quickly (which are interweaved), so it gives the *impression* that you are doing them \"at the same time.\" However, you are only doing 1 thing at a time.\n",
    "\n",
    "You don't know which thread is next, you don't know when the current thread is going to be paused, and you don't know which line of code the next thread will resume from. Well, you might think this is a bad idea--and you are correct.\n",
    "* Newbie: multithreading sounds like a bad idea.  \n",
    "* Intermediate: multithreading sounds like it could be good idea.  \n",
    "* Pro: multithreading is definitely a bad idea!\n",
    "\n",
    "In fact, multithreading is such a bad idea that Mozilla has this sign at their office. If you want to skip this entire section on multithreading, feel free to do so. This section is to edify you on *not* using this technique.\n",
    "<img src=\"images/be_this_tall_to_use_multithreading.jpg\" alt=\"Don't use multithreading\" style=\"width: 450px;\"/> \n",
    "\n",
    "\n",
    "\n",
    "Why is multithreading a bad idea? Multiple threads have access to the same/shared objects in memory, so different threads can modify the shared object. The shared object can be corrupted due to the unexpected order of thread execution--this is called a **race condition**. The typical example is a bank transaction. Suppose Alice and Bob are 2 threads that share 1 bank account, and both want to withdraw money \"at the same time\" (within milliseconds of each other). Here's is the linearized timeline (since only 1 thread is running at any moment in time) of Alice and Bob triggering a race condition and corrupting the state of the shared object: the bank account balance. In blockchain, this is called the double spend problem.\n",
    "* beginning bank account balance: \\$100\n",
    "* Alice: begins action of withdrawing \\$60\n",
    "* Alice: bank checks that balance (of \\$100) >= \\\\$60: yes!\n",
    "* Alice: ATM gives Alice \\$60\n",
    "* Bob: begins action of withdrawing \\$70\n",
    "* Bob: bank checks that balance (of \\$100) >= \\\\$70: yes!\n",
    "* Bob: ATM gives Bob \\$70\n",
    "* Alice: bank account decrements balance to \\$40 (\\\\$100 - \\\\$60)\n",
    "* Bob: bank account decrements balance to \\$30 (\\\\$1000 - \\\\$70)\n",
    "* ending bank account balance: \\$30 (even though \\\\$130 was successfully cashed out and that the bank should have prevented Bob from withdrawing money in the first place due to insufficient funds)\n",
    "\n",
    "\n",
    "Race conditions can also occur with the opposite problem where if Alice and Bob are both depositing money \"at the same time\" (basically milliseconds of each other) as multiple threads.\n",
    "* beginning bank account balance: \\$100\n",
    "* Alice: begins action of depositing \\$60\n",
    "* Alice: bank acccount gets current balance (\\$100) and is about to increment\n",
    "* Bob: begins action of depositing \\$70\n",
    "* Bob: bank account gets current balance (\\$100) and is about to increment\n",
    "* Bob: deposit is successful. Bank account increments balance to \\$170 (\\\\$100 + \\\\$70)\n",
    "* Alice: deposit is successful. Bank account increments balance to \\$160 (\\\\$100 + \\\\$60)\n",
    "* ending bank account balance: $160 (because it just so happens that Alice's thread finished last)\n",
    "\n",
    "As you can see, multithreaded actions on a shared object are not \"thread-safe\", as the actions (ie the functions) are not atomic. Thread 1 might do something halfway and pause, and thread 2 can do something else on the same object. Not even functional programming can save you now. This problem would have not occurred if you just did things serially (ie a for loop).\n",
    "\n",
    "The typical example of multithreading triggering a race condition is the bank transaction. Let's do something more fun: tug of war! A deque is double-ended queue--you can think of it as a list with fast O(1) operation for `popleft()`, `pop()` (ie `popright()`), `appendleft()`, and `append()` (ie `appendright()`). Notice that the deque length does not always stay at 9,999,999 (deque length - 1 for the popped value) and the resulting deque has lost its order."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "def popleft_and_appendright(dq, iterations, checkpoint_iterations):\n",
    "    for i in range(iterations):\n",
    "        value = dq.popleft()\n",
    "        if i % checkpoint_iterations == 0:\n",
    "            print(\"length of deque: {}\".format(len(dq)))\n",
    "        dq.append(value)\n",
    "    print(\"final length of deque: {}\".format(len(dq)))\n",
    "\n",
    "def popright_and_appendleft(dq, iterations, checkpoint_iterations):\n",
    "    for i in range(iterations):\n",
    "        value = dq.pop()\n",
    "        if i % checkpoint_iterations == 0:\n",
    "            print(\"length of deque: {}\".format(len(dq)))\n",
    "        dq.appendleft(value)\n",
    "    print(\"final length of deque: {}\".format(len(dq)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Threads are ready to start!\n",
      "length of deque: 9999999Thread 1 is started\n",
      "\n",
      "length of deque: 9999998Thread 2 is started\n",
      "\n",
      "Threads 1 and 2 are probably still running\n",
      "length of deque: 9999998\n",
      "length of deque: 9999998\n",
      "length of deque: 9999998length of deque: 9999998\n",
      "\n",
      "length of deque: 9999998length of deque: 9999998\n",
      "\n",
      "length of deque: 9999998length of deque: 9999998\n",
      "\n",
      "length of deque: 9999998length of deque: 9999998\n",
      "\n",
      "length of deque: 9999999length of deque: 9999998\n",
      "\n",
      "length of deque: 9999999\n",
      "length of deque: 9999998\n",
      "length of deque: 9999998length of deque: 9999998\n",
      "\n",
      "length of deque: 9999999\n",
      "length of deque: 9999998\n",
      "final length of deque: 10000000\n",
      "Thread 1 is done for sure\n",
      "final length of deque: 10000000\n",
      "Thread 2 is done for sure\n"
     ]
    }
   ],
   "source": [
    "from collections import deque\n",
    "from threading import Thread\n",
    "\n",
    "# double-ended queue (deque) has O(1) operation for popleft/popright, appendleft/appendright\n",
    "dq = deque(range(10000000))  # this is the shared object\n",
    "\n",
    "thread1 = Thread(target=popleft_and_appendright, args=(dq, 10000000, 1000000))\n",
    "thread2 = Thread(target=popright_and_appendleft, args=(dq, 10000000, 1000000))\n",
    "\n",
    "print(\"Threads are ready to start!\")\n",
    "thread1.start()\n",
    "print(\"Thread 1 is started\")\n",
    "thread2.start()\n",
    "print(\"Thread 2 is started\")\n",
    "\n",
    "print(\"Threads 1 and 2 are probably still running\")\n",
    "thread1.join()\n",
    "print(\"Thread 1 is done for sure\")\n",
    "thread2.join()\n",
    "print(\"Thread 2 is done for sure\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Are all the elements in order? Counter({True: 9819747, False: 180253})\n",
      "first 10: [9976631, 31728, 27419, 0, 9980878, 2, 3, 4, 5, 6]\n",
      "last 10: [9999987, 9999988, 9999989, 9999990, 9999991, 9999992, 9999993, 9999994, 9999995, 9999996]\n"
     ]
    }
   ],
   "source": [
    "from collections import Counter\n",
    "\n",
    "print(\"Are all the elements in order? {}\".format(Counter(x == y for x, y in zip(dq, range(len(dq))))))\n",
    "print(\"first 10: {}\".format(list(dq)[:10]))\n",
    "print(\"last 10: {}\".format(list(dq)[-10:]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here's another example of the race condition: basically the bank account example. The bank account starts at 0 and should end at 0--but where it ends up, nobody knows! 🤔\n",
    "\n",
    "Also notice that thread is triggered by calling `thread1.start()`. Before this moment, the thread is not executed--it is just sitting there. When the line `thread1.start()` is run, notice that it is \"non-blocking\", which means that the thread kind of runs in the \"background\", so Python can execute the next line before thread1 finishes its work. Just like thread1, `thread2.start()` is non-blocking, so you can run the next line of code (such as `print()`). \n",
    "\n",
    "So how do you know when a thread is done running (because theoretically Python could have run the final line of the .py file but the thread is still chugging along)? You can call `thread1.join()`, which IS a blocking operation--Python will wait until that thread has completed its task. If the thread is already done, then `.join()` will finish very quickly. If the thread is still running, then `.join()` will continue to run/block for an indeterminate amount of time before the next line is run."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def incrementer(lst, iterations, checkpoint_iterations):\n",
    "    for i in range(iterations):\n",
    "        lst[0] += 1\n",
    "        if i % checkpoint_iterations == 0:\n",
    "            print(\"lst: {}\".format(lst))\n",
    "    print(\"final lst: {}\".format(lst))\n",
    "\n",
    "def decrementer(lst, iterations, checkpoint_iterations):\n",
    "    for i in range(iterations):\n",
    "        lst[0] -= 1\n",
    "        if i % checkpoint_iterations == 0:\n",
    "            print(\"lst: {}\".format(lst))\n",
    "    print(\"final lst: {}\".format(lst))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "lst: [1]\n",
      "lst: [13564]\n",
      "lst: [207439]\n",
      "lst: [227883]\n",
      "lst: [673279]\n",
      "lst: [454662]\n",
      "lst: [721027]\n",
      "lst: [638504]\n",
      "lst: [610278]\n",
      "lst: [540047]\n",
      "lst: [587566]\n",
      "lst: [645003]\n",
      "lst: [797312]\n",
      "lst: [767610]\n",
      "lst: [854488]\n",
      "lst: [493231]\n",
      "lst: [521828]\n",
      "lst: [606092]\n",
      "lst: [704832]\n",
      "lst: [569396]\n",
      "final lst: [630574]\n",
      "final lst: [-273920]\n"
     ]
    }
   ],
   "source": [
    "lst = [0]  # this is the shared object\n",
    "\n",
    "thread1 = Thread(target=incrementer, args=(lst, 10000000, 1000000))\n",
    "thread2 = Thread(target=decrementer, args=(lst, 10000000, 1000000))\n",
    "\n",
    "thread1.start()\n",
    "thread2.start()\n",
    "\n",
    "thread1.join()\n",
    "thread2.join()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The purpose of this example is that multithreading causes function calls to be \"non-atomic\". Multiple threads exist in the same memory and have access to the same variables. 2 (or more) threads compete to access and mutate variables. Suppose the first thread is executing some lines that update a list, that thread can be paused at any time in the function--and you have no idea where. Then a second thread can  be activated and then mutate that same variable. By the time, Python goes back to the first thread, the variable can already have changed.\n",
    "\n",
    "How to make threads not trip over each other on getting/setting the same variable? You create a lock over a piece of code such that the thread has to acquire the lock before it can run that piece of code. When that piece of code is finished, then the thread can release the lock. However, a natural question is that wouldn't the threaded code run effectively sequentially since no 2 threads can run the same code protected by the lock at the same time? Yep, you are correct. Adding locks basically makes multithreaded code run sequentially.\n",
    "\n",
    "The only redeeming thing is that if you run threads that apply the same operation on different elements, then you can use `ThreadPool`, which also can function as a context manager."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "lst: [1]lst: [2]\n",
      "\n",
      "lst: [1194685]\n",
      "lst: [1313960]\n",
      "lst: [2481346]lst: [2504151]\n",
      "\n",
      "lst: [3845698]lst: [3864301]\n",
      "\n",
      "lst: [4996320]lst: [5056937]\n",
      "\n",
      "lst: [6183852]\n",
      "lst: [6272983]\n",
      "lst: [7384063]\n",
      "lst: [7437495]\n",
      "lst: [8687036]\n",
      "lst: [8695291]\n",
      "lst: [9921856]\n",
      "lst: [10007207]\n",
      "lst: [11182370]\n",
      "lst: [11243052]\n",
      "final lst: [12417084]\n",
      "final lst: [12518556]\n",
      "12518556\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing.pool import ThreadPool\n",
    "\n",
    "lst = [0]  # this is the shared object\n",
    "thread_pool = ThreadPool(2)  # creating 2 threads\n",
    "thread_pool.map(\n",
    "    func=lambda lst: incrementer(lst, 10000000, 1000000),  # hard coded the arguments `iterations` and `checkpoint_iterations`\n",
    "    iterable=[lst] * 2,  # both elements in the list point to the same object\n",
    ")\n",
    "thread_pool.close()  # remember to close the threadpool\n",
    "print(lst[0])  # find result should be 20000000 but it's not"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "lst: [1]lst: [2]\n",
      "\n",
      "lst: [1195744]\n",
      "lst: [1311003]\n",
      "lst: [2360466]\n",
      "lst: [2598903]\n",
      "lst: [3566836]\n",
      "lst: [3913564]\n",
      "lst: [4893890]\n",
      "lst: [5275909]\n",
      "lst: [6276431]\n",
      "lst: [6571892]\n",
      "lst: [7445152]\n",
      "lst: [7915186]\n",
      "lst: [8760764]\n",
      "lst: [9127098]\n",
      "lst: [10046542]\n",
      "lst: [10358581]\n",
      "lst: [11315160]\n",
      "lst: [11499093]\n",
      "final lst: [12558194]\n",
      "final lst: [12800597]\n",
      "12800597\n"
     ]
    }
   ],
   "source": [
    "# context manager syntax--better, no manual closing required!\n",
    "lst = [0]  # this is the shared object\n",
    "with ThreadPool(2) as thread_pool:\n",
    "    thread_pool.map(\n",
    "        func=lambda lst: incrementer(lst, 10000000, 1000000),  # hard coded the arguments `iterations` and `checkpoint_iterations`\n",
    "        iterable=[lst] * 2,  # both elements in the list point to the same object\n",
    ")\n",
    "print(lst[0])  # find result should be 20000000 but it's not"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, threads are *not* scalable. The scheduling of the threads (ie, which one comes next) is actually done by the OS, so although a couple milliseconds seems fast to humans, it is still slow. Computers operate at the nanosecond level (billions of operations per second). Switching between threads is called **context switching**. When there are a lot of threads (say 30), then the OS has to pick which thread out of the 30 runs next. In this case, more time is spent acquiring the GIL than actually doing work. It's like the Hunger Games of threads--may the odds be ever in your favor in acquiring the GIL. In this case, it might be better to use serial execution (for loop) compared to using 30 threads.\n",
    "In practice, acquiring the lock is not a costless operation. <img src=\"images/context_switching_cost.jpg\" alt=\"Context Switching Overhead\" style=\"width: 500px;\"/>\n",
    "\n",
    "\n",
    "So where does that leave us? Threads can mutate the same variables in memory and are non-blocking and may provide some speedup for IO operations. The simple answer is to not use multithreading. The more nuanced answer is don't use multithreading for real, production code--you can use multithreading on simple, toy projects where correctness is not essential. \n",
    "\n",
    "So why did I show all this *useless* stuff? It was to tee up for the *useful*: multiprocessing and asynchronous programming."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Multi-processing\n",
    "\n",
    "How do you get true parallelism (ie multiple things running at the *exact* same time)? You use multiple processes (multiprocessing), so there is no context switching and no problems with the Global Interpreter Lock (GIL). Each (sub)process is basically a separate Python process/kernel (with its own GIL) and can run on a different CPU--this is scheduled by the OS, so you don't need to do it yourself. You can think of multiprocessing as being able to create multiple Python (sub)processes (say 10) with very little code.\n",
    "\n",
    "Multiprocessing is very good for \"embarrassingly parallel\" problems, which can be subdivided and worked on separately \n",
    "and independently. If you want to see the predicted speedup of using multiprocessing, you can look at [Amdahl's law](https://en.wikipedia.org/wiki/Amdahl%27s_law), which is a formula about how much faster will multiprocessing be over regular ol' serial loops. Multiprocessing is not so good where the results of A depend on results of B, which depend on results of C. For example, multiprocessing is good for running many simulations, but it would not be good for something sequential like withdrawals from a bank account (because you want to know when the bank account runs out of money).\n",
    "\n",
    "A downside of multiprocessing is that each process has separate memory, not shared memory (which multithreading has). Hence, processes generally don't talk to each other. Communication is still possible through IPC (inter-process communication) or RPC (remote process communication), but these are not (explicitly) used often. The other small downside is the start up time for a new process--it takes on the order of 100 milliseconds whereas a new thread can be started much faster. But a 100 millisecond is approximately a blink of an eye.\n",
    "\n",
    "At a high level, you can think of multiprocessing as distributed functional programming because you are using `map()` over an iterable and the process pool will send each element of the iterable to a process. The process will apply the function onto the element and then return the result back to you.\n",
    "\n",
    "#### Sound Impressive While Doing Very Little (AKA Working Smarter)\n",
    "I want to share my dirty little secret. I ETLed (extract, transform, load) 150 Terabytes of data *without* Spark. To put that in perspective, that is the equivalent of 3,000 HD movies. I used 1 large machine (64 CPUs and 256 GB of RAM). The ETL job was to parse compressed zip files, perform some data transformation, and store into a database. In fact, my Python code was so performant that the machine running Python was only at ~50% CPU utilization but all the servers running NoSQL (HBase) were all maxed out at 100% CPU utilization. NoSQL is known for its rapid read and write speeds, so actually it was quite surprising that the NoSQL was the bounding constraint and not Python. This is a case study of Python being very performant and doing distributed, parallel computation.\n",
    "\n",
    "\n",
    "My other dirty secret is that multiprocessing is very easy to write. You can do impressive things with very little code. Don't work harder, work smarter.  \n",
    "Note: multiprocessing also has the context manager that closes the processes for you."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import multiprocessing\n",
    "\n",
    "def double(x): return 2 * x\n",
    "\n",
    "# unfortunately multiprocessing on Windows is buggy, you might need to put your code under if __name__ == \"__main__\":\n",
    "pool =  multiprocessing.Pool(processes=4)  # I decide to use 4 Python processes, which can use the full power of 4 CPUs\n",
    "result = pool.map(func=double, iterable=range(10))\n",
    "pool.close()  # remember to close your process pool when you are done\n",
    "result  # result is collected in the corresponding order as the input"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# context manager syntax, preferred syntax as automatically closes process pool regardless of hitting an exception\n",
    "with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:  # use all my CPUs\n",
    "    result = pool.map(func=double, iterable=range(10))\n",
    "\n",
    "result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`pool.map()` is a *blocking* method call. Python stays at that line until all the results are completed. However, you can make it non-blocking by `changing pool.map()` to `pool.map_async()`. Just make sure that you get all the results before you close the process pool."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "False\n"
     ]
    },
    {
     "ename": "ValueError",
     "evalue": "<multiprocessing.pool.MapResult object at 0x7fb73c3d3730> not ready",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mValueError\u001b[0m                                Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-3-964170f063e6>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      4\u001b[0m \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0masync_result\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mready\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 5\u001b[0;31m \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0masync_result\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msuccessful\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m  \u001b[0;31m# closed the process pool before all the results are completed\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      6\u001b[0m \u001b[0;31m#print(async_result.get())  # stuck forever as the results are not available\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/python3.8/multiprocessing/pool.py\u001b[0m in \u001b[0;36msuccessful\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    753\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0msuccessful\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    754\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mready\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 755\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mValueError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"{0!r} not ready\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    756\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_success\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    757\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mValueError\u001b[0m: <multiprocessing.pool.MapResult object at 0x7fb73c3d3730> not ready"
     ]
    }
   ],
   "source": [
    "with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:\n",
    "    async_result = pool.map_async(func=double, iterable=range(10))  # non-blocking so runs instantaneously\n",
    "\n",
    "print(async_result.ready())\n",
    "print(async_result.successful())  # closed the process pool before all the results are completed\n",
    "#print(async_result.get())  # stuck forever as the results are not available"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\n",
      "True\n",
      "[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]\n",
      "CPU times: user 0 ns, sys: 78.1 ms, total: 78.1 ms\n",
      "Wall time: 82.7 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:\n",
    "    async_result = pool.map_async(func=double, iterable=range(10))\n",
    "    async_result.wait()  # wait until all the results collected\n",
    "    # async_result.wait(1)  # you can control how long you want to wait for collecting results\n",
    "\n",
    "print(async_result.ready())\n",
    "print(async_result.successful())\n",
    "print(async_result.get())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]\n"
     ]
    }
   ],
   "source": [
    "with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:\n",
    "    async_result = pool.map_async(func=double, iterable=range(10))\n",
    "    print(async_result.get())  # get() is a blocking call but then it's basically equal to pool.map()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A common question question is: how many CPUs to use? It depends on the problem.\n",
    "* If your function is **CPU bound** (ie uses a lot of heavy computation), then you can use as many processes as up to the number of CPUs. The reason is that 1 function on 1 process will use up all the CPU cycles of 1 CPU.\n",
    "* If your function is **IO bound** (basically does read/writes and does not use a lot of CPUs), you can use more processes than number of CPUs. The reason is that multiple processes can be run on the same CPU since each process uses little CPU. IO bound problems, if they are not mission critical, can also be addressed with multithreading (mentioned above) or asynchronous programming (mentioned below).\n",
    "\n",
    "Some gotchas:\n",
    "* An additional constraint is RAM. If each process takes a lot of RAM, then together all the processes will take too much RAM and then the OS will terminate the process and you'll get `MemoryError`. Hence you might have to manually fiddle with the number of processes to get maximum performance while not hitting a RAM ceiling.\n",
    "* `multiprocessing` uses `pickle` to serialize between data and functions from your Python process to each subprocess. Hence objects that are not pickleable can cause trouble (ie lambda functions and generators)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "ename": "PicklingError",
     "evalue": "Can't pickle <function <lambda> at 0x7fb73c266a60>: attribute lookup <lambda> on __main__ failed",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mPicklingError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-6-4e2cdc3fc41a>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      2\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0;32mwith\u001b[0m \u001b[0mmultiprocessing\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPool\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mprocesses\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mmultiprocessing\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcpu_count\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0mpool\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 4\u001b[0;31m     \u001b[0masync_result\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mpool\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mmap\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfunc\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mdouble\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0miterable\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mrange\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;36m10\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m/usr/lib/python3.8/multiprocessing/pool.py\u001b[0m in \u001b[0;36mmap\u001b[0;34m(self, func, iterable, chunksize)\u001b[0m\n\u001b[1;32m    362\u001b[0m         \u001b[0;32min\u001b[0m \u001b[0ma\u001b[0m \u001b[0mlist\u001b[0m \u001b[0mthat\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0mreturned\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    363\u001b[0m         '''\n\u001b[0;32m--> 364\u001b[0;31m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_map_async\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfunc\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0miterable\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mmapstar\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mchunksize\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    365\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    366\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mstarmap\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mfunc\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0miterable\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mchunksize\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/python3.8/multiprocessing/pool.py\u001b[0m in \u001b[0;36mget\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m    766\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_value\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    767\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 768\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_value\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    769\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    770\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_set\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mi\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mobj\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/python3.8/multiprocessing/pool.py\u001b[0m in \u001b[0;36m_handle_tasks\u001b[0;34m(taskqueue, put, outqueue, pool, cache)\u001b[0m\n\u001b[1;32m    535\u001b[0m                         \u001b[0;32mbreak\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    536\u001b[0m                     \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 537\u001b[0;31m                         \u001b[0mput\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtask\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    538\u001b[0m                     \u001b[0;32mexcept\u001b[0m \u001b[0mException\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    539\u001b[0m                         \u001b[0mjob\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0midx\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mtask\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;36m2\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/python3.8/multiprocessing/connection.py\u001b[0m in \u001b[0;36msend\u001b[0;34m(self, obj)\u001b[0m\n\u001b[1;32m    204\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_check_closed\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    205\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_check_writable\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 206\u001b[0;31m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_send_bytes\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0m_ForkingPickler\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdumps\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mobj\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    207\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    208\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mrecv_bytes\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mmaxlength\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/python3.8/multiprocessing/reduction.py\u001b[0m in \u001b[0;36mdumps\u001b[0;34m(cls, obj, protocol)\u001b[0m\n\u001b[1;32m     49\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mdumps\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcls\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mobj\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mprotocol\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     50\u001b[0m         \u001b[0mbuf\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mio\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mBytesIO\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 51\u001b[0;31m         \u001b[0mcls\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mbuf\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mprotocol\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdump\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mobj\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     52\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mbuf\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgetbuffer\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     53\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mPicklingError\u001b[0m: Can't pickle <function <lambda> at 0x7fb73c266a60>: attribute lookup <lambda> on __main__ failed"
     ]
    }
   ],
   "source": [
    "double = lambda x: x * 2\n",
    "\n",
    "with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:\n",
    "    async_result = pool.map(func=double, iterable=range(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Advanced `multiprocessing` Arguments\n",
    "Everything above is probably all you need for most use cases. Here are some extra features of multiprocessing in case you want some power features.\n",
    "* `pool.map(chunksize=SOME_INTEGER_HERE)`: `chunksize` allows how many adjacent elements of the iterable to be sent to 1 process. Suppose you have 4 processes and the iterable is `range(40)` and `chunksize=5`, then elements 1,2,3,4,5 will be sent to 1 process and 6,7,8,9,10 will be sent to another process, etc. By default, chunksize=1.\n",
    "* `multiprocessing.Pool(maxtasksperchild=SOME_INTEGER_HERE)`: `maxtasksperchild` says how many times each process will run a chunk before that process is terminated and restarted. This is helpful in a long running job if you believe there might be a memory leak where the RAM utilization keeps increasing over time. By setting a small `maxtasksperchild`, then even if there is a small memory leak, the processs will be restarted to replace the previous process, so you won't hit your RAM ceiling. By default, maxtasksperchild=float(\"inf\"), so the same processes are used for all the elements of the iterable (ie there are no new processes that replace the old ones).\n",
    "* `multiprocessing.Pool(initializer=SOME_FUNCTION_HERE)`: `initializer` is a function that runs whenever a process is started up."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 2132\n",
      "1 2133\n",
      "2 2134\n",
      "3 2132\n",
      "4 2133\n",
      "5 2134\n",
      "6 2132\n",
      "7 2133\n",
      "8 2134\n",
      "9 2132\n",
      "[None, None, None, None, None, None, None, None, None, None]\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import time\n",
    "\n",
    "def print_process_id(num):\n",
    "    time.sleep(num)\n",
    "    print(num, os.getpid(), flush=True)\n",
    "\n",
    "with multiprocessing.Pool(processes=3) as pool:\n",
    "    async_result = pool.map(func=print_process_id, iterable=range(10))  # default chunksize=1, so process id alternates\n",
    "    print(async_result)  # just a bunch of Nones"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 2218\n",
      "1 2218\n",
      "2 2219\n",
      "4 2220\n",
      "3 2219\n",
      "6 2218\n",
      "5 2220\n",
      "8 2219\n",
      "7 2218\n",
      "9 2219\n"
     ]
    }
   ],
   "source": [
    "with multiprocessing.Pool(processes=3) as pool:\n",
    "    async_result = pool.map(func=print_process_id, iterable=range(10), chunksize=2)  # adjacent elements are run on the same process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 2304\n",
      "1 2305\n",
      "2 2306\n",
      "3 2304\n",
      "4 2305\n",
      "5 2306\n",
      "6 2342\n",
      "7 2351\n",
      "8 2360\n",
      "9 2342\n"
     ]
    }
   ],
   "source": [
    "with multiprocessing.Pool(processes=3, maxtasksperchild=2) as pool:\n",
    "    async_result = pool.map(func=print_process_id, iterable=range(10))  # elements 6 thru 9 go to new processes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Started process 2394\n",
      "Started process 2397\n",
      "0Started process 2402 \n",
      "2394\n",
      "1 2397\n",
      "2 2402\n",
      "3 2394\n",
      "Started process 2444\n",
      "4 2397\n",
      "Started process 2457\n",
      "5 2402\n",
      "Started process 2470\n",
      "6 2444\n",
      "7 2457\n",
      "8 2470\n",
      "9 2444\n"
     ]
    }
   ],
   "source": [
    "def start_process():\n",
    "    print(\"Started process {}\".format(os.getpid()), flush=True)\n",
    "\n",
    "# it is now more visible to see that the process is (terminated and) restarted\n",
    "with multiprocessing.Pool(processes=3, initializer=start_process, maxtasksperchild=2) as pool:    \n",
    "    async_result = pool.map(func=print_process_id, iterable=range(10), chunksize=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Multiprocessing Summary\n",
    "* Multiprocessing gives true parallelism (ie multiple things running at the *exact* same time by running on separate CPUs via separate processes).\n",
    "* Multiprocessing is great for CPU bound problem over an iterable. Basically you are loop over an iterable and each element is sent to a different process (which is run on a separate CPU), so you save time on getting the final results.\n",
    "\n",
    "#### Last Thoughts\n",
    "* I use `multiprocessing`. In Python 3, you have the option of using `multiprocessing.Pool` or `concurrent.futures.ProcessPoolExecutor`--both are basically the same thing.\n",
    "* Multiprocessing is used to run multiple things on **1** machine. If your job is super-duper intense (due to needing more CPUs or running out of RAM), then there are 2 options:\n",
    "    * **vertical scaling**: get a larger machine with more CPUs and RAM. Run everything on that 1 larger machine\n",
    "    * **horizontal scaling**: distributed the computations across multiple machines. Horizontal scaling is what the juicy section on *distributed computing frameworks* is all about (mentioned below)🤩!\n",
    "* If you have dependency across multiple things, you can use Dask delayed (also mentioned in the *distributed computing frameworks* section).\n",
    "* If you want to do IO bound problems (but you already know that multithreading is bad), then you want to use asynchronous programming."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Distributed Computing Frameworks\n",
    "In some sense, distributed computing (AKA *parallel computing* or *high-performance computing*) is glorified multiprocessing. Basically, distributed computing aims to *distribute* workloads horizontally instead of vertically. Horizontally scaling means using lots of small or medium-sized machines to do the work. Vertically scaling means using 1 very large machine (lots of RAM + CPU). Horizontal scaling is preferable because medium-sized machines are commodities--very easy to get access to and are cheap. Large machines are very expensive and sometimes an instance that large is not available.\n",
    "\n",
    "Like multiprocessing, distributed computing works very well with \"embarrassingly parallel\" problems, which are problems that can be subdivided and worked on separately and independently. Fortunately, distributed compouting frameworks can coordinate between the worker machines to aggregate results, which is not an embarrassingly parallel task. Take a look at [Amdahl's law](https://en.wikipedia.org/wiki/Amdahl%27s_law), which gives a formula about predicted speedup vs running a regular loop. A common lesson of distributed computing is that it isn’t about doing one thing exceedingly well; it’s about doing nothing poorly.\n",
    "\n",
    "In my opinion, distributed computing frameworks are glorified `map`/`filter`/`reduce` operations. Basically you want to chain a bunch of transformations together. If you have just 1 transformation, you might be better off with just `multiprocessing`. In distributed computing, you have a lot of elements (basically a list where the elements are distributed over multiple machines) and then you want to do 1 thing to it, get the outputted list, and then do another thing to that outputted list. Often the transformations are written using functional programming syntax and applied lazily, so they don't use memory or CPUs until you want to trigger the actual computation. Basically the purpose of the previous notebook (4_Functional_Python.ipynb) was to get you familiar with functional programming syntax, so you can use distributed computing frameworks.\n",
    "\n",
    "Popular distributed computing frameworks in Python are Apache Spark and Dask. Apache Beam and Ray are interested, but they do not have mainstream adoption--I added them just for fun. 🎉\n",
    "\n",
    "\n",
    "### MapReduce (old news)\n",
    "explain what is MapReduce and disadvantages  \n",
    "Out-of-core-computation (often mistakenly called out of memory computation): enables large matrix multiplications and querying large databases such as Hive query executed with MapReduce  \n",
    "A `reduce` operation that is commutative and associative can be partially parallelized using a technique call <i>combiner</i>.  \n",
    "Map reduce on-premise cannot be scalled up, so EMR on AWS gives you the ability to scale up when needed. Even Spark cluster is apportioned and probably isn't ideal for overallocation.  \n",
    "show mapreduce stage picture  \n",
    "\n",
    "### Apache Spark (industry standard 🏆)\n",
    "If you want to get a high-demand skill to get a new job, learn Spark. I always say: `If you know Spark, you can't get fired!`  \n",
    "Spark is basically MapReduce in memory. The bottleneck in large scale computation is getting the data to the CPU. RAM is way faster than disk; order of magnitudes, get picture of CPU cache, RAM, disk, network  \n",
    "Simple Spark  \n",
    "Spark and Dask: often a functional approach, functoolz syntax translates to Dask bags, FP allows embarrassingly parallel and laziness.  \n",
    "Don't need Scala for speed  \n",
    "Weird Java exceptions  \n",
    "Show Simple-Spark  \n",
    "Weakness: no ability to do multiple actions. The workaround is to split pipeline into 2 such as word count and character simultaneously and then do groupby.  \n",
    "DataFrame, RDD\n",
    "show Spark architecture picture  \n",
    "\n",
    "### Dask (new framework with growing community adoption 🌱)\n",
    "Distributed + Task  \n",
    "`If you learn Dask, then you know Spark. And if you know Spark, you can't get fired!`  \n",
    "Compare and contrast  \n",
    "No weird Java exceptions  \n",
    "Show Dask tutorial (tree aggregation example)  \n",
    "delayed and futures  \n",
    "x = x.map_blocks(compress).persist().map_blocks(decompress)  \n",
    "Can do multiple actions  \n",
    "DataFrame, array, bag, delayed\n",
    "show Dask architecture picture (basically the same thing)  \n",
    "\n",
    "### Apache Beam (theoretically interesting but nobody uses 😞)\n",
    "explain why it is cool: unified API btween Batch + Stream, auto-scaling, serverless, templates, can do multiple actions  \n",
    "pcollections, ParDo  \n",
    "explain its weaknesses: does not support data with schema, no graph algorithms like K-Means, also no concept of action so have to use trick to trigger finality  \n",
    "Throw picture in: 62 machines running for 30 minutes cost \\$2.50.  \n",
    "<p align=\"center\"><img src=\"images/Dataflow_pipeline.png\" width=1000></p>\n",
    "\n",
    "Spark, Dask, Beam  \n",
    "Features:  \n",
    "* **batch**: first classs, first class, first class\n",
    "* stream: first class, kinda, first class\n",
    "* **real time Dashboard (for debugging and throughput)**: yes, yes, yes\n",
    "* **interactive mode/REPL**: yes, yes, yes\n",
    "* **persist (memory and disk)**: memory+disk, only memory, no\n",
    "* **with persist, can execute different sections independently**: yes, yes, no\n",
    "* **can split 1 input to multiple output pipelines**: yes, yes, yes\n",
    "* **simultaneous actions on multiple output pipelines**: no, yes, yes\n",
    "* support for branching logic of pipeline: yes, yes, yes\n",
    "* support for iterative algorithms: yes, yes, no\n",
    "* **GROUP BY logic**: yes, yes, yes\n",
    "* **JOIN logic**: yes, yes, yes\n",
    "* **tagged output pipelines**: no, no, yes\n",
    "* **combiner/aggregate logic**: yes, yes, yes\n",
    "* **SQL support**: yes, kinda, no\n",
    "* distributed numpy array: no, yes, no\n",
    "* ML algorithms: yes, yes, no\n",
    "* futures: no, yes, no\n",
    "* delayed: no, yes, no\n",
    "* support to address \"hot keys\": yes, yes, yes\n",
    "* **shared variables**: broad variable and accumulators, no, idk\n",
    "* supports logging for debugging: yes, yes, yes\n",
    "* **supports multiple input/output file types**: yes, yes, yes\n",
    "* click-based or GUI-based template: no, no, yes\n",
    "* **auto scaling**: idk, idk, yes\n",
    "\n",
    "Other support:\n",
    "* integration: many, growing, primarily only GCP\n",
    "* pure Python (3): no, yes, yes\n",
    "* **easy to run/test on your own machine**: yes, yes, yes\n",
    "* performant local cluster on your machine: idk, yes, no\n",
    "* speed: Scala (fast), Python (less fast), Python (less fast)\n",
    "* **documentation**: great, great, lacking\n",
    "* user base: big, medium, low\n",
    "\n",
    "\n",
    "Which one to use: Spark is a very hot data engineering skill. I tell my teammates: \"if you know Spark, then you can't be fired.\" Since I work on Dask, I also tell my teammates: \"if you know Dask, then you know Spark. If you know Spark, then you can't be fired.\"  \n",
    "main problem is RAM vs CPU  \n",
    "MapReduce for RAM. The better way is to use Spark, that scales horizontally instead of vertically. It is still doing multiprocessing. for RAM\n",
    "\n",
    "\n",
    "### Ray (and Actor Model)\n",
    "like a blend of functional and OOP because it is stateful between transformations and dependency. Like Dask futures  \n",
    "Plasma -> Apache Arrow    \n",
    "totally novel architecture that overcomes IPC/RPC latency  \n",
    "show Ray architecture picture  \n",
    "\n",
    "\n",
    "These things are not built in a vacuum: progress stands on the shoulders of giants. Their histories are intertwined. Spark is basically mapreduce but in memory for the speed. Apache Beam originated from Google's internal Millwheel project and was marketed as true streaming. Spark basically copied Beam's approach. Dask was created originally for distributed arrays because Spark doesn't support it. Ray uses Redis and was probably inspired by Akka (framework in Scala). Dask adopted Ray's actor model. Basically copy ideas that work and dump the rest. A lot of frameworks that don't make the cut and just fade out. Other frameworks: Apache Apex (dead), Apache Gearpump (basically dead), Apache Samza (~~lame~~ only Java), Apache Flink (seems like a Hadoop thing, so basically for legacy applications)  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Asynchronous Programming/Event-Driven Programming\n",
    "asyncio is in some sense serial with indirection of a coroutine. asynchronous immplemented in 3 ways: callback (hell from nested structure), futures/promises (looks like [flat] method chaining) and flatter, async/await  \n",
    "yield from  \n",
    "yield and return in the same function in a coroutine  \n",
    "you don't need asyncio to do asynchronous programming  \n",
    "C10k problem  \n",
    "basically a queue--asynchronous because not executed immediately. The event loop is the buffer that manages queue  \n",
    "\n",
    "cooperative vs competitive multitasking  \n",
    "https://pybay.com/site_media/slides/raymond2017-keynote/async_examples.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Virtual Machines vs Containers\n",
    "VM vs Docker/containerization, multi-tenancy  \n",
    "serverless computing, scalable/decoupled, notifications and (asynchronous) PubSub connected to AWS Lambda (in some sense distributed but independent multiprocessing)  \n",
    "world has moved on from hardware -> co-location -> virtualization (rented VM) -> container -> function: https://read.acloud.guru/the-evolution-from-servers-to-functions-21833b576744  \n",
    "Focus on your business logic and delegate the rest to AWS. Who knows (and who cares) how its really implemented? Are you in the business of creating your private cloud/Hadoop or solving business problems that make $$$?  \n",
    "\n",
    "\n",
    "## What is the Cloud?\n",
    "There's an old saying: \"Nobody gets fired for buying IBM\". That was decades ago. I think the modern version would be \"Nobody gets fired for buying AWS\". The cloud is so versatile with all the services builtin, that if you are still buying hardware for your own datacenters (without good reasons), you *should* be fired. One of the clients I worked on is such an old-boys club that they still buy hardware, explain story.  \n",
    "AWS is really supply chain maximization, squeezing out efficiencies from elementary pieces: compute, storage, and networking. With these Lego blocks, they can create managed services that are just a copy of a Apache/open-source software and charge for it. The cloud doesn't cost you money--it saves you money. Overall win-win scenario. Like Costco or Trade Joe model. Show all the logos of services from AWS or GCP. Show the open source version and then AWS service    \n",
    "\n",
    "Why do you buy a car? Because it is better, faster, and/or cheaper than you can make it yourself. It just \"works out of the box.\" You could build a car yourself; it would be worse functionality, slower to get to market, more expensive build from scratch, and a nightmare to maintain. Just buy it off-the-shelf. Focus on your main problem. Don't roll your own--you must be high if you do. Plus, if something breaks, there's somebody you can ~~complain to~~ ask for help.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extra Resources\n",
    "* Raymond Hettinger, Keynote on Concurrency, PyBay 2017: https://www.youtube.com/watch?v=9zinZmE3Ogk and presentation materials at:\n",
    "    * multithreading: https://pybay.com/site_media/slides/raymond2017-keynote/threading.html\n",
    "    * multiprocessing: https://pybay.com/site_media/slides/raymond2017-keynote/process.html\n",
    "    * asynchronous: https://pybay.com/site_media/slides/raymond2017-keynote/async_examples.html    \n",
    "* About asynchornous programming from the creator of Twisted and key influencer to asyncio: https://glyph.twistedmatrix.com/2014/02/unyielding.html"
   ]
  }
 ],
 "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}