Rapids ETL Timing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "373f39a0",
   "metadata": {},
   "source": [
    "# Creating Data Subsets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "6b358d60",
   "metadata": {},
   "outputs": [],
   "source": [
    "import cudf\n",
    "import sys\n",
    "import pandas as pd\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b893db9c",
   "metadata": {},
   "source": [
    "## Reading in Datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "6563c253",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reading in the Carbon Dioxide Dataset\n",
    "co_df = cudf.read_csv(\"1980-2008-CO.csv\")\n",
    "co_df[\"RAW_VALUE\"] = co_df.RAW_VALUE.astype(float)\n",
    "co_df[\"ROUNDED_VALUE\"] = co_df.RAW_VALUE.astype(float)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "08ffa0c0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reading in the CO station information\n",
    "station_co_df = cudf.read_csv(\"bc_air_monitoring_stations.csv\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "25df0cb1",
   "metadata": {},
   "source": [
    "## Creating Tests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "8e35ff26",
   "metadata": {},
   "outputs": [],
   "source": [
    "def mean(co_subset):\n",
    "    # Computing the average raw value\n",
    "    return co_subset['RAW_VALUE'].mean()\n",
    "\n",
    "def sort(co_subset):\n",
    "    # Sorting based on the raw CO value \n",
    "    return co_subset.sort_values(['RAW_VALUE'], ascending=True)\n",
    "\n",
    "def merge(co_subset, station_co_df):\n",
    "    # Merging with the CO Stations to find which station made each measurement\n",
    "    return co_subset.merge(station_co_df, how='left', on='STATION_NAME')\n",
    "\n",
    "def filters(co_subset):\n",
    "    # Filtering to the \"Victoria Topaz\" station\n",
    "    filterd_pan_df = co_subset[co_subset['STATION_NAME'] == 'Victoria Topaz']\n",
    "    \n",
    "def all_tests(co_subset, station_co_df):\n",
    "    # Computing the average raw value\n",
    "    pan_mean = co_subset['RAW_VALUE'].mean()\n",
    "\n",
    "    # Sorting based on the raw CO value \n",
    "    pan_sort_df = co_subset.sort_values(['RAW_VALUE'], ascending=True)\n",
    "\n",
    "    # Merging with the CO Stations to find which station made each measurement\n",
    "    panmerge_df = co_subset.merge(station_co_df, how='left', on='STATION_NAME')\n",
    "\n",
    "    # Filtering to the \"Victoria Topaz\" station\n",
    "    filterd_pan_df = co_subset[co_subset['STATION_NAME'] == 'Victoria Topaz']\n",
    "    \n",
    "def tester(test, co_subset, station_co_df, NUM_EXECUTIONS_PER_TEST):    \n",
    "    if test == 'mean':\n",
    "        # Starting timer\n",
    "        t0 = time.time()   \n",
    "\n",
    "        for i in range(0,NUM_EXECUTIONS_PER_TEST):\n",
    "            mean(co_subset)\n",
    "            \n",
    "        # Stopping clock\n",
    "        t1 = time.time()\n",
    "\n",
    "        # Recording Results\n",
    "        total_time = t1-t0\n",
    "        avg_time = total_time/NUM_EXECUTIONS_PER_TEST\n",
    "\n",
    "        return total_time, avg_time\n",
    "    \n",
    "    elif test == 'sort':\n",
    "        # Starting timer\n",
    "        t0 = time.time()   \n",
    "\n",
    "        for i in range(0,NUM_EXECUTIONS_PER_TEST):\n",
    "            sort(co_subset)\n",
    "            \n",
    "        # Stopping clock\n",
    "        t1 = time.time()\n",
    "\n",
    "        # Recording Results\n",
    "        total_time = t1-t0\n",
    "        avg_time = total_time/NUM_EXECUTIONS_PER_TEST\n",
    "\n",
    "        return total_time, avg_time\n",
    "      \n",
    "    elif test == 'merge':\n",
    "        # Starting timer\n",
    "        t0 = time.time()   \n",
    "\n",
    "        for i in range(0,NUM_EXECUTIONS_PER_TEST):\n",
    "            merge(co_subset, station_co_df)\n",
    "            \n",
    "        # Stopping clock\n",
    "        t1 = time.time()\n",
    "\n",
    "        # Recording Results\n",
    "        total_time = t1-t0\n",
    "        avg_time = total_time/NUM_EXECUTIONS_PER_TEST\n",
    "\n",
    "        return total_time, avg_time\n",
    "    elif test == 'filter':\n",
    "        # Starting timer\n",
    "        t0 = time.time()   \n",
    "\n",
    "        for i in range(0,NUM_EXECUTIONS_PER_TEST):\n",
    "            filters(co_subset)\n",
    "            \n",
    "        # Stopping clock\n",
    "        t1 = time.time()\n",
    "\n",
    "        # Recording Results\n",
    "        total_time = t1-t0\n",
    "        avg_time = total_time/NUM_EXECUTIONS_PER_TEST\n",
    "\n",
    "        return total_time, avg_time\n",
    "    else:\n",
    "        # Starting timer\n",
    "        t0 = time.time()   \n",
    "\n",
    "        for i in range(0,NUM_EXECUTIONS_PER_TEST):\n",
    "            all_tests(co_subset, station_co_df)\n",
    "            \n",
    "        # Stopping clock\n",
    "        t1 = time.time()\n",
    "\n",
    "        # Recording Results\n",
    "        total_time = t1-t0\n",
    "        avg_time = total_time/NUM_EXECUTIONS_PER_TEST\n",
    "\n",
    "        return total_time, avg_time\n",
    "        \n",
    "          \n",
    "    # Stopping clock\n",
    "    t1 = time.time()\n",
    "    \n",
    "    # Recording Results\n",
    "    total_time = t1-t0\n",
    "    avg_time = total_time/NUM_EXECUTIONS_PER_TEST\n",
    "    \n",
    "    return total_time, avg_time\n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "2f0db4d7",
   "metadata": {},
   "outputs": [],
   "source": [
    "NUM_START_ROWS = 2500\n",
    "\n",
    "NUM_EXECUTIONS_PER_TEST = 3\n",
    "\n",
    "NUM_DSIZE_DOUBLINGS = 12"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "92911255",
   "metadata": {},
   "source": [
    "## Executing Tests"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b9a79158",
   "metadata": {},
   "source": [
    "### RAPIDS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "fceaa348",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Test: 0\n"
     ]
    }
   ],
   "source": [
    "numRows = NUM_START_ROWS\n",
    "\n",
    "rapids = []\n",
    "for i in range(0, NUM_DSIZE_DOUBLINGS):\n",
    "    print('Test:', i)\n",
    "    \n",
    "    numRows = numRows * 2\n",
    "    \n",
    "    if numRows > len(co_df):\n",
    "        co_subset = co_subset.append(co_subset)\n",
    "    else:\n",
    "        co_subset = co_df.iloc[0:numRows]\n",
    "        \n",
    "    # ******************************************************************************\n",
    "    # MEAN TEST\n",
    "    \n",
    "    test = {'Test':'Mean'}\n",
    "    total_time, avg_time = tester('mean', co_subset, station_co_df, NUM_EXECUTIONS_PER_TEST)\n",
    "    test['Total'] = total_time\n",
    "    test['Average'] = avg_time\n",
    "    \n",
    "    rapids.append(test)\n",
    "    \n",
    "    # ******************************************************************************\n",
    "    \n",
    "    # SORT TEST\n",
    "    \n",
    "    test = {'Test':'Sort'}\n",
    "    \n",
    "    total_time, avg_time = tester('sort', co_subset, station_co_df, NUM_EXECUTIONS_PER_TEST)\n",
    "    \n",
    "    test['Total'] = total_time\n",
    "    test['Average'] = avg_time\n",
    "    \n",
    "    rapids.append(test)\n",
    "    \n",
    "    # ******************************************************************************\n",
    "    \n",
    "    # MERGE TEST\n",
    "    \n",
    "    test = {'Test':'Merge'}\n",
    "    \n",
    "    total_time, avg_time = tester('merge', co_subset, station_co_df, NUM_EXECUTIONS_PER_TEST)\n",
    "    \n",
    "    test['Total'] = total_time\n",
    "    test['Average'] = avg_time\n",
    "    \n",
    "    rapids.append(test)\n",
    "    \n",
    "    # ******************************************************************************\n",
    "    \n",
    "    # FILTER TEST\n",
    "    \n",
    "    test = {'Test':'Filter'}\n",
    "    \n",
    "    total_time, avg_time = tester('filter', co_subset, station_co_df, NUM_EXECUTIONS_PER_TEST)\n",
    "    \n",
    "    test['Total'] = total_time\n",
    "    test['Average'] = avg_time\n",
    "        \n",
    "    rapids.append(test)\n",
    "    \n",
    "    # ******************************************************************************\n",
    "    \n",
    "    # ALL TEST\n",
    "    \n",
    "    test = {'Test':'All'}\n",
    "    \n",
    "    total_time, avg_time = tester('all', co_subset, station_co_df, NUM_EXECUTIONS_PER_TEST)\n",
    "    \n",
    "    test['Total'] = total_time\n",
    "    test['Average'] = avg_time\n",
    "    \n",
    "    rapids.append(test)\n",
    "    \n",
    "    # ******************************************************************************\n",
    "    \n",
    "results_df_rapids = pd.DataFrame(rapids)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "bdabe127",
   "metadata": {},
   "outputs": [],
   "source": [
    "results_df_rapids.to_csv('results_df_rapids_etl.csv')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}