02_preprocess.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#default_exp data.preprocess"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Preprocessing\n",
    "\n",
    "> This module reads data generated in create-sample-data step from parquet files, and saves two datasets partitioned by customer_id and email in order to enable next steps to efficiently perform aggregations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "import dask.dataframe as dd\n",
    "\n",
    "from hopeit.app.context import EventContext\n",
    "from hopeit.app.api import event_api\n",
    "from hopeit.app.logger import app_logger\n",
    "\n",
    "from fraud_poc.jobs import get_client, MakeSampleDataJob, PreprocessingJob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "__steps__ = ['run']\n",
    "\n",
    "logger = app_logger()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "def run(job: MakeSampleDataJob, context: EventContext) -> PreprocessingJob:\n",
    "    base_path = context.env['data']['partitioned']\n",
    "    client = get_client(context)\n",
    "    try:\n",
    "        df = dd.read_parquet(job.path, \n",
    "                             engine='fastparquet', \n",
    "                             columns=['order_id', 'order_date', 'customer_id', 'email', 'ip_addr', 'order_amount'])\n",
    "        partitioned = {\n",
    "            'customer_id': f'{base_path}/customer_id/',\n",
    "            'email': f'{base_path}/email'\n",
    "        }\n",
    "        for key, path in partitioned.items():\n",
    "            logger.info(context, f\"Partitioning on {key} to {path}...\")\n",
    "            df.set_index(key).to_parquet(path)\n",
    "        return PreprocessingJob(\n",
    "            source=job.path,\n",
    "            partitioned=partitioned\n",
    "        )\n",
    "    except Exception as e:\n",
    "        logger.error(context, e)\n",
    "        return None\n",
    "    finally:\n",
    "        client.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test from notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2021-05-18 20:25:55,943 | INFO | fraud-poc training data.preprocess ALT00617 75563 | Partitioning on customer_id to ./data/partitioned/customer_id/... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2021-05-18T20:25:54.672544+00:00 | stream.name= | stream.msg_id= | stream.consumer_group=\n",
      "2021-05-18 20:25:57,151 | INFO | fraud-poc training data.preprocess ALT00617 75563 | Partitioning on email to ./data/partitioned/email... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2021-05-18T20:25:54.672544+00:00 | stream.name= | stream.msg_id= | stream.consumer_group=\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "PreprocessingJob(source='./data/raw', partitioned={'customer_id': './data/partitioned/customer_id/', 'email': './data/partitioned/email'})"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from hopeit.testing.apps import config, execute_event\n",
    "\n",
    "app_config = config('config/training-pipeline.json')\n",
    "job = MakeSampleDataJob(path='./data/raw', num_batches=2, batch_size=1000, batch_span_days=10, \n",
    "                        num_customers=1000, num_emails=1000, num_ips=1000)\n",
    "result = await execute_event(app_config, 'data.preprocess', job)\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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>order_id</th>\n",
       "      <th>order_date</th>\n",
       "      <th>email</th>\n",
       "      <th>ip_addr</th>\n",
       "      <th>order_amount</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>customer_id</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>00c00d42-6923-4297-87c4-107bc70035be</th>\n",
       "      <td>7beaa6ea-27ea-4082-a0ca-e216b93521a1</td>\n",
       "      <td>2021-05-07 22:49:55+00:00</td>\n",
       "      <td>af9ec8893d8c212276b9916b2599efe88510207d</td>\n",
       "      <td>e18142d2fcd6626e86d40573fd268c2f5f449a60</td>\n",
       "      <td>301.888156</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>048e9104-8369-407b-a0a1-fb06c06760c1</th>\n",
       "      <td>4069005c-02d0-4f41-b065-ad4b82412f20</td>\n",
       "      <td>2021-05-07 05:29:05+00:00</td>\n",
       "      <td>30650ad8775163a27cc73b642d3de1f46a337aaa</td>\n",
       "      <td>110fef07ad03b67a2f30fcebd6ab190bc2d11252</td>\n",
       "      <td>86.057737</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>09da86be-63c3-4265-a799-8fe5efaf270c</th>\n",
       "      <td>3e5bb9cc-bdef-4693-8870-49967d510ae3</td>\n",
       "      <td>2021-05-03 10:08:39+00:00</td>\n",
       "      <td>79cdeec4056ed0714127c4c255d9dafa334ee2a7</td>\n",
       "      <td>ee714f6b18d2b01e2d632f1299d8cb7618099070</td>\n",
       "      <td>945.678360</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>09da86be-63c3-4265-a799-8fe5efaf270c</th>\n",
       "      <td>9cc8c9aa-e342-4e7a-8269-4ebb7bc0d208</td>\n",
       "      <td>2021-05-15 04:11:54+00:00</td>\n",
       "      <td>79cdeec4056ed0714127c4c255d9dafa334ee2a7</td>\n",
       "      <td>ee714f6b18d2b01e2d632f1299d8cb7618099070</td>\n",
       "      <td>955.929156</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0c5e0c2c-fb79-4a60-94c4-501bf2d07278</th>\n",
       "      <td>5b3900c1-e2ca-4226-81ee-0aa47d825005</td>\n",
       "      <td>2021-05-07 09:52:48+00:00</td>\n",
       "      <td>86665ef2c644c5f6a50e7702a7d5b0b88ee3c68b</td>\n",
       "      <td>8a799f9e07fa26a47e84fd8f64ea20ae5601a489</td>\n",
       "      <td>525.625661</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                                  order_id  \\\n",
       "customer_id                                                                  \n",
       "00c00d42-6923-4297-87c4-107bc70035be  7beaa6ea-27ea-4082-a0ca-e216b93521a1   \n",
       "048e9104-8369-407b-a0a1-fb06c06760c1  4069005c-02d0-4f41-b065-ad4b82412f20   \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c  3e5bb9cc-bdef-4693-8870-49967d510ae3   \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c  9cc8c9aa-e342-4e7a-8269-4ebb7bc0d208   \n",
       "0c5e0c2c-fb79-4a60-94c4-501bf2d07278  5b3900c1-e2ca-4226-81ee-0aa47d825005   \n",
       "\n",
       "                                                    order_date  \\\n",
       "customer_id                                                      \n",
       "00c00d42-6923-4297-87c4-107bc70035be 2021-05-07 22:49:55+00:00   \n",
       "048e9104-8369-407b-a0a1-fb06c06760c1 2021-05-07 05:29:05+00:00   \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c 2021-05-03 10:08:39+00:00   \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c 2021-05-15 04:11:54+00:00   \n",
       "0c5e0c2c-fb79-4a60-94c4-501bf2d07278 2021-05-07 09:52:48+00:00   \n",
       "\n",
       "                                                                         email  \\\n",
       "customer_id                                                                      \n",
       "00c00d42-6923-4297-87c4-107bc70035be  af9ec8893d8c212276b9916b2599efe88510207d   \n",
       "048e9104-8369-407b-a0a1-fb06c06760c1  30650ad8775163a27cc73b642d3de1f46a337aaa   \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c  79cdeec4056ed0714127c4c255d9dafa334ee2a7   \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c  79cdeec4056ed0714127c4c255d9dafa334ee2a7   \n",
       "0c5e0c2c-fb79-4a60-94c4-501bf2d07278  86665ef2c644c5f6a50e7702a7d5b0b88ee3c68b   \n",
       "\n",
       "                                                                       ip_addr  \\\n",
       "customer_id                                                                      \n",
       "00c00d42-6923-4297-87c4-107bc70035be  e18142d2fcd6626e86d40573fd268c2f5f449a60   \n",
       "048e9104-8369-407b-a0a1-fb06c06760c1  110fef07ad03b67a2f30fcebd6ab190bc2d11252   \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c  ee714f6b18d2b01e2d632f1299d8cb7618099070   \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c  ee714f6b18d2b01e2d632f1299d8cb7618099070   \n",
       "0c5e0c2c-fb79-4a60-94c4-501bf2d07278  8a799f9e07fa26a47e84fd8f64ea20ae5601a489   \n",
       "\n",
       "                                      order_amount  \n",
       "customer_id                                         \n",
       "00c00d42-6923-4297-87c4-107bc70035be    301.888156  \n",
       "048e9104-8369-407b-a0a1-fb06c06760c1     86.057737  \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c    945.678360  \n",
       "09da86be-63c3-4265-a799-8fe5efaf270c    955.929156  \n",
       "0c5e0c2c-fb79-4a60-94c4-501bf2d07278    525.625661  "
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dd.read_parquet(result.partitioned['customer_id']).head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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>order_id</th>\n",
       "      <th>order_date</th>\n",
       "      <th>customer_id</th>\n",
       "      <th>ip_addr</th>\n",
       "      <th>order_amount</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>email</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>06b27c6a29714eead08e96789b9ed37c6b33c67b</th>\n",
       "      <td>564ca53c-92ae-44e9-b550-857bc3f32299</td>\n",
       "      <td>2021-05-07 16:47:55+00:00</td>\n",
       "      <td>1161afda-4b7f-498b-b07c-a19c8ff25383</td>\n",
       "      <td>2061d686466fd7bdd0c4ff12c0c0431adfb4b584</td>\n",
       "      <td>296.665980</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>06b27c6a29714eead08e96789b9ed37c6b33c67b</th>\n",
       "      <td>78210c4e-1997-477f-92f7-c78da8f6e2b0</td>\n",
       "      <td>2021-05-12 00:37:48+00:00</td>\n",
       "      <td>1161afda-4b7f-498b-b07c-a19c8ff25383</td>\n",
       "      <td>e1bf51a3b72ef02c554f338b28ea67a61f2effa9</td>\n",
       "      <td>675.094589</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>06b27c6a29714eead08e96789b9ed37c6b33c67b</th>\n",
       "      <td>bab6d5be-14fe-4705-9765-ad5fc5a14b82</td>\n",
       "      <td>2021-05-15 01:52:11+00:00</td>\n",
       "      <td>1161afda-4b7f-498b-b07c-a19c8ff25383</td>\n",
       "      <td>e1bf51a3b72ef02c554f338b28ea67a61f2effa9</td>\n",
       "      <td>860.661434</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>06b27c6a29714eead08e96789b9ed37c6b33c67b</th>\n",
       "      <td>d12ec21f-c115-45c4-b5c9-904578faa733</td>\n",
       "      <td>2021-05-01 23:42:34+00:00</td>\n",
       "      <td>1161afda-4b7f-498b-b07c-a19c8ff25383</td>\n",
       "      <td>e1bf51a3b72ef02c554f338b28ea67a61f2effa9</td>\n",
       "      <td>944.537976</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>06b27c6a29714eead08e96789b9ed37c6b33c67b</th>\n",
       "      <td>c1adb330-64c2-40c1-931a-86ba8dde545d</td>\n",
       "      <td>2021-05-14 06:56:31+00:00</td>\n",
       "      <td>1161afda-4b7f-498b-b07c-a19c8ff25383</td>\n",
       "      <td>e1bf51a3b72ef02c554f338b28ea67a61f2effa9</td>\n",
       "      <td>135.732576</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                                      order_id  \\\n",
       "email                                                                            \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  564ca53c-92ae-44e9-b550-857bc3f32299   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  78210c4e-1997-477f-92f7-c78da8f6e2b0   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  bab6d5be-14fe-4705-9765-ad5fc5a14b82   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  d12ec21f-c115-45c4-b5c9-904578faa733   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  c1adb330-64c2-40c1-931a-86ba8dde545d   \n",
       "\n",
       "                                                        order_date  \\\n",
       "email                                                                \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b 2021-05-07 16:47:55+00:00   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b 2021-05-12 00:37:48+00:00   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b 2021-05-15 01:52:11+00:00   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b 2021-05-01 23:42:34+00:00   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b 2021-05-14 06:56:31+00:00   \n",
       "\n",
       "                                                                   customer_id  \\\n",
       "email                                                                            \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  1161afda-4b7f-498b-b07c-a19c8ff25383   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  1161afda-4b7f-498b-b07c-a19c8ff25383   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  1161afda-4b7f-498b-b07c-a19c8ff25383   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  1161afda-4b7f-498b-b07c-a19c8ff25383   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  1161afda-4b7f-498b-b07c-a19c8ff25383   \n",
       "\n",
       "                                                                           ip_addr  \\\n",
       "email                                                                                \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  2061d686466fd7bdd0c4ff12c0c0431adfb4b584   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  e1bf51a3b72ef02c554f338b28ea67a61f2effa9   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  e1bf51a3b72ef02c554f338b28ea67a61f2effa9   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  e1bf51a3b72ef02c554f338b28ea67a61f2effa9   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b  e1bf51a3b72ef02c554f338b28ea67a61f2effa9   \n",
       "\n",
       "                                          order_amount  \n",
       "email                                                   \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b    296.665980  \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b    675.094589  \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b    860.661434  \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b    944.537976  \n",
       "06b27c6a29714eead08e96789b9ed37c6b33c67b    135.732576  "
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dd.read_parquet(result.partitioned['email']).head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}