evaluations.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# IOMax Evaluations\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask\n",
    "import dask.dataframe as dd\n",
    "import itertools as it\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import pyarrow\n",
    "from time import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Query Definitions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define indices & combinations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "indices = ['file_name', 'proc_name', 'io_cat', 'acc_pat']\n",
    "combinations = list(it.combinations(indices, r=2))[:5]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Simple Queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "def base_simple_pandas_queries(log_dir):\n",
    "    print(f\",base_simple_pandas_queries\")\n",
    "    df = pd.read_parquet(log_dir, engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix in ['file_name', 'proc_name']:\n",
    "        for col in ['size', 'duration', 'index', 'bandwidth', 'io_cat']: \n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            res = df.groupby([ix]).agg({col: 'sum'})\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"simple,Q{query_index},{time_elapsed},{res.memory_usage(deep=True).sum()}\")\n",
    "\n",
    "def base_simple_dask_queries(log_dir):\n",
    "    print(f\",base_simple_dask_queries\")\n",
    "    query_index = 0\n",
    "    ddf = dd.read_parquet(f\"{log_dir}/*.parquet\", engine='pyarrow')\n",
    "    for ix in ['file_name', 'proc_name']:\n",
    "        for col in ['size', 'duration', 'index', 'bandwidth', 'io_cat']:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            res = ddf.groupby([ix]).agg({col: 'sum'}).compute()\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{res.memory_usage(deep=True).sum()}\")\n",
    "\n",
    "def iomax_simple_pandas_queries(log_dir):\n",
    "    print(f\",iomax_simple_pandas_queries\")\n",
    "    query_index = 0\n",
    "    df = pd.read_parquet(log_dir, engine='pyarrow')\n",
    "    cols = ['size', 'duration', 'index', 'bandwidth', 'io_cat']\n",
    "    agg_dict = {col: sum for col in cols}\n",
    "    for ix in ['file_name', 'proc_name']:\n",
    "        for col in cols:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            m = 0\n",
    "            if query_index == 1:\n",
    "                x = df.groupby(['file_name', 'proc_name']).agg(agg_dict)\n",
    "                m = x.memory_usage(deep=True).sum()\n",
    "            x.groupby([ix]).agg({col: 'sum'})\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{m}\")\n",
    "\n",
    "def iomax_simple_dask_queries(log_dir):\n",
    "    print(f\",iomax_simple_dask_queries\")\n",
    "    query_index = 0\n",
    "    ddf = dd.read_parquet(f\"{log_dir}/*.parquet\", engine='pyarrow')\n",
    "    cols = ['size', 'duration', 'index', 'bandwidth', 'io_cat']\n",
    "    agg_dict = {col: sum for col in cols}\n",
    "    for ix in ['file_name', 'proc_name']:\n",
    "        for col in cols:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            m = 0\n",
    "            if query_index == 1:\n",
    "                x = ddf.groupby(['file_name', 'proc_name']).agg(agg_dict).compute()\n",
    "                m = x.memory_usage(deep=True).sum()\n",
    "            x.groupby([ix]).agg({col: 'sum'})\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{m}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Medium Queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "def base_medium_pandas_queries(log_dir):\n",
    "    print(f\",base_medium_pandas_queries\")\n",
    "    df = pd.read_parquet(log_dir, engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix, iy in combinations:\n",
    "        for col in ['size', 'duration']:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            res = df.groupby([ix, iy]).agg({col: 'sum'}).groupby([ix]).sum()\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"medium,Q{query_index},{time_elapsed},{res.memory_usage(deep=True).sum()}\")\n",
    "            \n",
    "\n",
    "def base_medium_dask_queries(log_dir):\n",
    "    print(f\",base_medium_dask_queries\")\n",
    "    ddf = dd.read_parquet(f\"{log_dir}/*.parquet\", engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix, iy in combinations:\n",
    "        for col in ['size', 'duration']:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            res = ddf.groupby([ix, iy]).agg({col: 'sum'}).groupby([ix]).sum().compute()\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{res.memory_usage(deep=True).sum()}\")\n",
    "\n",
    "def iomax_medium_pandas_queries(log_dir):\n",
    "    print(f\",iomax_medium_pandas_queries\")\n",
    "    df = pd.read_parquet(log_dir, engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix, iy in combinations:\n",
    "        for col in ['size', 'duration']:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            m = 0\n",
    "            if query_index == 1:\n",
    "                x = df.groupby(indices).agg({'size': sum, 'duration': sum})\n",
    "                m = x.memory_usage(deep=True).sum()\n",
    "            x.groupby([ix, iy]).agg({col: 'sum'}).groupby([ix]).sum()   \n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{m}\")\n",
    "\n",
    "def iomax_medium_dask_queries(log_dir):\n",
    "    print(f\",iomax_medium_dask_queries\")\n",
    "    ddf = dd.read_parquet(f\"{log_dir}/*.parquet\", engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix, iy in combinations:\n",
    "        for col in ['size', 'duration']:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            m = 0\n",
    "            if query_index == 1:\n",
    "                x = ddf.groupby(indices).agg({'size': sum, 'duration': sum}).compute()\n",
    "                m = x.memory_usage(deep=True).sum()\n",
    "            x.groupby([ix, iy]).agg({col: 'sum'}).groupby([ix]).sum()\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{m}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Hard Queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "def base_hard_pandas_queries(log_dir):\n",
    "    print(f\",base_hard_pandas_queries\")\n",
    "    df = pd.read_parquet(log_dir, engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix, iy in combinations:\n",
    "        for col in ['size', 'duration']:\n",
    "            query_index += 1\n",
    "            t1 = time()\n",
    "            res = df.groupby([ix]).agg({iy: list, col: sum}).explode(iy).reset_index().groupby([iy]).agg({ix: list, col: sum})\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"hard,Q{query_index},{time_elapsed},{res.memory_usage(deep=True).sum()}\")\n",
    "\n",
    "def base_hard_dask_queries(log_dir):\n",
    "    print(f\",base_hard_dask_queries\")\n",
    "    ddf = dd.read_parquet(f\"{log_dir}/*.parquet\", engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix, iy in combinations:\n",
    "        for col in ['size', 'duration']:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            res = ddf.groupby([ix]).agg({iy: list, col: sum}).explode(iy).reset_index().groupby([iy]).agg({ix: list, col: sum}).compute()\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{res.memory_usage(deep=True).sum()}\")\n",
    "\n",
    "def iomax_hard_pandas_queries(log_dir):\n",
    "    print(f\",iomax_hard_pandas_queries\")\n",
    "    df = pd.read_parquet(log_dir, engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix, iy in combinations:\n",
    "        for col in ['size', 'duration']:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            m = 0\n",
    "            if query_index == 1:\n",
    "                x = df.groupby(indices).agg({'size': sum, 'duration': sum}).reset_index()\n",
    "                m = x.memory_usage(deep=True).sum()\n",
    "            x.groupby([ix]) \\\n",
    "                .agg({iy: list, col: sum}) \\\n",
    "                .reset_index() \\\n",
    "                .explode(iy) \\\n",
    "                .groupby([iy]) \\\n",
    "                .agg({ix: list, col: sum})\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{m}\")\n",
    "\n",
    "def iomax_hard_dask_queries(log_dir):\n",
    "    print(f\",iomax_hard_dask_queries\")\n",
    "    ddf = dd.read_parquet(f\"{log_dir}/*.parquet\", engine='pyarrow')\n",
    "    query_index = 0\n",
    "    for ix, iy in combinations:\n",
    "        for col in ['size', 'duration']:\n",
    "            query_index += 1 \n",
    "            t1 = time()\n",
    "            m = 0\n",
    "            if query_index == 1:\n",
    "                x = ddf.groupby(indices).agg({'size': sum, 'duration': sum}).reset_index().compute()\n",
    "                m = x.memory_usage(deep=True).sum()\n",
    "            x.groupby([ix]) \\\n",
    "                .agg({iy: list, col: sum}) \\\n",
    "                .reset_index() \\\n",
    "                .explode(iy) \\\n",
    "                .groupby([iy]) \\\n",
    "                .agg({ix: list, col: sum})\n",
    "            time_elapsed = time() - t1\n",
    "            print(f\"{time_elapsed},{m}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Datasets"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Download Datasets"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "datasets = [\n",
    "    'dataset-125m',\n",
    "    'dataset-25m',\n",
    "    'dataset-5m',\n",
    "    'dataset-1m',\n",
    "    'dataset-200k',\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create `datasets` folder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "!mkdir -p datasets"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Download datasets from Zenodo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2023-09-29 19:24:53--  https://zenodo.org/record/8393987/files/dataset-125m.tar.gz?download=1\n",
      "Resolving zenodo.org (zenodo.org)... 188.185.124.72\n",
      "Connecting to zenodo.org (zenodo.org)|188.185.124.72|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 1574979255 (1.5G) [application/octet-stream]\n",
      "Saving to: ‘datasets/dataset-125m.tar.gz’\n",
      "\n",
      " 1% [                                       ] 27,827,194  3.18MB/s  eta 8m 14s ^C\n",
      "--2023-09-29 19:25:05--  https://zenodo.org/record/8393987/files/dataset-25m.tar.gz?download=1\n",
      "Resolving zenodo.org (zenodo.org)... 188.185.124.72\n",
      "Connecting to zenodo.org (zenodo.org)|188.185.124.72|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 314991520 (300M) [application/octet-stream]\n",
      "Saving to: ‘datasets/dataset-25m.tar.gz’\n",
      "\n",
      " 5% [=>                                     ] 17,030,140  3.20MB/s  eta 98s    ^C\n",
      "--2023-09-29 19:25:12--  https://zenodo.org/record/8393987/files/dataset-5m.tar.gz?download=1\n",
      "Resolving zenodo.org (zenodo.org)... 188.185.124.72\n",
      "Connecting to zenodo.org (zenodo.org)|188.185.124.72|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 62945079 (60M) [application/octet-stream]\n",
      "Saving to: ‘datasets/dataset-5m.tar.gz’\n",
      "\n",
      "100%[======================================>] 62,945,079  3.20MB/s   in 20s    \n",
      "\n",
      "2023-09-29 19:25:34 (2.96 MB/s) - ‘datasets/dataset-5m.tar.gz’ saved [62945079/62945079]\n",
      "\n",
      "--2023-09-29 19:25:34--  https://zenodo.org/record/8393987/files/dataset-1m.tar.gz?download=1\n",
      "Resolving zenodo.org (zenodo.org)... 188.185.124.72\n",
      "Connecting to zenodo.org (zenodo.org)|188.185.124.72|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 13336035 (13M) [application/octet-stream]\n",
      "Saving to: ‘datasets/dataset-1m.tar.gz’\n",
      "\n",
      "100%[======================================>] 13,336,035  1.44MB/s   in 10s    \n",
      "\n",
      "2023-09-29 19:25:46 (1.23 MB/s) - ‘datasets/dataset-1m.tar.gz’ saved [13336035/13336035]\n",
      "\n",
      "--2023-09-29 19:25:46--  https://zenodo.org/record/8393987/files/dataset-200k.tar.gz?download=1\n",
      "Resolving zenodo.org (zenodo.org)... 188.185.124.72\n",
      "Connecting to zenodo.org (zenodo.org)|188.185.124.72|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 3037937 (2.9M) [application/octet-stream]\n",
      "Saving to: ‘datasets/dataset-200k.tar.gz’\n",
      "\n",
      "100%[======================================>] 3,037,937    599KB/s   in 8.0s   \n",
      "\n",
      "2023-09-29 19:25:57 (369 KB/s) - ‘datasets/dataset-200k.tar.gz’ saved [3037937/3037937]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "for ds in datasets:\n",
    "    !wget https://zenodo.org/record/8393987/files/{ds}.tar.gz?download=1 -O datasets/{ds}.tar.gz"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Extract datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for ds in datasets:\n",
    "    !mkdir -p datasets/{ds}\n",
    "    !tar -xzvf datasets/{ds}.tar.gz -C datasets/{ds}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Evaluations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Simple Queries"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 200k Records"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ",base_simple_pandas_queries\n",
      "simple,Q1,0.05765700340270996,83196\n",
      "simple,Q2,0.049272775650024414,80676\n",
      "simple,Q3,0.04894280433654785,83196\n",
      "simple,Q4,0.04857945442199707,80676\n",
      "simple,Q5,0.048729658126831055,80676\n",
      "simple,Q6,0.03307533264160156,4419\n",
      "simple,Q7,0.032759904861450195,4247\n",
      "simple,Q8,0.03292679786682129,4419\n",
      "simple,Q9,0.032509565353393555,4247\n",
      "simple,Q10,0.032901763916015625,4247\n"
     ]
    }
   ],
   "source": [
    "base_simple_pandas_queries('datasets/dataset-200k')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ",iomax_simple_pandas_queries\n",
      "0.10136675834655762,103210\n",
      "0.0011987686157226562,0\n",
      "0.0014908313751220703,0\n",
      "0.0012526512145996094,0\n",
      "0.0014369487762451172,0\n",
      "0.0013620853424072266,0\n",
      "0.0012354850769042969,0\n",
      "0.0013713836669921875,0\n",
      "0.0012328624725341797,0\n",
      "0.0013530254364013672,0\n"
     ]
    }
   ],
   "source": [
    "iomax_simple_pandas_queries('datasets/dataset-200k')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ",base_simple_dask_queries\n",
      "0.20629167556762695,83196\n",
      "0.1940934658050537,80676\n",
      "0.19583749771118164,83196\n",
      "0.1996769905090332,80676\n",
      "0.2173151969909668,80676\n",
      "0.1799612045288086,4419\n",
      "0.18725967407226562,4247\n",
      "0.1804027557373047,4419\n",
      "0.18625211715698242,4247\n",
      "0.1864008903503418,4247\n"
     ]
    }
   ],
   "source": [
    "base_simple_dask_queries('datasets/dataset-200k')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ",iomax_simple_dask_queries\n",
      "0.24787020683288574,120834\n",
      "0.0023605823516845703,0\n",
      "0.0017986297607421875,0\n",
      "0.0015556812286376953,0\n",
      "0.0016965866088867188,0\n",
      "0.001600027084350586,0\n",
      "0.00145721435546875,0\n",
      "0.0016016960144042969,0\n",
      "0.0014636516571044922,0\n",
      "0.0015964508056640625,0\n"
     ]
    }
   ],
   "source": [
    "iomax_simple_dask_queries('datasets/dataset-200k')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "iopp",
   "language": "python",
   "name": "iopp"
  },
  "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": 2
}