Mongo as data prep.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Mongo as data prep\n",
    "\n",
    "This is some discovery work to start to identify how we can execute data preprocessing jobs in MongoDB."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Pretty printing has been turned OFF\n"
     ]
    }
   ],
   "source": [
    "%pprint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Necessary dependecies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: docker in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (4.2.0)\n",
      "Requirement already satisfied: pymongo in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (3.10.1)\n",
      "Requirement already satisfied: pandas in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (1.0.3)\n",
      "Requirement already satisfied: requests!=2.18.0,>=2.14.2 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from docker) (2.23.0)\n",
      "Requirement already satisfied: six>=1.4.0 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from docker) (1.14.0)\n",
      "Requirement already satisfied: websocket-client>=0.32.0 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from docker) (0.57.0)\n",
      "Requirement already satisfied: numpy>=1.13.3 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from pandas) (1.18.4)\n",
      "Requirement already satisfied: pytz>=2017.2 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from pandas) (2020.1)\n",
      "Requirement already satisfied: python-dateutil>=2.6.1 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from pandas) (2.8.1)\n",
      "Requirement already satisfied: idna<3,>=2.5 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from requests!=2.18.0,>=2.14.2->docker) (2.9)\n",
      "Requirement already satisfied: chardet<4,>=3.0.2 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from requests!=2.18.0,>=2.14.2->docker) (3.0.4)\n",
      "Requirement already satisfied: urllib3!=1.25.0,!=1.25.1,<1.26,>=1.21.1 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from requests!=2.18.0,>=2.14.2->docker) (1.24.3)\n",
      "Requirement already satisfied: certifi>=2017.4.17 in /home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/lib/python3.6/site-packages (from requests!=2.18.0,>=2.14.2->docker) (2020.4.5.1)\n",
      "\u001b[33mWARNING: You are using pip version 20.0.2; however, version 20.1 is available.\n",
      "You should consider upgrading via the '/home/rachelle2/.pyenv/versions/3.6.9/envs/laputa-3.6/bin/python3.6 -m pip install --upgrade pip' command.\u001b[0m\n",
      "Note: you may need to restart the kernel to use updated packages.\n"
     ]
    }
   ],
   "source": [
    "%pip install docker pymongo pandas"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run a mongo container"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import docker\n",
    "\n",
    "docker_client = docker.from_env()\n",
    "\n",
    "mongo_container = docker_client.containers.run(\n",
    "    'mongo:latest',\n",
    "    detach=True,\n",
    "    name='mongo-data-prep',\n",
    "    remove=True,\n",
    "    ports={'27017/tcp': 27018},\n",
    "    mem_limit='512M',\n",
    "    volumes={\n",
    "        os.path.join(os.getcwd(), 'db'): {'bind': '/data/db', 'mode': 'rw'},\n",
    "        os.path.join(os.getcwd(), 'mongo.conf'): {'bind': '/etc/mongo.conf', 'mode': 'ro'}\n",
    "    }\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To get its state:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "b'2020-05-02T12:46:04.050+0000 I  CONTROL  [main] Automatically disabling TLS 1.0, to force-enable TLS 1.0 specify --sslDisabledProtocols \\'none\\'\\n2020-05-02T12:46:04.057+0000 W  ASIO     [main] No TransportLayer configured during NetworkInterface startup\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] MongoDB starting : pid=1 port=27017 dbpath=/data/db 64-bit host=ca620174dbc5\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] db version v4.2.5\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] git version: 2261279b51ea13df08ae708ff278f0679c59dc32\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] OpenSSL version: OpenSSL 1.1.1  11 Sep 2018\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] allocator: tcmalloc\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] modules: none\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] build environment:\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten]     distmod: ubuntu1804\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten]     distarch: x86_64\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten]     target_arch: x86_64\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] 256 MB of memory available to the process out of 15829 MB total system memory\\n2020-05-02T12:46:04.057+0000 I  CONTROL  [initandlisten] options: { net: { bindIp: \"*\" } }\\n2020-05-02T12:46:04.058+0000 W  STORAGE  [initandlisten] Detected unclean shutdown - /data/db/mongod.lock is not empty.\\n2020-05-02T12:46:04.058+0000 I  STORAGE  [initandlisten] Detected data files in /data/db created by the \\'wiredTiger\\' storage engine, so setting the active storage engine to \\'wiredTiger\\'.\\n2020-05-02T12:46:04.058+0000 W  STORAGE  [initandlisten] Recovering data from the last clean checkpoint.\\n2020-05-02T12:46:04.058+0000 I  STORAGE  [initandlisten] \\n2020-05-02T12:46:04.058+0000 I  STORAGE  [initandlisten] ** WARNING: Using the XFS filesystem is strongly recommended with the WiredTiger storage engine\\n2020-05-02T12:46:04.058+0000 I  STORAGE  [initandlisten] **          See http://dochub.mongodb.org/core/prodnotes-filesystem\\n2020-05-02T12:46:04.058+0000 I  STORAGE  [initandlisten] \\n2020-05-02T12:46:04.058+0000 I  STORAGE  [initandlisten] ** WARNING: The configured WiredTiger cache size is more than 80% of available RAM.\\n2020-05-02T12:46:04.058+0000 I  STORAGE  [initandlisten] **          See http://dochub.mongodb.org/core/faq-memory-diagnostics-wt\\n2020-05-02T12:46:04.058+0000 I  STORAGE  [initandlisten] wiredtiger_open config: create,cache_size=256M,cache_overflow=(file_max=0M),session_max=33000,eviction=(threads_min=4,threads_max=4),config_base=false,statistics=(fast),log=(enabled=true,archive=true,path=journal,compressor=snappy),file_manager=(close_idle_time=100000,close_scan_interval=10,close_handle_minimum=250),statistics_log=(wait=0),verbose=[recovery_progress,checkpoint_progress],\\n2020-05-02T12:46:04.650+0000 I  STORAGE  [initandlisten] WiredTiger message [1588423564:650049][1:0x7f46cd25ab00], txn-recover: Recovering log 9 through 10\\n2020-05-02T12:46:05.081+0000 I  STORAGE  [initandlisten] WiredTiger message [1588423565:81749][1:0x7f46cd25ab00], txn-recover: Recovering log 10 through 10\\n2020-05-02T12:46:05.145+0000 I  STORAGE  [initandlisten] WiredTiger message [1588423565:145407][1:0x7f46cd25ab00], txn-recover: Main recovery loop: starting at 9/59015040 to 10/256\\n2020-05-02T12:46:05.146+0000 I  STORAGE  [initandlisten] WiredTiger message [1588423565:146211][1:0x7f46cd25ab00], txn-recover: Recovering log 9 through 10\\n2020-05-02T12:46:06.805+0000 I  STORAGE  [initandlisten] WiredTiger message [1588423566:805494][1:0x7f46cd25ab00], file:sizeStorer.wt, txn-recover: Recovering log 10 through 10\\n2020-05-02T12:46:06.926+0000 I  STORAGE  [initandlisten] WiredTiger message [1588423566:925770][1:0x7f46cd25ab00], file:sizeStorer.wt, txn-recover: Set global recovery timestamp: (0, 0)\\n2020-05-02T12:46:07.070+0000 I  RECOVERY [initandlisten] WiredTiger recoveryTimestamp. Ts: Timestamp(0, 0)\\n2020-05-02T12:46:07.099+0000 I  STORAGE  [initandlisten] Timestamp monitor starting\\n2020-05-02T12:46:07.109+0000 I  CONTROL  [initandlisten] \\n2020-05-02T12:46:07.109+0000 I  CONTROL  [initandlisten] ** WARNING: Access control is not enabled for the database.\\n2020-05-02T12:46:07.109+0000 I  CONTROL  [initandlisten] **          Read and write access to data and configuration is unrestricted.\\n2020-05-02T12:46:07.110+0000 I  CONTROL  [initandlisten] \\n2020-05-02T12:46:07.130+0000 I  SHARDING [initandlisten] Marking collection local.system.replset as collection version: <unsharded>\\n2020-05-02T12:46:07.139+0000 I  STORAGE  [initandlisten] Flow Control is enabled on this deployment.\\n2020-05-02T12:46:07.143+0000 I  SHARDING [initandlisten] Marking collection admin.system.roles as collection version: <unsharded>\\n2020-05-02T12:46:07.145+0000 I  SHARDING [initandlisten] Marking collection admin.system.version as collection version: <unsharded>\\n2020-05-02T12:46:07.154+0000 I  SHARDING [initandlisten] Marking collection local.startup_log as collection version: <unsharded>\\n2020-05-02T12:46:07.156+0000 I  FTDC     [initandlisten] Initializing full-time diagnostic data capture with directory \\'/data/db/diagnostic.data\\'\\n2020-05-02T12:46:07.164+0000 I  SHARDING [LogicalSessionCacheReap] Marking collection config.system.sessions as collection version: <unsharded>\\n2020-05-02T12:46:07.164+0000 I  NETWORK  [listener] Listening on /tmp/mongodb-27017.sock\\n2020-05-02T12:46:07.164+0000 I  NETWORK  [listener] Listening on 0.0.0.0\\n2020-05-02T12:46:07.164+0000 I  NETWORK  [listener] waiting for connections on port 27017\\n2020-05-02T12:46:07.166+0000 I  SHARDING [LogicalSessionCacheReap] Marking collection config.transactions as collection version: <unsharded>\\n2020-05-02T12:46:08.056+0000 I  FTDC     [ftdc] Unclean full-time diagnostic data capture shutdown detected, found interim file, some metrics may have been lost. OK\\n2020-05-02T12:46:08.134+0000 I  SHARDING [ftdc] Marking collection local.oplog.rs as collection version: <unsharded>\\n2020-05-02T12:46:11.283+0000 I  NETWORK  [listener] connection accepted from 172.17.0.1:58364 #1 (1 connection now open)\\n2020-05-02T12:46:11.286+0000 I  NETWORK  [conn1] received client metadata from 172.17.0.1:58364 conn1: { driver: { name: \"PyMongo\", version: \"3.10.1\" }, os: { type: \"Linux\", name: \"Linux\", architecture: \"x86_64\", version: \"5.4.0-28-generic\" }, platform: \"CPython 3.8.2.final.0\" }\\n'"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mongo_container.logs()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To stop it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "mongo_container.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connect using PyMongo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pymongo import MongoClient\n",
    "mongo_client = MongoClient('127.0.0.1', 27018)\n",
    "db = mongo_client['data-prep']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'version': '4.2.5',\n",
       " 'gitVersion': '2261279b51ea13df08ae708ff278f0679c59dc32',\n",
       " 'modules': [],\n",
       " 'allocator': 'tcmalloc',\n",
       " 'javascriptEngine': 'mozjs',\n",
       " 'sysInfo': 'deprecated',\n",
       " 'versionArray': [4, 2, 5, 0],\n",
       " 'openssl': {'running': 'OpenSSL 1.1.1  11 Sep 2018',\n",
       "  'compiled': 'OpenSSL 1.1.1  11 Sep 2018'},\n",
       " 'buildEnvironment': {'distmod': 'ubuntu1804',\n",
       "  'distarch': 'x86_64',\n",
       "  'cc': '/opt/mongodbtoolchain/v3/bin/gcc: gcc (GCC) 8.2.0',\n",
       "  'ccflags': '-fno-omit-frame-pointer -fno-strict-aliasing -ggdb -pthread -Wall -Wsign-compare -Wno-unknown-pragmas -Winvalid-pch -Werror -O2 -Wno-unused-local-typedefs -Wno-unused-function -Wno-deprecated-declarations -Wno-unused-const-variable -Wno-unused-but-set-variable -Wno-missing-braces -fstack-protector-strong -fno-builtin-memcmp',\n",
       "  'cxx': '/opt/mongodbtoolchain/v3/bin/g++: g++ (GCC) 8.2.0',\n",
       "  'cxxflags': '-Woverloaded-virtual -Wno-maybe-uninitialized -fsized-deallocation -std=c++17',\n",
       "  'linkflags': '-pthread -Wl,-z,now -rdynamic -Wl,--fatal-warnings -fstack-protector-strong -fuse-ld=gold -Wl,--build-id -Wl,--hash-style=gnu -Wl,-z,noexecstack -Wl,--warn-execstack -Wl,-z,relro',\n",
       "  'target_arch': 'x86_64',\n",
       "  'target_os': 'linux'},\n",
       " 'bits': 64,\n",
       " 'debug': False,\n",
       " 'maxBsonObjectSize': 16777216,\n",
       " 'storageEngines': ['biggie', 'devnull', 'ephemeralForTest', 'wiredTiger'],\n",
       " 'ok': 1.0}"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "mongo_client.server_info()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A note on why we need why the docker memory limit is not enough.\n",
    "We can see below that the databse thinks it can use all the host memory (see note in https://docs.mongodb.com/v3.2/reference/configuration-options/#storage.wiredTiger.engineConfig.cacheSizeGB).\n",
    "We need to limit manually how much memory it can take using the mongo.conf file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'currentTime': datetime.datetime(2020, 5, 3, 8, 34, 45, 646000),\n",
       " 'hostname': '3b67778f62f8',\n",
       " 'cpuAddrSize': 64,\n",
       " 'memSizeMB': 15829,\n",
       " 'memLimitMB': 512,\n",
       " 'numCores': 8,\n",
       " 'cpuArch': 'x86_64',\n",
       " 'numaEnabled': False}"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "db.command('hostInfo')['system']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating some random data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There is a total of 26474401 rows\n",
      "There is 26474 chunks\n"
     ]
    }
   ],
   "source": [
    "from datetime import date\n",
    "import pandas as pd\n",
    "from numpy.random import default_rng\n",
    "\n",
    "rng = default_rng()\n",
    "\n",
    "date_range = pd.date_range(\n",
    "    end=date.today(),\n",
    "    start=date.fromisoformat('1970-01-01'),\n",
    "    freq='min',\n",
    "    name='moment',\n",
    ")\n",
    "print(f'There is a total of {date_range.size} rows')\n",
    " \n",
    "# Generate only a chunk at a time to avoid having thbe whole dataframe in memory\n",
    "def generate_sample_chunk_df(chunk_size=100000):\n",
    "    rows_count = date_range.size\n",
    "    chunks_count = int(rows_count / chunk_size)\n",
    "    print(f'There is {chunks_count} chunks')\n",
    "\n",
    "    for i in range(chunks_count + 1):\n",
    "        slice_begin = i * chunk_size\n",
    "        if i <= chunks_count:\n",
    "            slice_end = slice_begin + chunk_size - 1\n",
    "        else:  # For the last chunk\n",
    "            slice_end = slice_begin + 1 + rows_count % chunk_size\n",
    "\n",
    "        chunk_date_range = date_range[slice_begin : slice_end]\n",
    "\n",
    "        df = pd.DataFrame(index=chunk_date_range, columns=[])\n",
    "\n",
    "        df['quantity'] = rng.integers(0, 1000, chunk_date_range.size)\n",
    "        df['color'] = rng.choice(['black', 'white', 'purple', 'cyan'], chunk_date_range.size)\n",
    "\n",
    "        # Creates the moment column from the index\n",
    "        df.reset_index(inplace=True)\n",
    "\n",
    "        # Creates an index column\n",
    "        df.reset_index(inplace=True)\n",
    "        df['index'] += slice_begin\n",
    "\n",
    "        yield df\n",
    "\n",
    "# The first chunk\n",
    "g = generate_sample_chunk_df()\n",
    "next(g)\n",
    "\n",
    "# Wanna see the last one? use reversed(range(...)) in the generator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inserting in mongo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There is 264 chunks\n",
      "Inserting chunk 1...\n",
      "Inserting chunk 2...\n",
      "Inserting chunk 3...\n",
      "Inserting chunk 4...\n",
      "Inserting chunk 5...\n",
      "Inserting chunk 6...\n",
      "Inserting chunk 7...\n",
      "Inserting chunk 8...\n",
      "Inserting chunk 9...\n",
      "Inserting chunk 10...\n",
      "Inserting chunk 11...\n",
      "Inserting chunk 12...\n",
      "Inserting chunk 13...\n",
      "Inserting chunk 14...\n",
      "Inserting chunk 15...\n",
      "Inserting chunk 16...\n",
      "Inserting chunk 17...\n",
      "Inserting chunk 18...\n",
      "Inserting chunk 19...\n",
      "Inserting chunk 20...\n",
      "Inserting chunk 21...\n",
      "Inserting chunk 22...\n",
      "Inserting chunk 23...\n",
      "Inserting chunk 24...\n",
      "Inserting chunk 25...\n",
      "Inserting chunk 26...\n",
      "Inserting chunk 27...\n",
      "Inserting chunk 28...\n",
      "Inserting chunk 29...\n",
      "Inserting chunk 30...\n",
      "Inserting chunk 31...\n",
      "Inserting chunk 32...\n",
      "Inserting chunk 33...\n",
      "Inserting chunk 34...\n",
      "Inserting chunk 35...\n",
      "Inserting chunk 36...\n",
      "Inserting chunk 37...\n",
      "Inserting chunk 38...\n",
      "Inserting chunk 39...\n",
      "Inserting chunk 40...\n",
      "Inserting chunk 41...\n",
      "Inserting chunk 42...\n",
      "Inserting chunk 43...\n",
      "Inserting chunk 44...\n",
      "Inserting chunk 45...\n",
      "Inserting chunk 46...\n",
      "Inserting chunk 47...\n",
      "Inserting chunk 48...\n",
      "Inserting chunk 49...\n",
      "Inserting chunk 50...\n",
      "Inserting chunk 51...\n",
      "Inserting chunk 52...\n",
      "Inserting chunk 53...\n",
      "Inserting chunk 54...\n",
      "Inserting chunk 55...\n",
      "Inserting chunk 56...\n",
      "Inserting chunk 57...\n",
      "Inserting chunk 58...\n",
      "Inserting chunk 59...\n",
      "Inserting chunk 60...\n",
      "Inserting chunk 61...\n",
      "Inserting chunk 62...\n",
      "Inserting chunk 63...\n",
      "Inserting chunk 64...\n",
      "Inserting chunk 65...\n",
      "Inserting chunk 66...\n",
      "Inserting chunk 67...\n",
      "Inserting chunk 68...\n",
      "Inserting chunk 69...\n",
      "Inserting chunk 70...\n",
      "Inserting chunk 71...\n",
      "Inserting chunk 72...\n",
      "Inserting chunk 73...\n",
      "Inserting chunk 74...\n",
      "Inserting chunk 75...\n",
      "Inserting chunk 76...\n",
      "Inserting chunk 77...\n",
      "Inserting chunk 78...\n",
      "Inserting chunk 79...\n",
      "Inserting chunk 80...\n",
      "Inserting chunk 81...\n",
      "Inserting chunk 82...\n",
      "Inserting chunk 83...\n",
      "Inserting chunk 84...\n",
      "Inserting chunk 85...\n",
      "Inserting chunk 86...\n",
      "Inserting chunk 87...\n",
      "Inserting chunk 88...\n",
      "Inserting chunk 89...\n",
      "Inserting chunk 90...\n",
      "Inserting chunk 91...\n",
      "Inserting chunk 92...\n",
      "Inserting chunk 93...\n",
      "Inserting chunk 94...\n",
      "Inserting chunk 95...\n",
      "Inserting chunk 96...\n",
      "Inserting chunk 97...\n",
      "Inserting chunk 98...\n",
      "Inserting chunk 99...\n",
      "Inserting chunk 100...\n",
      "Inserting chunk 101...\n",
      "Inserting chunk 102...\n",
      "Inserting chunk 103...\n",
      "Inserting chunk 104...\n",
      "Inserting chunk 105...\n",
      "Inserting chunk 106...\n",
      "Inserting chunk 107...\n",
      "Inserting chunk 108...\n",
      "Inserting chunk 109...\n",
      "Inserting chunk 110...\n",
      "Inserting chunk 111...\n",
      "Inserting chunk 112...\n",
      "Inserting chunk 113...\n",
      "Inserting chunk 114...\n",
      "Inserting chunk 115...\n",
      "Inserting chunk 116...\n",
      "Inserting chunk 117...\n",
      "Inserting chunk 118...\n",
      "Inserting chunk 119...\n",
      "Inserting chunk 120...\n",
      "Inserting chunk 121...\n",
      "Inserting chunk 122...\n",
      "Inserting chunk 123...\n",
      "Inserting chunk 124...\n",
      "Inserting chunk 125...\n",
      "Inserting chunk 126...\n",
      "Inserting chunk 127...\n",
      "Inserting chunk 128...\n",
      "Inserting chunk 129...\n",
      "Inserting chunk 130...\n",
      "Inserting chunk 131...\n",
      "Inserting chunk 132...\n",
      "Inserting chunk 133...\n",
      "Inserting chunk 134...\n",
      "Inserting chunk 135...\n",
      "Inserting chunk 136...\n",
      "Inserting chunk 137...\n",
      "Inserting chunk 138...\n",
      "Inserting chunk 139...\n",
      "Inserting chunk 140...\n",
      "Inserting chunk 141...\n",
      "Inserting chunk 142...\n",
      "Inserting chunk 143...\n",
      "Inserting chunk 144...\n",
      "Inserting chunk 145...\n",
      "Inserting chunk 146...\n",
      "Inserting chunk 147...\n",
      "Inserting chunk 148...\n",
      "Inserting chunk 149...\n",
      "Inserting chunk 150...\n",
      "Inserting chunk 151...\n",
      "Inserting chunk 152...\n",
      "Inserting chunk 153...\n",
      "Inserting chunk 154...\n",
      "Inserting chunk 155...\n",
      "Inserting chunk 156...\n",
      "Inserting chunk 157...\n",
      "Inserting chunk 158...\n",
      "Inserting chunk 159...\n",
      "Inserting chunk 160...\n",
      "Inserting chunk 161...\n",
      "Inserting chunk 162...\n",
      "Inserting chunk 163...\n",
      "Inserting chunk 164...\n",
      "Inserting chunk 165...\n",
      "Inserting chunk 166...\n",
      "Inserting chunk 167...\n",
      "Inserting chunk 168...\n",
      "Inserting chunk 169...\n",
      "Inserting chunk 170...\n",
      "Inserting chunk 171...\n",
      "Inserting chunk 172...\n",
      "Inserting chunk 173...\n",
      "Inserting chunk 174...\n",
      "Inserting chunk 175...\n",
      "Inserting chunk 176...\n",
      "Inserting chunk 177...\n",
      "Inserting chunk 178...\n",
      "Inserting chunk 179...\n",
      "Inserting chunk 180...\n",
      "Inserting chunk 181...\n",
      "Inserting chunk 182...\n",
      "Inserting chunk 183...\n",
      "Inserting chunk 184...\n",
      "Inserting chunk 185...\n",
      "Inserting chunk 186...\n",
      "Inserting chunk 187...\n",
      "Inserting chunk 188...\n",
      "Inserting chunk 189...\n",
      "Inserting chunk 190...\n",
      "Inserting chunk 191...\n",
      "Inserting chunk 192...\n",
      "Inserting chunk 193...\n",
      "Inserting chunk 194...\n",
      "Inserting chunk 195...\n",
      "Inserting chunk 196...\n",
      "Inserting chunk 197...\n",
      "Inserting chunk 198...\n",
      "Inserting chunk 199...\n",
      "Inserting chunk 200...\n",
      "Inserting chunk 201...\n",
      "Inserting chunk 202...\n",
      "Inserting chunk 203...\n",
      "Inserting chunk 204...\n",
      "Inserting chunk 205...\n",
      "Inserting chunk 206...\n",
      "Inserting chunk 207...\n",
      "Inserting chunk 208...\n",
      "Inserting chunk 209...\n",
      "Inserting chunk 210...\n",
      "Inserting chunk 211...\n",
      "Inserting chunk 212...\n",
      "Inserting chunk 213...\n",
      "Inserting chunk 214...\n",
      "Inserting chunk 215...\n",
      "Inserting chunk 216...\n",
      "Inserting chunk 217...\n",
      "Inserting chunk 218...\n",
      "Inserting chunk 219...\n",
      "Inserting chunk 220...\n",
      "Inserting chunk 221...\n",
      "Inserting chunk 222...\n",
      "Inserting chunk 223...\n",
      "Inserting chunk 224...\n",
      "Inserting chunk 225...\n",
      "Inserting chunk 226...\n",
      "Inserting chunk 227...\n",
      "Inserting chunk 228...\n",
      "Inserting chunk 229...\n",
      "Inserting chunk 230...\n",
      "Inserting chunk 231...\n",
      "Inserting chunk 232...\n",
      "Inserting chunk 233...\n",
      "Inserting chunk 234...\n",
      "Inserting chunk 235...\n",
      "Inserting chunk 236...\n",
      "Inserting chunk 237...\n",
      "Inserting chunk 238...\n",
      "Inserting chunk 239...\n",
      "Inserting chunk 240...\n",
      "Inserting chunk 241...\n",
      "Inserting chunk 242...\n",
      "Inserting chunk 243...\n",
      "Inserting chunk 244...\n",
      "Inserting chunk 245...\n",
      "Inserting chunk 246...\n",
      "Inserting chunk 247...\n",
      "Inserting chunk 248...\n",
      "Inserting chunk 249...\n",
      "Inserting chunk 250...\n",
      "Inserting chunk 251...\n",
      "Inserting chunk 252...\n",
      "Inserting chunk 253...\n",
      "Inserting chunk 254...\n",
      "Inserting chunk 255...\n",
      "Inserting chunk 256...\n",
      "Inserting chunk 257...\n",
      "Inserting chunk 258...\n",
      "Inserting chunk 259...\n",
      "Inserting chunk 260...\n",
      "Inserting chunk 261...\n",
      "Inserting chunk 262...\n",
      "Inserting chunk 263...\n",
      "Inserting chunk 264...\n",
      "Inserting chunk 265...\n",
      "CPU times: user 9min 35s, sys: 6.51 s, total: 9min 42s\n",
      "Wall time: 11min 33s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "db.input_data.drop()\n",
    "\n",
    "i = 0\n",
    "for chunk in generate_sample_chunk_df():\n",
    "    i += 1\n",
    "    print(f'Inserting chunk {i}...')\n",
    "    db.input_data.insert_many(chunk.to_dict(orient='record'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "During this long operation, we can monitor that the mongo container never reach its limit:\n",
    "`docker stats`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "26472697"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "db.input_data.count_documents({})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "ename": "TypeError",
     "evalue": "'Collection' object is not callable. If you meant to call the 'data_size' method on a 'Collection' object it is failing because no such method exists.",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-4-e96f60572423>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mdb\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0minput_data\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdata_size\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~/.pyenv/versions/3.8.2/envs/mongo-data-prep/lib/python3.8/site-packages/pymongo/collection.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args, **kwargs)\u001b[0m\n\u001b[1;32m   3338\u001b[0m                             \u001b[0;34m\"exists.\"\u001b[0m \u001b[0;34m%\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3339\u001b[0m                             self.__name)\n\u001b[0;32m-> 3340\u001b[0;31m         raise TypeError(\"'Collection' object is not callable. If you meant to \"\n\u001b[0m\u001b[1;32m   3341\u001b[0m                         \u001b[0;34m\"call the '%s' method on a 'Collection' object it is \"\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3342\u001b[0m                         \u001b[0;34m\"failing because no such method exists.\"\u001b[0m \u001b[0;34m%\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mTypeError\u001b[0m: 'Collection' object is not callable. If you meant to call the 'data_size' method on a 'Collection' object it is failing because no such method exists."
     ]
    }
   ],
   "source": [
    "db.input_data.data_size()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Running an aggregation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A simple aggregation on a subset (100 documents) is fast:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "match_all = {\n",
    "    '$match': {}\n",
    "}\n",
    "\n",
    "limit_100 = {\n",
    "    '$limit': 100\n",
    "}\n",
    "\n",
    "add_year_field = {\n",
    "    '$addFields': {\n",
    "        'year': {'$year': '$moment'},\n",
    "    }\n",
    "}\n",
    "\n",
    "aggregate_by_year = {\n",
    "    '$group': {\n",
    "        '_id': {\n",
    "            'year': '$year'\n",
    "        },\n",
    "        'quantity': {'$avg': '$quantity'},\n",
    "        'color': {'$first': '$color'},\n",
    "        'start_index': {'$first': '$index'},\n",
    "        'end_index': {'$last': '$index'},\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 6.07 ms, sys: 0 ns, total: 6.07 ms\n",
      "Wall time: 6.36 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[{'_id': {'year': 1970},\n",
       "  'quantity': 507.17,\n",
       "  'color': 'cyan',\n",
       "  'start_index': 0,\n",
       "  'end_index': 99}]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "list(db.input_data.aggregate([\n",
    "    match_all,\n",
    "    limit_100,\n",
    "    add_year_field,\n",
    "    aggregate_by_year,\n",
    "]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But without the limit, it takes a while but does works without crashing anything:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 9.58 ms, sys: 6.6 ms, total: 16.2 ms\n",
      "Wall time: 51.6 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "results = db.input_data.aggregate([\n",
    "    match_all,\n",
    "    add_year_field,\n",
    "    aggregate_by_year,\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "scrolled": true
   },
   "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>_id</th>\n",
       "      <th>quantity</th>\n",
       "      <th>color</th>\n",
       "      <th>start_index</th>\n",
       "      <th>end_index</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>{'year': 1985}</td>\n",
       "      <td>498.973548</td>\n",
       "      <td>cyan</td>\n",
       "      <td>7889760</td>\n",
       "      <td>8415359</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>{'year': 1977}</td>\n",
       "      <td>498.434233</td>\n",
       "      <td>white</td>\n",
       "      <td>3682080</td>\n",
       "      <td>4207679</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>{'year': 2001}</td>\n",
       "      <td>499.745214</td>\n",
       "      <td>cyan</td>\n",
       "      <td>16305120</td>\n",
       "      <td>16830719</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>{'year': 2009}</td>\n",
       "      <td>499.932941</td>\n",
       "      <td>white</td>\n",
       "      <td>20512800</td>\n",
       "      <td>21038399</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>{'year': 1998}</td>\n",
       "      <td>500.043629</td>\n",
       "      <td>black</td>\n",
       "      <td>14726880</td>\n",
       "      <td>15252479</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>{'year': 2015}</td>\n",
       "      <td>500.070008</td>\n",
       "      <td>cyan</td>\n",
       "      <td>23667840</td>\n",
       "      <td>24193439</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>{'year': 2020}</td>\n",
       "      <td>498.643856</td>\n",
       "      <td>purple</td>\n",
       "      <td>26297280</td>\n",
       "      <td>26472960</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>{'year': 1988}</td>\n",
       "      <td>498.907750</td>\n",
       "      <td>cyan</td>\n",
       "      <td>9466560</td>\n",
       "      <td>9993599</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>{'year': 2013}</td>\n",
       "      <td>499.772079</td>\n",
       "      <td>cyan</td>\n",
       "      <td>22616640</td>\n",
       "      <td>23142239</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>{'year': 2016}</td>\n",
       "      <td>498.678129</td>\n",
       "      <td>cyan</td>\n",
       "      <td>24193440</td>\n",
       "      <td>24720479</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10</th>\n",
       "      <td>{'year': 2018}</td>\n",
       "      <td>498.623556</td>\n",
       "      <td>black</td>\n",
       "      <td>25246080</td>\n",
       "      <td>25771679</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>11</th>\n",
       "      <td>{'year': 2000}</td>\n",
       "      <td>499.776117</td>\n",
       "      <td>cyan</td>\n",
       "      <td>15778080</td>\n",
       "      <td>16305119</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>12</th>\n",
       "      <td>{'year': 1976}</td>\n",
       "      <td>499.533788</td>\n",
       "      <td>black</td>\n",
       "      <td>3155040</td>\n",
       "      <td>3682079</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>13</th>\n",
       "      <td>{'year': 1973}</td>\n",
       "      <td>500.543646</td>\n",
       "      <td>cyan</td>\n",
       "      <td>1578240</td>\n",
       "      <td>2103839</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>14</th>\n",
       "      <td>{'year': 1972}</td>\n",
       "      <td>499.141046</td>\n",
       "      <td>purple</td>\n",
       "      <td>1051200</td>\n",
       "      <td>1578239</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>15</th>\n",
       "      <td>{'year': 1989}</td>\n",
       "      <td>499.346281</td>\n",
       "      <td>purple</td>\n",
       "      <td>9993600</td>\n",
       "      <td>10519199</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>16</th>\n",
       "      <td>{'year': 2006}</td>\n",
       "      <td>499.377353</td>\n",
       "      <td>white</td>\n",
       "      <td>18934560</td>\n",
       "      <td>19460159</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>17</th>\n",
       "      <td>{'year': 1995}</td>\n",
       "      <td>499.431144</td>\n",
       "      <td>black</td>\n",
       "      <td>13148640</td>\n",
       "      <td>13674239</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>18</th>\n",
       "      <td>{'year': 1974}</td>\n",
       "      <td>499.409087</td>\n",
       "      <td>purple</td>\n",
       "      <td>2103840</td>\n",
       "      <td>2629439</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>19</th>\n",
       "      <td>{'year': 1997}</td>\n",
       "      <td>499.998862</td>\n",
       "      <td>black</td>\n",
       "      <td>14201280</td>\n",
       "      <td>14726879</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>20</th>\n",
       "      <td>{'year': 1990}</td>\n",
       "      <td>499.034372</td>\n",
       "      <td>white</td>\n",
       "      <td>10519200</td>\n",
       "      <td>11044799</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>21</th>\n",
       "      <td>{'year': 2008}</td>\n",
       "      <td>499.854656</td>\n",
       "      <td>white</td>\n",
       "      <td>19985760</td>\n",
       "      <td>20512799</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>22</th>\n",
       "      <td>{'year': 2012}</td>\n",
       "      <td>499.139361</td>\n",
       "      <td>purple</td>\n",
       "      <td>22089600</td>\n",
       "      <td>22616639</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>23</th>\n",
       "      <td>{'year': 1978}</td>\n",
       "      <td>499.993917</td>\n",
       "      <td>white</td>\n",
       "      <td>4207680</td>\n",
       "      <td>4733279</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>24</th>\n",
       "      <td>{'year': 1986}</td>\n",
       "      <td>499.575401</td>\n",
       "      <td>white</td>\n",
       "      <td>8415360</td>\n",
       "      <td>8940959</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>25</th>\n",
       "      <td>{'year': 1979}</td>\n",
       "      <td>499.251041</td>\n",
       "      <td>white</td>\n",
       "      <td>4733280</td>\n",
       "      <td>5258879</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>26</th>\n",
       "      <td>{'year': 1987}</td>\n",
       "      <td>500.354248</td>\n",
       "      <td>black</td>\n",
       "      <td>8940960</td>\n",
       "      <td>9466559</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>27</th>\n",
       "      <td>{'year': 2003}</td>\n",
       "      <td>499.356864</td>\n",
       "      <td>black</td>\n",
       "      <td>17356320</td>\n",
       "      <td>17881919</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>28</th>\n",
       "      <td>{'year': 1975}</td>\n",
       "      <td>499.450098</td>\n",
       "      <td>purple</td>\n",
       "      <td>2629440</td>\n",
       "      <td>3155039</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>29</th>\n",
       "      <td>{'year': 2005}</td>\n",
       "      <td>499.600386</td>\n",
       "      <td>cyan</td>\n",
       "      <td>18408960</td>\n",
       "      <td>18934559</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>30</th>\n",
       "      <td>{'year': 2002}</td>\n",
       "      <td>499.513148</td>\n",
       "      <td>cyan</td>\n",
       "      <td>16830720</td>\n",
       "      <td>17356319</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>31</th>\n",
       "      <td>{'year': 1994}</td>\n",
       "      <td>499.477824</td>\n",
       "      <td>cyan</td>\n",
       "      <td>12623040</td>\n",
       "      <td>13148639</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>32</th>\n",
       "      <td>{'year': 1981}</td>\n",
       "      <td>500.114881</td>\n",
       "      <td>white</td>\n",
       "      <td>5785920</td>\n",
       "      <td>6311519</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>33</th>\n",
       "      <td>{'year': 2007}</td>\n",
       "      <td>499.624441</td>\n",
       "      <td>white</td>\n",
       "      <td>19460160</td>\n",
       "      <td>19985759</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34</th>\n",
       "      <td>{'year': 2017}</td>\n",
       "      <td>499.946225</td>\n",
       "      <td>cyan</td>\n",
       "      <td>24720480</td>\n",
       "      <td>25246079</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>35</th>\n",
       "      <td>{'year': 1984}</td>\n",
       "      <td>500.129767</td>\n",
       "      <td>black</td>\n",
       "      <td>7362720</td>\n",
       "      <td>7889759</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>36</th>\n",
       "      <td>{'year': 1991}</td>\n",
       "      <td>499.189576</td>\n",
       "      <td>white</td>\n",
       "      <td>11044800</td>\n",
       "      <td>11570399</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>37</th>\n",
       "      <td>{'year': 2011}</td>\n",
       "      <td>499.554568</td>\n",
       "      <td>cyan</td>\n",
       "      <td>21564000</td>\n",
       "      <td>22089599</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>38</th>\n",
       "      <td>{'year': 2019}</td>\n",
       "      <td>499.575921</td>\n",
       "      <td>cyan</td>\n",
       "      <td>25771680</td>\n",
       "      <td>26297279</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>39</th>\n",
       "      <td>{'year': 1970}</td>\n",
       "      <td>499.291251</td>\n",
       "      <td>cyan</td>\n",
       "      <td>0</td>\n",
       "      <td>525599</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>40</th>\n",
       "      <td>{'year': 1971}</td>\n",
       "      <td>499.710452</td>\n",
       "      <td>cyan</td>\n",
       "      <td>525600</td>\n",
       "      <td>1051199</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>41</th>\n",
       "      <td>{'year': 1983}</td>\n",
       "      <td>499.869877</td>\n",
       "      <td>black</td>\n",
       "      <td>6837120</td>\n",
       "      <td>7362719</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>42</th>\n",
       "      <td>{'year': 1993}</td>\n",
       "      <td>500.214987</td>\n",
       "      <td>cyan</td>\n",
       "      <td>12097440</td>\n",
       "      <td>12623039</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>43</th>\n",
       "      <td>{'year': 1999}</td>\n",
       "      <td>499.623001</td>\n",
       "      <td>white</td>\n",
       "      <td>15252480</td>\n",
       "      <td>15778079</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>44</th>\n",
       "      <td>{'year': 2004}</td>\n",
       "      <td>499.565861</td>\n",
       "      <td>black</td>\n",
       "      <td>17881920</td>\n",
       "      <td>18408959</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>45</th>\n",
       "      <td>{'year': 1982}</td>\n",
       "      <td>499.715810</td>\n",
       "      <td>black</td>\n",
       "      <td>6311520</td>\n",
       "      <td>6837119</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>46</th>\n",
       "      <td>{'year': 2010}</td>\n",
       "      <td>499.184220</td>\n",
       "      <td>purple</td>\n",
       "      <td>21038400</td>\n",
       "      <td>21563999</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>47</th>\n",
       "      <td>{'year': 2014}</td>\n",
       "      <td>499.631707</td>\n",
       "      <td>black</td>\n",
       "      <td>23142240</td>\n",
       "      <td>23667839</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>48</th>\n",
       "      <td>{'year': 1992}</td>\n",
       "      <td>499.459823</td>\n",
       "      <td>black</td>\n",
       "      <td>11570400</td>\n",
       "      <td>12097439</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>49</th>\n",
       "      <td>{'year': 1996}</td>\n",
       "      <td>499.471182</td>\n",
       "      <td>cyan</td>\n",
       "      <td>13674240</td>\n",
       "      <td>14201279</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>50</th>\n",
       "      <td>{'year': 1980}</td>\n",
       "      <td>499.906082</td>\n",
       "      <td>cyan</td>\n",
       "      <td>5258880</td>\n",
       "      <td>5785919</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "               _id    quantity   color  start_index  end_index\n",
       "0   {'year': 1985}  498.973548    cyan      7889760    8415359\n",
       "1   {'year': 1977}  498.434233   white      3682080    4207679\n",
       "2   {'year': 2001}  499.745214    cyan     16305120   16830719\n",
       "3   {'year': 2009}  499.932941   white     20512800   21038399\n",
       "4   {'year': 1998}  500.043629   black     14726880   15252479\n",
       "5   {'year': 2015}  500.070008    cyan     23667840   24193439\n",
       "6   {'year': 2020}  498.643856  purple     26297280   26472960\n",
       "7   {'year': 1988}  498.907750    cyan      9466560    9993599\n",
       "8   {'year': 2013}  499.772079    cyan     22616640   23142239\n",
       "9   {'year': 2016}  498.678129    cyan     24193440   24720479\n",
       "10  {'year': 2018}  498.623556   black     25246080   25771679\n",
       "11  {'year': 2000}  499.776117    cyan     15778080   16305119\n",
       "12  {'year': 1976}  499.533788   black      3155040    3682079\n",
       "13  {'year': 1973}  500.543646    cyan      1578240    2103839\n",
       "14  {'year': 1972}  499.141046  purple      1051200    1578239\n",
       "15  {'year': 1989}  499.346281  purple      9993600   10519199\n",
       "16  {'year': 2006}  499.377353   white     18934560   19460159\n",
       "17  {'year': 1995}  499.431144   black     13148640   13674239\n",
       "18  {'year': 1974}  499.409087  purple      2103840    2629439\n",
       "19  {'year': 1997}  499.998862   black     14201280   14726879\n",
       "20  {'year': 1990}  499.034372   white     10519200   11044799\n",
       "21  {'year': 2008}  499.854656   white     19985760   20512799\n",
       "22  {'year': 2012}  499.139361  purple     22089600   22616639\n",
       "23  {'year': 1978}  499.993917   white      4207680    4733279\n",
       "24  {'year': 1986}  499.575401   white      8415360    8940959\n",
       "25  {'year': 1979}  499.251041   white      4733280    5258879\n",
       "26  {'year': 1987}  500.354248   black      8940960    9466559\n",
       "27  {'year': 2003}  499.356864   black     17356320   17881919\n",
       "28  {'year': 1975}  499.450098  purple      2629440    3155039\n",
       "29  {'year': 2005}  499.600386    cyan     18408960   18934559\n",
       "30  {'year': 2002}  499.513148    cyan     16830720   17356319\n",
       "31  {'year': 1994}  499.477824    cyan     12623040   13148639\n",
       "32  {'year': 1981}  500.114881   white      5785920    6311519\n",
       "33  {'year': 2007}  499.624441   white     19460160   19985759\n",
       "34  {'year': 2017}  499.946225    cyan     24720480   25246079\n",
       "35  {'year': 1984}  500.129767   black      7362720    7889759\n",
       "36  {'year': 1991}  499.189576   white     11044800   11570399\n",
       "37  {'year': 2011}  499.554568    cyan     21564000   22089599\n",
       "38  {'year': 2019}  499.575921    cyan     25771680   26297279\n",
       "39  {'year': 1970}  499.291251    cyan            0     525599\n",
       "40  {'year': 1971}  499.710452    cyan       525600    1051199\n",
       "41  {'year': 1983}  499.869877   black      6837120    7362719\n",
       "42  {'year': 1993}  500.214987    cyan     12097440   12623039\n",
       "43  {'year': 1999}  499.623001   white     15252480   15778079\n",
       "44  {'year': 2004}  499.565861   black     17881920   18408959\n",
       "45  {'year': 1982}  499.715810   black      6311520    6837119\n",
       "46  {'year': 2010}  499.184220  purple     21038400   21563999\n",
       "47  {'year': 2014}  499.631707   black     23142240   23667839\n",
       "48  {'year': 1992}  499.459823   black     11570400   12097439\n",
       "49  {'year': 1996}  499.471182    cyan     13674240   14201279\n",
       "50  {'year': 1980}  499.906082    cyan      5258880    5785919"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.DataFrame(list(results))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note the weird order of results! MongoDB docs states that documents are not sorted after a `$group` stage."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Conclusion: the memory limit really protects the mongo process. It takes all the CPU time of the container, so it'll be required to test if only one core is used in full or if the whole mongo server will become unresponsive."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's try an aggregation with many more documents:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "add_hourly_field = {\n",
    "    '$addFields': {\n",
    "        'year': {'$year': '$moment'},\n",
    "        'month': {'$month': '$moment'},\n",
    "        'day': {'$dayOfMonth': '$moment'},\n",
    "        'hour': {'$hour': '$moment'},\n",
    "    }\n",
    "}\n",
    "\n",
    "aggregate_by_hour = {\n",
    "    '$group': {\n",
    "        '_id': {\n",
    "            'year': '$year',\n",
    "            'month': '$month',\n",
    "            'day': '$day',\n",
    "            'hour': '$hour',\n",
    "        },\n",
    "        'quantity': {'$avg': '$quantity'},\n",
    "        'color': {'$first': '$color'},\n",
    "        'start_index': {'$first': '$index'},\n",
    "        'end_index': {'$last': '$index'},\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "ename": "ServerSelectionTimeoutError",
     "evalue": "127.0.0.1:27018: [Errno 111] Connection refused",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mServerSelectionTimeoutError\u001b[0m               Traceback (most recent call last)",
      "\u001b[0;32m<timed exec>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n",
      "\u001b[0;32m~/.pyenv/versions/3.8.2/envs/mongo-data-prep/lib/python3.8/site-packages/pymongo/collection.py\u001b[0m in \u001b[0;36maggregate\u001b[0;34m(self, pipeline, session, **kwargs)\u001b[0m\n\u001b[1;32m   2372\u001b[0m             \u001b[0mhttps\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m//\u001b[0m\u001b[0mdocs\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mmongodb\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcom\u001b[0m\u001b[0;34m/\u001b[0m\u001b[0mmanual\u001b[0m\u001b[0;34m/\u001b[0m\u001b[0mreference\u001b[0m\u001b[0;34m/\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m/\u001b[0m\u001b[0maggregate\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   2373\u001b[0m         \"\"\"\n\u001b[0;32m-> 2374\u001b[0;31m         \u001b[0;32mwith\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__database\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mclient\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_tmp_session\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msession\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mclose\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mFalse\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0ms\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   2375\u001b[0m             return self._aggregate(_CollectionAggregationCommand,\n\u001b[1;32m   2376\u001b[0m                                    \u001b[0mpipeline\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.pyenv/versions/3.8.2/lib/python3.8/contextlib.py\u001b[0m in \u001b[0;36m__enter__\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    111\u001b[0m         \u001b[0;32mdel\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0margs\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mkwds\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[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    112\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--> 113\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mnext\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgen\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    114\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mStopIteration\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    115\u001b[0m             \u001b[0;32mraise\u001b[0m \u001b[0mRuntimeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"generator didn't yield\"\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.pyenv/versions/3.8.2/envs/mongo-data-prep/lib/python3.8/site-packages/pymongo/mongo_client.py\u001b[0m in \u001b[0;36m_tmp_session\u001b[0;34m(self, session, close)\u001b[0m\n\u001b[1;32m   1821\u001b[0m             \u001b[0;32mreturn\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1822\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1823\u001b[0;31m         \u001b[0ms\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_ensure_session\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msession\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   1824\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0ms\u001b[0m \u001b[0;32mand\u001b[0m \u001b[0mclose\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1825\u001b[0m             \u001b[0;32mwith\u001b[0m \u001b[0ms\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.pyenv/versions/3.8.2/envs/mongo-data-prep/lib/python3.8/site-packages/pymongo/mongo_client.py\u001b[0m in \u001b[0;36m_ensure_session\u001b[0;34m(self, session)\u001b[0m\n\u001b[1;32m   1808\u001b[0m             \u001b[0;31m# Don't make implicit sessions causally consistent. Applications\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1809\u001b[0m             \u001b[0;31m# should always opt-in.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1810\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__start_session\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcausal_consistency\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mFalse\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   1811\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0;34m(\u001b[0m\u001b[0mConfigurationError\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mInvalidOperation\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   1812\u001b[0m             \u001b[0;31m# Sessions not supported, or multiple users authenticated.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.pyenv/versions/3.8.2/envs/mongo-data-prep/lib/python3.8/site-packages/pymongo/mongo_client.py\u001b[0m in \u001b[0;36m__start_session\u001b[0;34m(self, implicit, **kwargs)\u001b[0m\n\u001b[1;32m   1761\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1762\u001b[0m         \u001b[0;31m# Raises ConfigurationError if sessions are not supported.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1763\u001b[0;31m         \u001b[0mserver_session\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_get_server_session\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   1764\u001b[0m         \u001b[0mopts\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mclient_session\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mSessionOptions\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m**\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1765\u001b[0m         return client_session.ClientSession(\n",
      "\u001b[0;32m~/.pyenv/versions/3.8.2/envs/mongo-data-prep/lib/python3.8/site-packages/pymongo/mongo_client.py\u001b[0m in \u001b[0;36m_get_server_session\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m   1794\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_get_server_session\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   1795\u001b[0m         \u001b[0;34m\"\"\"Internal: start or resume a _ServerSession.\"\"\"\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1796\u001b[0;31m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_topology\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget_server_session\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   1797\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1798\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_return_server_session\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mserver_session\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mlock\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~/.pyenv/versions/3.8.2/envs/mongo-data-prep/lib/python3.8/site-packages/pymongo/topology.py\u001b[0m in \u001b[0;36mget_server_session\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    480\u001b[0m                 \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_description\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtopology_type\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0mTOPOLOGY_TYPE\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mSingle\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    481\u001b[0m                     \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_description\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mhas_known_servers\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 482\u001b[0;31m                         self._select_servers_loop(\n\u001b[0m\u001b[1;32m    483\u001b[0m                             \u001b[0many_server_selector\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    484\u001b[0m                             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_settings\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mserver_selection_timeout\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.pyenv/versions/3.8.2/envs/mongo-data-prep/lib/python3.8/site-packages/pymongo/topology.py\u001b[0m in \u001b[0;36m_select_servers_loop\u001b[0;34m(self, selector, timeout, address)\u001b[0m\n\u001b[1;32m    206\u001b[0m             \u001b[0;31m# No suitable servers.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    207\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0mtimeout\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;36m0\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0mnow\u001b[0m \u001b[0;34m>\u001b[0m \u001b[0mend_time\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 208\u001b[0;31m                 raise ServerSelectionTimeoutError(\n\u001b[0m\u001b[1;32m    209\u001b[0m                     self._error_message(selector))\n\u001b[1;32m    210\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mServerSelectionTimeoutError\u001b[0m: 127.0.0.1:27018: [Errno 111] Connection refused"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "results = db.input_data.aggregate([\n",
    "    match_all,\n",
    "    add_hourly_field,\n",
    "    aggregate_by_hour,\n",
    "])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Fails with `OperationFailure: Exceeded memory limit for $group, but didn't allow external sort. Pass allowDiskUse:true to opt in.`\n",
    "\n",
    "Two things next:\n",
    "- first, let's enable the `allowDiskUse`\n",
    "- second, let's write the output in another collection\n",
    "\n",
    "We do not want data to get out of the db to our pymongo client, for it will deplete our host's memory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "write_in_output_collection = {\n",
    "    '$out': 'output_data'\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 34.7 ms, sys: 10.9 ms, total: 45.6 ms\n",
      "Wall time: 1min 33s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "results = db.input_data.aggregate([\n",
    "    match_all,\n",
    "    add_hourly_field,\n",
    "    aggregate_by_hour,\n",
    "    write_in_output_collection\n",
    "], allowDiskUse=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'_id': {'year': 1970, 'month': 1, 'day': 1, 'hour': 0},\n",
       " 'quantity': 501.65,\n",
       " 'color': 'cyan',\n",
       " 'start_index': 0,\n",
       " 'end_index': 59}"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "r = db.output_data.find({})\n",
    "r[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note: this seems to indicate we could monitor the process by having a second command querying for the output results: https://stackoverflow.com/questions/22725814/view-progress-of-long-running-mongodb-aggregation-job"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}