Create ParallelCollection.ipynb

notebooks_2021/ParallelCollection.ipynb

@@ -0,0 +1,136 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 48,
+   "id": "03a12777",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from concurrent.futures import ThreadPoolExecutor as Pool\n",
+    "from itertools import tee\n",
+    "from time import sleep\n",
+    "from operator import add\n",
+    "from functools import reduce\n",
+    "\n",
+    "class ParallelCollection:\n",
+    "    \n",
+    "    def __init__(self, data, np):\n",
+    "        \n",
+    "        self.data = data\n",
+    "        self.np = np\n",
+    "        \n",
+    "    def map( self, func):\n",
+    "        with Pool(self.np) as pool:\n",
+    "            results = pool.map( func, self.data)\n",
+    "        return ParallelCollection(results, self.np)\n",
+    "            \n",
+    "    def filter( self, func):\n",
+    "        with Pool(self.np) as pool:\n",
+    "            futures = [pool.submit(func, x) for x in self.data]\n",
+    "            results = []\n",
+    "            for x,f in zip(self.data, futures):\n",
+    "                if f.result():\n",
+    "                    results.append(x)\n",
+    "        return ParallelCollection(results, self.np)\n",
+    "        \n",
+    "    def sum( self ):\n",
+    "        return sum(self.data)\n",
+    "    \n",
+    "        \n",
+    "        \n",
+    "        \n",
+    "        "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 50,
+   "id": "5c45bc2c",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "36\n",
+      "12\n",
+      "CPU times: user 4.06 ms, sys: 3.44 ms, total: 7.5 ms\n",
+      "Wall time: 4.01 s\n"
+     ]
+    }
+   ],
+   "source": [
+    "%%time\n",
+    "\n",
+    "rdd = ParallelCollection( data, 4)\n",
+    "\n",
+    "print(rdd.map(slow_add).sum())\n",
+    "print(rdd.filter(slow_odd).sum())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 51,
+   "id": "e0990a7e",
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "36\n",
+      "12\n",
+      "CPU times: user 2.58 ms, sys: 2.16 ms, total: 4.74 ms\n",
+      "Wall time: 16 s\n"
+     ]
+    }
+   ],
+   "source": [
+    "%%time\n",
+    "\n",
+    "def slow_add(x):\n",
+    "    sleep(1)\n",
+    "    return x + 1\n",
+    "\n",
+    "def slow_odd(x):\n",
+    "    sleep(1)\n",
+    "    return x % 2 == 0\n",
+    "\n",
+    "data = list(range(8))\n",
+    "\n",
+    "print(sum(map(slow_add, data)))\n",
+    "print(sum(filter(slow_odd, data)))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "61cd0621",
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "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": 5
+}