5_adv_features_6.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-09-01T07:17:17.571570Z",
     "iopub.status.busy": "2021-09-01T07:17:17.570576Z",
     "iopub.status.idle": "2021-09-01T07:17:17.585569Z",
     "shell.execute_reply": "2021-09-01T07:17:17.584584Z",
     "shell.execute_reply.started": "2021-09-01T07:17:17.571570Z"
    }
   },
   "source": [
    "<font size=6> <b> Advanced Python : week #6</b> </font>\n",
    "<div class=\"alert alert-block alert-success\">\n",
    "   Advanced Python features <br>\n",
    "    <ol>\n",
    "        <li> multi-threading / multi-processing </li>\n",
    "    </ol>\n",
    "</div>\n",
    "\n",
    "<p style=\"text-align:right;\"> sumyeon@gmail.com </p>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Multi-threading / Multi-processing\n",
    "\n",
    "> GIL (Global Interpreter Lock) \n",
    "- GIL prevents mulitple threads of Python code from running simultaneously\n",
    "- In python, only one thread can acquire the GIL, and other threads must acquire it before running. => no parallel execution\n",
    "- Howver, most external C-based functions release the GIL before running => most I/O related functions are for threading\n",
    "\n",
    "> Multi-threading vs. Multi-processing\n",
    "- it is very complicated whether to use parallelism and also which way - ex) multi-threading vs. multi-processing\n",
    "- the simplest way to determine the way is \"multi-threading for I/O intensive works and multi-processing for computation intensive works\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How to use - Summary"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### basic API\n",
    "- multiprocessing : multiprocessing.Process\n",
    " > process create, start, waiting <br>\n",
    " > NOTE) communication among processes : Queue/Pipe\n",
    "- threading : threading.Thread\n",
    " > thread create, start, wait\n",
    " "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### simple API\n",
    "- multiprocessing.Pool : Processing Pool\n",
    " > processing pool api for simpler use of process\n",
    "- multiprocesssing.pool.ThreadPool (= multiprocessing.dummy.Pool) : ThreadPool\n",
    " > threading pool api for simpler use of thread\n",
    " "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### simplest API\n",
    "- concurrent.futures.ThreadPoolExecutor/ProcessingPoolExecutor : Processing/Thread Pool Executor\n",
    " > simplest multi-processing/threading API "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## multiprocessing \n",
    "- a package that supports spawning processing \n",
    " > multiprocessing.Process for spawning a new process <br>\n",
    " > (process.)start() for start, and (process.)join() to wait for the end of execution <br>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:42:38.876019Z",
     "iopub.status.busy": "2021-10-06T13:42:38.875021Z",
     "iopub.status.idle": "2021-10-06T13:42:38.895030Z",
     "shell.execute_reply": "2021-10-06T13:42:38.894027Z",
     "shell.execute_reply.started": "2021-10-06T13:42:38.876019Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting multiprocessing_sample.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile multiprocessing_sample.py\n",
    "import time\n",
    "import multiprocessing\n",
    "\n",
    "\n",
    "def calc_square(numbers):\n",
    "    for i in numbers:\n",
    "        time.sleep(3)  # artificial time-delay\n",
    "        print('square: ', str(i * i))\n",
    "\n",
    "\n",
    "def calc_cube(numbers):\n",
    "    for i in numbers:\n",
    "        time.sleep(3)\n",
    "        print('cube: ', str(i * i * i))\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    arr = [2, 3, 8, 9]\n",
    "    \n",
    "    # creating two Process here p1 & p2\n",
    "    p1 = multiprocessing.Process(target=calc_square, args=(arr,))\n",
    "    p2 = multiprocessing.Process(target=calc_cube, args=(arr,))\n",
    "    \n",
    "    start = time.time()\n",
    "    # starting Processes here parallel by using start function.\n",
    "    p1.start()\n",
    "    p2.start()\n",
    "    \n",
    "    # this join() will wait until the calc_square() function is finished.\n",
    "    p1.join()\n",
    "    # this join() will wait unit the calc_cube() function is finished.\n",
    "    p2.join()\n",
    "    \n",
    "    print(\"Successes!\")\n",
    "    print(\"Elapsed time: \", time.time() - start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T12:09:16.986907Z",
     "iopub.status.busy": "2021-10-06T12:09:16.986907Z"
    }
   },
   "source": [
    "#### please open a terminal and run the multiprocessing_sample.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### threading\n",
    "- a package that supports spawning threads\n",
    " > threading.Thread for spawning a new process <br>\n",
    " > (therad.)start() for start, and (thread.)join() to wait for the end of execution <br>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:42:40.701637Z",
     "iopub.status.busy": "2021-10-06T13:42:40.701637Z",
     "iopub.status.idle": "2021-10-06T13:42:48.766643Z",
     "shell.execute_reply": "2021-10-06T13:42:48.764631Z",
     "shell.execute_reply.started": "2021-10-06T13:42:40.701637Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Calculate square numbers: \n",
      "Calculate cube numbers: \n",
      "square:  4\n",
      "cube:  8\n",
      "square:  9\n",
      "cube:  27\n",
      "square:  64\n",
      "cube:  512\n",
      "square:  81\n",
      "cube:  729\n",
      "Successes!\n",
      "Elapsed time:  8.043993949890137\n"
     ]
    }
   ],
   "source": [
    "import time\n",
    "import threading\n",
    "\n",
    "\n",
    "def calc_square(numbers):\n",
    "    print(\"Calculate square numbers: \")\n",
    "    for i in numbers:\n",
    "        time.sleep(2)  # artificial time-delay\n",
    "        print('square: ', str(i * i))\n",
    "\n",
    "def calc_cube(numbers):\n",
    "    print(\"Calculate cube numbers: \")\n",
    "    for i in numbers:\n",
    "        time.sleep(2)\n",
    "        print('cube: ', str(i * i * i))\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    arr = [2, 3, 8, 9]\n",
    "    \n",
    "    # creating two threads here t1 & t2\n",
    "    t1 = threading.Thread(target=calc_square, args=(arr,))\n",
    "    t2 = threading.Thread(target=calc_cube, args=(arr,))\n",
    "    \n",
    "    start = time.time()\n",
    "    # starting threads here parallel by using start function.\n",
    "    t1.start()\n",
    "    t2.start()\n",
    "    \n",
    "    # this join() will wait until the cal_square() function is finished.\n",
    "    t1.join()\n",
    "    # this join() will wait unit the cal_cube() function is finished.\n",
    "    t2.join()\n",
    "    \n",
    "    print(\"Successes!\")\n",
    "    print(\"Elapsed time: \", time.time() - start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### multiprocessing.Pool\n",
    "- a convenient way of parallelizing the execution of a fuction across multiple input values, distributing the input data across processes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:42:48.769636Z",
     "iopub.status.busy": "2021-10-06T13:42:48.768651Z",
     "iopub.status.idle": "2021-10-06T13:42:48.798635Z",
     "shell.execute_reply": "2021-10-06T13:42:48.794639Z",
     "shell.execute_reply.started": "2021-10-06T13:42:48.769636Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mppool_example.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile mppool_example.py\n",
    "from multiprocessing import Pool\n",
    "import time\n",
    "\n",
    "iterable = [1,2,3,4,5]\n",
    "\n",
    "def f(x):\n",
    "    time.sleep(2)\n",
    "    return x*x\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    start = time.time()\n",
    "    with Pool(5) as pool:\n",
    "        print(pool.map(f, iterable))  # pool.map return the list of return values\n",
    "    print(\"Elapsed time: \", time.time() - start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T12:09:16.986907Z",
     "iopub.status.busy": "2021-10-06T12:09:16.986907Z"
    }
   },
   "source": [
    "#### please open a terminal and run the mppool_example.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### multiprocesssing.pool.ThreadPool (= multiprocessing.dummy.Pool) \n",
    "- a convenient way of parallelizing the execution of a fuction across multiple input values, distributing the input data across threads"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:42:50.931123Z",
     "iopub.status.busy": "2021-10-06T13:42:50.930128Z",
     "iopub.status.idle": "2021-10-06T13:42:50.939125Z",
     "shell.execute_reply": "2021-10-06T13:42:50.938129Z",
     "shell.execute_reply.started": "2021-10-06T13:42:50.931123Z"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting mpthreadpool.py\n"
     ]
    }
   ],
   "source": [
    "from multiprocessing.pool import ThreadPool\n",
    "import time\n",
    "\n",
    "iterable = [1,2,3,4,5]\n",
    "\n",
    "def f(x):\n",
    "    time.sleep(2)\n",
    "    return x*x\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    start = time.time()\n",
    "    with ThreadPool(5) as p:\n",
    "        results = p.map(f, iterable)\n",
    "        print(results)\n",
    "    print(\"Elapsed time: \", time.time() - start)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T12:09:16.986907Z",
     "iopub.status.busy": "2021-10-06T12:09:16.986907Z"
    }
   },
   "source": [
    "#### please open a terminal and run the mpthreadpool.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Concurrent Futures\n",
    "- simpler interface that works very much the same for both multi-threading & multi-processing\n",
    "- in the below, we will show only ThreadPoolExecutor. way of using ProcessPoolExcutor will be pretty much same as that."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ThreadPoolExecutor\n",
    "- provides useful methods to execute callables asynchrhonously\n",
    "- <b>submit</b> : schedule the callable and return a Future which represents the execution of the callable\n",
    "- <b>as_completed</b> : Future generator which yield fututures as ordered by their completion\n",
    "- <b>map</b>    : schedule several execution of the callable on the given iterable, and then return an iterator of the results itself\n",
    "- <b>shutdown</b> : signal the executor to free up resources it is using when the currently pening futures are done"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:42:54.152559Z",
     "iopub.status.busy": "2021-10-06T13:42:54.152559Z",
     "iopub.status.idle": "2021-10-06T13:42:54.171542Z",
     "shell.execute_reply": "2021-10-06T13:42:54.169551Z",
     "shell.execute_reply.started": "2021-10-06T13:42:54.152559Z"
    }
   },
   "outputs": [],
   "source": [
    "import time\n",
    "import concurrent"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:43:16.386834Z",
     "iopub.status.busy": "2021-10-06T13:43:16.385822Z",
     "iopub.status.idle": "2021-10-06T13:43:16.408819Z",
     "shell.execute_reply": "2021-10-06T13:43:16.407827Z",
     "shell.execute_reply.started": "2021-10-06T13:43:16.386834Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total time 0.0039942264556884766\n",
      "f( 4 ) completed\n",
      "f( 3 ) completed\n",
      "f( 2 ) completed\n",
      "f( 1 ) completed\n"
     ]
    }
   ],
   "source": [
    "executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)\n",
    "\n",
    "def f(x):\n",
    "    time.sleep(10-x)\n",
    "    print('f(',x,') completed')\n",
    "    return x\n",
    "\n",
    "start =  time.time()\n",
    "a =  executor.submit(f, 1)\n",
    "b =  executor.submit(f, 2)\n",
    "c =  executor.submit(f, 3)\n",
    "d =  executor.submit(f, 4)\n",
    "print(f\"total time {time.time()-start}\")     "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:43:27.722088Z",
     "iopub.status.busy": "2021-10-06T13:43:27.721088Z",
     "iopub.status.idle": "2021-10-06T13:43:33.757040Z",
     "shell.execute_reply": "2021-10-06T13:43:33.755039Z",
     "shell.execute_reply.started": "2021-10-06T13:43:27.722088Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "f( 4 ) completed\n",
      "f(4) result is  4\n",
      "total time 6.018961191177368\n",
      "f( 3 ) completed\n",
      "f( 2 ) completed\n",
      "f( 1 ) completed\n"
     ]
    }
   ],
   "source": [
    "executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)\n",
    "\n",
    "start =  time.time()\n",
    "a =  executor.submit(f, 1)\n",
    "b =  executor.submit(f, 2)\n",
    "c =  executor.submit(f, 3)\n",
    "d =  executor.submit(f, 4)\n",
    "\n",
    "print(\"f(4) result is \", d.result())\n",
    "print(f\"total time {time.time()-start}\")     \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### concurrent.futures.as_completed\n",
    "- return future iterator which return the Future in the order of when they are done"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:43:44.709836Z",
     "iopub.status.busy": "2021-10-06T13:43:44.709836Z",
     "iopub.status.idle": "2021-10-06T13:43:53.736445Z",
     "shell.execute_reply": "2021-10-06T13:43:53.734448Z",
     "shell.execute_reply.started": "2021-10-06T13:43:44.709836Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "2\n",
      "3\n",
      "4\n",
      "total time 9.008634090423584\n"
     ]
    }
   ],
   "source": [
    "import concurrent.futures\n",
    "\n",
    "iterable = [1,2,3,4]\n",
    "def f(x):\n",
    "    time.sleep(10-x)\n",
    "    return x\n",
    "\n",
    "start = time.time()\n",
    "with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:\n",
    "    result_futures = list(map(lambda x : executor.submit(f,x), iterable))  # submit return Future\n",
    "    for future in result_futures:\n",
    "        print(future.result())\n",
    "print(f\"total time {time.time()-start}\")        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:43:57.480575Z",
     "iopub.status.busy": "2021-10-06T13:43:57.479573Z",
     "iopub.status.idle": "2021-10-06T13:44:06.509270Z",
     "shell.execute_reply": "2021-10-06T13:44:06.507274Z",
     "shell.execute_reply.started": "2021-10-06T13:43:57.480575Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4\n",
      "3\n",
      "2\n",
      "1\n",
      "total time 9.009706497192383\n"
     ]
    }
   ],
   "source": [
    "start = time.time()\n",
    "with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:\n",
    "    result_futures = list(map(lambda x : executor.submit(f,x), iterable))  # submit return Future\n",
    "    for future in concurrent.futures.as_completed(result_futures):\n",
    "        print(future.result())\n",
    "print(f\"total time {time.time()-start}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### map\n",
    "- map apply the executor to an iterble and return the generator of return value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:44:14.800771Z",
     "iopub.status.busy": "2021-10-06T13:44:14.799788Z",
     "iopub.status.idle": "2021-10-06T13:44:23.823157Z",
     "shell.execute_reply": "2021-10-06T13:44:23.822170Z",
     "shell.execute_reply.started": "2021-10-06T13:44:14.800771Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1\n",
      "2\n",
      "3\n",
      "4\n",
      "total time 9.005383968353271\n"
     ]
    }
   ],
   "source": [
    "start = time.time()\n",
    "with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:\n",
    "    result_gen = executor.map(f,iterable)  # submit return Future\n",
    "    for future in result_gen:\n",
    "        print(future)\n",
    "print(f\"total time {time.time()-start}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ProcessPoolExecutor\n",
    "-exactly same as ThreadPoolExecutor only except concurrent.futures.ProcessPoolExecutor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:44:23.824155Z",
     "iopub.status.busy": "2021-10-06T13:44:23.824155Z",
     "iopub.status.idle": "2021-10-06T13:44:23.838155Z",
     "shell.execute_reply": "2021-10-06T13:44:23.837158Z",
     "shell.execute_reply.started": "2021-10-06T13:44:23.824155Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting processpool_sample.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile processpool_sample.py\n",
    "\n",
    "import time\n",
    "import concurrent.futures\n",
    "\n",
    "iterable = [1,2,3,4]\n",
    "\n",
    "def f(x):\n",
    "    time.sleep(10-x)\n",
    "    return x\n",
    "\n",
    "def main():\n",
    "    start = time.time()\n",
    "    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:\n",
    "        result_futures = list(map(lambda x : executor.submit(f,x), iterable))  # submit return Future\n",
    "        for future in concurrent.futures.as_completed(result_futures):\n",
    "            print(future.result())\n",
    "    print(f\"total time {time.time()-start}\")\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Please open a terminal and run the command <it>\"python processpool_sample.py\"</it>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# [Example] Using multiprocessing/threading to make Pandas apply functions faster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:44:23.840156Z",
     "iopub.status.busy": "2021-10-06T13:44:23.839158Z",
     "iopub.status.idle": "2021-10-06T13:44:24.006155Z",
     "shell.execute_reply": "2021-10-06T13:44:24.005157Z",
     "shell.execute_reply.started": "2021-10-06T13:44:23.840156Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>feature1</th>\n",
       "      <th>feature2</th>\n",
       "      <th>feature3</th>\n",
       "      <th>feature4</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>a</th>\n",
       "      <td>1</td>\n",
       "      <td>11</td>\n",
       "      <td>1</td>\n",
       "      <td>1100</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>b</th>\n",
       "      <td>2</td>\n",
       "      <td>12</td>\n",
       "      <td>32</td>\n",
       "      <td>1200</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>c</th>\n",
       "      <td>3</td>\n",
       "      <td>13</td>\n",
       "      <td>243</td>\n",
       "      <td>1300</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>d</th>\n",
       "      <td>4</td>\n",
       "      <td>14</td>\n",
       "      <td>1024</td>\n",
       "      <td>1400</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>e</th>\n",
       "      <td>5</td>\n",
       "      <td>15</td>\n",
       "      <td>3125</td>\n",
       "      <td>1500</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>f</th>\n",
       "      <td>6</td>\n",
       "      <td>16</td>\n",
       "      <td>7776</td>\n",
       "      <td>1600</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>g</th>\n",
       "      <td>7</td>\n",
       "      <td>17</td>\n",
       "      <td>16807</td>\n",
       "      <td>1700</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>h</th>\n",
       "      <td>8</td>\n",
       "      <td>18</td>\n",
       "      <td>32768</td>\n",
       "      <td>1800</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   feature1  feature2  feature3  feature4\n",
       "a         1        11         1      1100\n",
       "b         2        12        32      1200\n",
       "c         3        13       243      1300\n",
       "d         4        14      1024      1400\n",
       "e         5        15      3125      1500\n",
       "f         6        16      7776      1600\n",
       "g         7        17     16807      1700\n",
       "h         8        18     32768      1800"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import random\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "from multiprocessing.pool import ThreadPool\n",
    "\n",
    "def add_features(df):\n",
    "    # computing intensive processing\n",
    "    df['feature3'] = df['feature1'].apply(lambda x : x*x*x*x*x)\n",
    "    df['feature4'] = df['feature2'].apply(lambda x : x*100)\n",
    "    return df\n",
    "\n",
    "datadf = pd.DataFrame({'feature1': [1,2,3,4,5,6,7,8], \n",
    "                       'feature2' : [11,12,13,14,15,16,17,18]}, \n",
    "                      index=['a','b','c','d','e','f','g','h'])\n",
    "\n",
    "def parallelize_dataframe(df, func, n_cores=4):\n",
    "    df_split = np.array_split(df, n_cores)\n",
    "    pool = ThreadPool(n_cores)\n",
    "    df = pd.concat(pool.map(func, df_split))\n",
    "    pool.close()\n",
    "    pool.join()\n",
    "    return df\n",
    "\n",
    "complexdf = parallelize_dataframe(datadf, add_features)\n",
    "complexdf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-06T13:44:24.008156Z",
     "iopub.status.busy": "2021-10-06T13:44:24.007156Z",
     "iopub.status.idle": "2021-10-06T13:44:24.021158Z",
     "shell.execute_reply": "2021-10-06T13:44:24.020156Z",
     "shell.execute_reply.started": "2021-10-06T13:44:24.008156Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting fasterpandas.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile fasterpandas.py\n",
    "\n",
    "import random\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "from multiprocessing import  Pool\n",
    "\n",
    "def add_features(df):\n",
    "    # computing intensive processing\n",
    "    df['feature3'] = df['feature1'].apply(lambda x : x*x*x*x*x)\n",
    "    df['feature4'] = df['feature2'].apply(lambda x : x*100)\n",
    "    return df\n",
    "\n",
    "datadf = pd.DataFrame({'feature1': [1,2,3,4,5,6,7,8], 'feature2' : [11,12,13,14,15,16,17,18]}, index=['a','b','c','d','e','f','g','h'])\n",
    "\n",
    "def parallelize_dataframe(df, func, n_cores=4):\n",
    "    df_split = np.array_split(df, n_cores)\n",
    "    pool = Pool(n_cores)\n",
    "    df = pd.concat(pool.map(func, df_split))\n",
    "    pool.close()\n",
    "    pool.join()\n",
    "    return df\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    complexdf = parallelize_dataframe(datadf, add_features)\n",
    "    print(complexdf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-09-29T14:07:53.889751Z",
     "iopub.status.busy": "2021-09-29T14:07:53.889751Z",
     "iopub.status.idle": "2021-09-29T14:07:53.900741Z",
     "shell.execute_reply": "2021-09-29T14:07:53.899736Z",
     "shell.execute_reply.started": "2021-09-29T14:07:53.889751Z"
    }
   },
   "source": [
    "# [Reference] Communication among processes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-block alert-info\">\n",
    "    <b>Communication Examples Among Processes</b> <br>\n",
    "    - queue \n",
    "<pre>\n",
    "    from multiprocessing import Process, Queue <br>\n",
    "                                                <br>\n",
    "    def f(q): <br>\n",
    "        q.put([42, None, 'hello']) <br>\n",
    "                                                <br>\n",
    "    if __name__ == '__main__': <br>\n",
    "        q = Queue() <br>\n",
    "        p = Process(target=f, args=(q,)) <br>\n",
    "        p.start() <br>\n",
    "        print(q.get())    # prints \"[42, None, 'hello']\" <br>\n",
    "        p.join() <br>\n",
    "</pre>\n",
    "    <br>\n",
    "    - pipes <br>\n",
    "<pre>\n",
    "   from multiprocessing import Process, Pipe <br>\n",
    "                                               <br>\n",
    "    def f(conn): <br>\n",
    "        conn.send([42, None, 'hello']) <br>\n",
    "        conn.close() <br>\n",
    "        <br>\n",
    "    if __name__ == '__main__': <br>\n",
    "        parent_conn, child_conn = Pipe() <br>\n",
    "        p = Process(target=f, args=(child_conn,)) <br>\n",
    "        p.start() <br>\n",
    "        print(parent_conn.recv())   # prints \"[42, None, 'hello']\" <br>\n",
    "        p.join()  <br>\n",
    "</pre>\n",
    "    <br>\n",
    "    - Shared Memory <br>\n",
    "<pre>\n",
    "    from multiprocessing import Process, Value, Array <br>\n",
    "                    <br>\n",
    "    def f(n, a):<br>\n",
    "        n.value = 3.1415927<br>\n",
    "        for i in range(len(a)):<br>\n",
    "            a[i] = -a[i]<br>\n",
    "<br>\n",
    "    if __name__ == '__main__':<br>\n",
    "        num = Value('d', 0.0)<br>\n",
    "        arr = Array('i', range(10))<br>\n",
    "<br>\n",
    "        p = Process(target=f, args=(num, arr))<br>\n",
    "        p.start()<br>\n",
    "        p.join()<br>\n",
    "<br>\n",
    "        print(num.value)<br>\n",
    "        print(arr[:])<br>\n",
    "</pre>\n",
    "    <br>\n",
    "    - Server process <br>\n",
    "<pre>\n",
    "    from multiprocessing import Process, Manager<br>\n",
    "<br>\n",
    "    def f(d, l):<br>\n",
    "        d[1] = '1'<br>\n",
    "        d['2'] = 2<br>\n",
    "        d[0.25] = None<br>\n",
    "        l.reverse()<br>\n",
    "<br>\n",
    "    if __name__ == '__main__':<br>\n",
    "        with Manager() as manager:<br>\n",
    "            d = manager.dict()<br>\n",
    "            l = manager.list(range(10))<br>\n",
    "<br>\n",
    "            p = Process(target=f, args=(d, l))<br>\n",
    "            p.start()<br>\n",
    "            p.join()<br>\n",
    "<br>\n",
    "            print(d)<br>\n",
    "            print(l)<br>\n",
    "</pre>\n",
    "</div>"
   ]
  }
 ],
 "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.7.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}