07_prepare-db.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#default_exp data.prepare_db"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Prepare Database\n",
    "\n",
    "> This step uses calculated features, and extract most recent values for each customer_id and email in the dataset to initialize a real time database (Redis) that can be used by live fraud prediction service."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "from typing import Dict\n",
    "from datetime import datetime, timezone, timedelta\n",
    "import random\n",
    "import math\n",
    "import dask.dataframe as dd\n",
    "import numpy as np\n",
    "import redis\n",
    "\n",
    "from hopeit.server.serialization import serialize, Serialization, deserialize\n",
    "from hopeit.server.compression import Compression\n",
    "from hopeit.app.context import EventContext\n",
    "from hopeit.app.events import Spawn, SHUFFLE\n",
    "from hopeit.app.api import event_api\n",
    "from hopeit.app.logger import app_logger\n",
    "from hopeit.redis_storage import RedisStorage\n",
    "\n",
    "from fraud_poc.jobs import get_client, FeatureCalcJob, PrepareDbJob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "__steps__ = ['update_database']\n",
    "\n",
    "logger = app_logger()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "def _save_values_by_key(key, path, db_host, db_port):\n",
    "    df = dd.read_parquet(path, engine='fastparquet')\n",
    "    df['key'] = df[key]\n",
    "    return (key, df.map_partitions(_foreach_partition, db_host, db_port, meta=('value', object)).count().compute().item())\n",
    "            \n",
    "def _foreach_partition(df, db_host, db_port):\n",
    "    db = redis.Redis(host=db_host, port=db_port, db=0)\n",
    "    items = df.groupby(['key'])[df.columns].apply(_last_item)\n",
    "    items['order_date'] = items['order_date'].apply(lambda x: x.isoformat())\n",
    "    items = items.apply(lambda x: _persist(x, db), axis=1)\n",
    "    db.close()\n",
    "    return items\n",
    "\n",
    "def _last_item(group):\n",
    "    group = group.sort_values('order_date')\n",
    "    return group.tail(1)     \n",
    "\n",
    "def _persist(item, db):\n",
    "    v = item.to_dict()\n",
    "    key = v['key']\n",
    "    payload = serialize(v, Serialization.PICKLE4, Compression.LZ4)\n",
    "    db.set(key, payload)\n",
    "    return v"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "async def update_database(job: FeatureCalcJob, context: EventContext):\n",
    "    client = get_client(context)\n",
    "    db_host = context.env['db']['host']\n",
    "    db_port = context.env['db']['port']\n",
    "    logger.info(context, f\"Preparing to save to database {db_host}:{db_port}...\")\n",
    "    try:\n",
    "        tasks = []\n",
    "        for key, path in job.features.items():\n",
    "            logger.info(context, f\"Saving latest state for {key} features...\")\n",
    "            tasks.append(client.submit(_save_values_by_key, key, path, db_host, db_port))\n",
    "        res = client.gather(tasks)\n",
    "        return PrepareDbJob(\n",
    "            features=job.features,\n",
    "            db=f'{db_host}:{db_port}',\n",
    "            saved=dict(res)\n",
    "        )\n",
    "    except Exception as e:\n",
    "        logger.error(context, e)\n",
    "        return None\n",
    "    finally:\n",
    "        client.close()    \n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test from notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2020-07-08 14:47:30,084 | INFO | fraud-poc 0.0.1-training data.prepare-db leos13 27299 | Preparing to save to database localhost:6379... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T14:47:28.743022+00:00 | stream.name= | stream.msg_id= | stream.consumer_group=\n",
      "2020-07-08 14:47:30,085 | INFO | fraud-poc 0.0.1-training data.prepare-db leos13 27299 | Saving latest state for customer_id features... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T14:47:28.743022+00:00 | stream.name= | stream.msg_id= | stream.consumer_group=\n",
      "2020-07-08 14:47:30,087 | INFO | fraud-poc 0.0.1-training data.prepare-db leos13 27299 | Saving latest state for email features... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T14:47:28.743022+00:00 | stream.name= | stream.msg_id= | stream.consumer_group=\n",
      "distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing\n",
      "distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "PrepareDbJob(features={'customer_id': './data/features/customer_id/', 'email': './data/features/email/'}, db='localhost:6379', saved={'customer_id': 64, 'email': 100})"
      ]
     },
     "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 = FeatureCalcJob(sources={'customer_id': './data/partitioned/customer_id/', 'email': './data/partitioned/email'}, \n",
    "                     features={'customer_id': './data/features/customer_id/', 'email': './data/features/email/'})\n",
    "result = await execute_event(app_config, 'data.prepare-db', job)\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "ename": "TypeError",
     "evalue": "a bytes-like object is required, not 'NoneType'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------\u001b[0m",
      "\u001b[0;31mTypeError\u001b[0m         Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-8-d14ad5c5c568>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      2\u001b[0m \u001b[0mdb\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mredis\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mRedis\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mhost\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'localhost'\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mport\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;36m6379\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdb\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0mitem\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mdb\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'd555b585-5511-4a16-9f22-819834110239'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 4\u001b[0;31m \u001b[0mdeserialize\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mitem\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mSerialization\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPICKLE4\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mCompression\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mLZ4\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdict\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/opt/dev/anaconda3/envs/fraud-poc/lib/python3.7/site-packages/hopeit/server/serialization.py\u001b[0m in \u001b[0;36mdeserialize\u001b[0;34m(data, serialization, compression, datatype)\u001b[0m\n\u001b[1;32m     57\u001b[0m                 datatype: Type[EventPayloadType]) -> EventPayload:\n\u001b[1;32m     58\u001b[0m     \u001b[0malgos\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0m_SERDESER\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0mserialization\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 59\u001b[0;31m     \u001b[0mdecomp\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mdecompress\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdata\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcompression\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     60\u001b[0m     \u001b[0;32mreturn\u001b[0m \u001b[0malgos\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m2\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdecomp\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdatatype\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/dev/anaconda3/envs/fraud-poc/lib/python3.7/site-packages/hopeit/server/compression.py\u001b[0m in \u001b[0;36mdecompress\u001b[0;34m(data, compression)\u001b[0m\n\u001b[1;32m     98\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     99\u001b[0m \u001b[0;32mdef\u001b[0m \u001b[0mdecompress\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdata\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mbytes\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcompression\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mCompression\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;34m->\u001b[0m \u001b[0mbytes\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 100\u001b[0;31m     \u001b[0;32mreturn\u001b[0m \u001b[0m_DECOMPRESS\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0mcompression\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdata\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/opt/dev/anaconda3/envs/fraud-poc/lib/python3.7/site-packages/hopeit/server/compression.py\u001b[0m in \u001b[0;36m_decompress_lz4\u001b[0;34m(data)\u001b[0m\n\u001b[1;32m     29\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     30\u001b[0m \u001b[0;32mdef\u001b[0m \u001b[0m_decompress_lz4\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdata\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mbytes\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;34m->\u001b[0m \u001b[0mbytes\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 31\u001b[0;31m     \u001b[0;32mreturn\u001b[0m \u001b[0mlz4\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mframe\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdecompress\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdata\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     32\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     33\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mTypeError\u001b[0m: a bytes-like object is required, not 'NoneType'"
     ]
    }
   ],
   "source": [
    "#customer id aggregation\n",
    "db = redis.Redis(host='localhost', port=6379, db=0)\n",
    "item = db.get('d555b585-5511-4a16-9f22-819834110239')\n",
    "deserialize(item, Serialization.PICKLE4, Compression.LZ4, dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#email aggregations\n",
    "item = db.get('1f5d34b02ef1975d5a82dcfe2e53fad6182e118c')\n",
    "deserialize(item, Serialization.PICKLE4, Compression.LZ4, dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}