module_2/client/module_2_client.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Retrieving on demand features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Instantiate a `FeatureStore` object"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from feast import FeatureStore\n",
    "import pandas as pd\n",
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "store = FeatureStore(repo_path=\".\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Retrieve historical features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### model_v2 feature service\n",
    "This one leverages dummy `val_to_add` and `val_to_add_2` request data. Request data must be passed in as part of the `entity_df`. This may come from the same source that includes your labels for your model.\n",
    "\n",
    "A quick reminder of the on demand feature view (ODFV) being used in this feature service:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "@on_demand_feature_view(\n",
      "    sources=[driver_hourly_stats_view, val_to_add_request],\n",
      "    schema=[\n",
      "        Field(name=\"conv_rate_plus_val1\", dtype=Float64),\n",
      "        Field(name=\"conv_rate_plus_val2\", dtype=Float64),\n",
      "    ],\n",
      ")\n",
      "def transformed_conv_rate(inputs: pd.DataFrame) -> pd.DataFrame:\n",
      "    df = pd.DataFrame()\n",
      "    df[\"conv_rate_plus_val1\"] = inputs[\"conv_rate\"] + inputs[\"val_to_add\"]\n",
      "    df[\"conv_rate_plus_val2\"] = inputs[\"conv_rate\"] + inputs[\"val_to_add_2\"]\n",
      "    return df\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import dill\n",
    "print(dill.source.getsource(store.get_on_demand_feature_view(\"transformed_conv_rate\").udf))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now let's retrieve historical features from this feature service. The transformation will happen on the fly after doing the point-in-time join to produce the `conv_rate_plus_val1` and `conv_rate_plus_val2` features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "      driver_id           event_timestamp  val_to_add  val_to_add_2  \\\n",
      "360        1001 2021-04-12 10:59:42+00:00           1            10   \n",
      "721        1002 2021-04-12 08:12:10+00:00           2            20   \n",
      "1084       1003 2021-04-12 16:40:26+00:00           3            30   \n",
      "1445       1004 2021-04-12 15:01:12+00:00           4            40   \n",
      "\n",
      "      conv_rate  conv_rate_plus_val1  conv_rate_plus_val2  \n",
      "360    0.521149             1.521149            10.521149  \n",
      "721    0.089014             2.089014            20.089014  \n",
      "1084   0.188855             3.188855            30.188855  \n",
      "1445   0.296492             4.296492            40.296492  \n"
     ]
    }
   ],
   "source": [
    "entity_df = pd.DataFrame.from_dict(\n",
    "    {\n",
    "        \"driver_id\": [1001, 1002, 1003, 1004],\n",
    "        \"event_timestamp\": [\n",
    "            datetime(2021, 4, 12, 10, 59, 42),\n",
    "            datetime(2021, 4, 12, 8, 12, 10),\n",
    "            datetime(2021, 4, 12, 16, 40, 26),\n",
    "            datetime(2021, 4, 12, 15, 1, 12),\n",
    "        ],\n",
    "        \"val_to_add\": [1, 2, 3, 4],\n",
    "        \"val_to_add_2\": [10, 20, 30, 40],\n",
    "    }\n",
    ")\n",
    "training_df = store.get_historical_features(\n",
    "    entity_df=entity_df,\n",
    "    features=store.get_feature_service(\"model_v2\"),\n",
    ").to_df()\n",
    "print(training_df.head())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### model_v3 feature service\n",
    "This one generates geohash features from latitudes and longitudes. This is useful for generating features relating to geographic regions (e.g. A geohash of `qfb9c3mw8hte` represents a sub-region within the region represented by a `qfb` geohash.)\n",
    "\n",
    "Let's look at the on demand feature view used in this feature service:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "@on_demand_feature_view(\n",
      "    sources=[driver_daily_features_view],\n",
      "    schema=[Field(name=f\"geohash_{i}\", dtype=String) for i in range(1, 7)],\n",
      ")\n",
      "def location_features_from_push(inputs: pd.DataFrame) -> pd.DataFrame:\n",
      "    import pygeohash as gh\n",
      "\n",
      "    df = pd.DataFrame()\n",
      "    df[\"geohash\"] = inputs.apply(lambda x: gh.encode(x.lat, x.lon), axis=1).astype(\n",
      "        \"string\"\n",
      "    )\n",
      "\n",
      "    for i in range(1, 7):\n",
      "        df[f\"geohash_{i}\"] = df[\"geohash\"].str[:i].astype(\"string\")\n",
      "    return df\n",
      "\n"
     ]
    }
   ],
   "source": [
    "print(dill.source.getsource(store.get_on_demand_feature_view(\"location_features_from_push\").udf))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we retrieve features. This will compute the `geohash_X` features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "      driver_id           event_timestamp  daily_miles_driven       lat  \\\n",
      "360        1001 2021-04-12 10:59:42+00:00           18.926695  1.265647   \n",
      "721        1002 2021-04-12 08:12:10+00:00           12.005569  0.722192   \n",
      "1084       1003 2021-04-12 16:40:26+00:00           23.490234  1.330712   \n",
      "1445       1004 2021-04-12 15:01:12+00:00           19.204191  0.961260   \n",
      "\n",
      "           lon       geohash geohash_1 geohash_2 geohash_3 geohash_4  \\\n",
      "360   1.150815  s00z4nmuzvtv         s        s0       s00      s00z   \n",
      "721   0.290492  s00hne7x0fqj         s        s0       s00      s00h   \n",
      "1084  2.996348  s04ps4jzgyxq         s        s0       s04      s04p   \n",
      "1445  5.048517  s05t6yupwzyu         s        s0       s05      s05t   \n",
      "\n",
      "     geohash_5 geohash_6  \n",
      "360      s00z4    s00z4n  \n",
      "721      s00hn    s00hne  \n",
      "1084     s04ps    s04ps4  \n",
      "1445     s05t6    s05t6y  \n"
     ]
    }
   ],
   "source": [
    "entity_df = pd.DataFrame.from_dict(\n",
    "    {\n",
    "        \"driver_id\": [1001, 1002, 1003, 1004],\n",
    "        \"event_timestamp\": [\n",
    "            datetime(2021, 4, 12, 10, 59, 42),\n",
    "            datetime(2021, 4, 12, 8, 12, 10),\n",
    "            datetime(2021, 4, 12, 16, 40, 26),\n",
    "            datetime(2021, 4, 12, 15, 1, 12),\n",
    "        ]\n",
    "    }\n",
    ")\n",
    "\n",
    "training_df = store.get_historical_features(\n",
    "    entity_df=entity_df,\n",
    "    features=store.get_feature_service(\"model_v3\"),\n",
    ").to_df()\n",
    "print(training_df.head())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. Retrieve online features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### model_v2 feature service\n",
    "This one leverages dummy `val_to_add` and `val_to_add_2` request data so this is passed into the `entity_rows` parameter. The on demand transformation is executed on the fly and combines in this request data with the pre-computed `conv_rate` feature in the online store."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "conv_rate  :  [0.4045884609222412]\n",
      "conv_rate_plus_val1  :  [1000.4045884609222]\n",
      "conv_rate_plus_val2  :  [2000.4045884609222]\n",
      "driver_id  :  [1001]\n"
     ]
    }
   ],
   "source": [
    "features = store.get_online_features(\n",
    "    features=store.get_feature_service(\"model_v2\"),\n",
    "    entity_rows=[{\"driver_id\": 1001, \"val_to_add\": 1000, \"val_to_add_2\": 2000,}],\n",
    ").to_dict()\n",
    "for key, value in sorted(features.items()):\n",
    "    print(key, \" : \", value)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### model_v3 feature service\n",
    "This one generates geohash features from latitude and longitude values in the online store.\n",
    "\n",
    "Note that this feature service relies on a `PushSource` so no lat / lon values are needed at request time. Perhaps there's a separate thread on the driver's app that asynchronously pushes the driver's location to a Kafka topic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "daily_miles_driven  :  [350.6502685546875]\n",
      "driver_id  :  [1001]\n",
      "geohash_1  :  ['s']\n",
      "geohash_2  :  ['s0']\n",
      "geohash_3  :  ['s07']\n",
      "geohash_4  :  ['s07z']\n",
      "geohash_5  :  ['s07z6']\n",
      "geohash_6  :  ['s07z6m']\n",
      "lat  :  [2.71002197265625]\n",
      "lon  :  [5.3769989013671875]\n"
     ]
    }
   ],
   "source": [
    "features = store.get_online_features(\n",
    "    features=store.get_feature_service(\"model_v3\"),\n",
    "    entity_rows=[{\"driver_id\": 1001}],\n",
    ").to_dict()\n",
    "for key, value in sorted(features.items()):\n",
    "    print(key, \" : \", value)"
   ]
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "7d634b9af180bcb32a446a43848522733ff8f5bbf0cc46dba1a83bede04bf237"
  },
  "kernelspec": {
   "display_name": "Python 3.8.10 64-bit ('python-3.8')",
   "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.10"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}