{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Feast Basic Customer Transactions Example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is a minimal example of using Feast. In this example we will\n",
    "1. Create a synthetic customer feature dataset\n",
    "2. Register a feature set to represent these features in Feast\n",
    "3. Ingest these features into Feast\n",
    "4. Create a feature query and retrieve historical feature data\n",
    "5. Create a feature query and retrieve online feature data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Clone Feast and install all dependencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!git clone https://github.com/gojek/feast.git \\\n",
    "&& cd feast/sdk/python/ &&  pip install --upgrade --quiet -e . \\\n",
    "&& pip install --quiet --upgrade pandas numpy protobuf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Import necessary modules"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "from pytz import timezone, utc\n",
    "from feast import Client, FeatureSet, Entity, ValueType\n",
    "from feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest\n",
    "from feast.types.Value_pb2 import Value as Value\n",
    "from google.protobuf.duration_pb2 import Duration\n",
    "from datetime import datetime, timedelta\n",
    "from random import randrange\n",
    "import random"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Configure Feast services and connect the Feast client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "CORE_URL = 'localhost:6565'\n",
    "ONLINE_SERVING_URL = 'localhost:6566'\n",
    "BATCH_SERVING_URL = 'localhost:6567'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client = Client(core_url=CORE_URL, serving_url=BATCH_SERVING_URL) # Connect to Feast Core\n",
    "client.create_project('customer_project')\n",
    "client.set_project('customer_project')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Create synthetic customer features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we will create customer features for 5 customers over a month. Each customer will have a set of features for every day."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "days = [datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0).replace(tzinfo=utc) \\\n",
    "        - timedelta(day) for day in range(31)]\n",
    "\n",
    "customers = [1001, 1002, 1003, 1004, 1005]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "customer_features = pd.DataFrame(\n",
    "    {\n",
    "        \"datetime\": [day for day in days for customer in customers],\n",
    "        \"customer_id\": [customer for day in days for customer in customers],\n",
    "        \"daily_transactions\": [np.random.rand() * 10 for _ in range(len(days) * len(customers))],\n",
    "        \"total_transactions\": [np.random.randint(100) for _ in range(len(days) * len(customers))],\n",
    "    }\n",
    ")\n",
    "\n",
    "print(customer_features.head(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5. Create feature set for customer features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we will create a feature set for these features. Feature sets are essentially a schema that represent\n",
    "feature values. Feature sets allow Feast to both identify feature values and their structure. \n",
    "\n",
    "In this case we need to define any entity columns as well as the maximum age. The entity column in this case is \"customer_id\". Max age is set to 1 day (defined in seconds). This means that for each feature query during retrieval, the serving API will only retrieve features up to a maximum of 1 day per provided timestamp and entity combination. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "customer_fs = FeatureSet(\n",
    "    \"customer_transactions\",\n",
    "    max_age=Duration(seconds=86400),\n",
    "    entities=[Entity(name='customer_id', dtype=ValueType.INT64)]\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we are automatically inferring the schema from the provided dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "customer_fs.infer_fields_from_df(customer_features, replace_existing_features=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6. Register feature set with Feast Core"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The apply() method will register the provided feature set with Feast core, allowing users to retrieve features from this feature set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.apply(customer_fs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We test the retrieval of this feature set object (not its data), to ensure that we have the latest version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "customer_fs = client.get_feature_set(\"customer_transactions\")\n",
    "print(customer_fs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 7. Ingest data into Feast for a feature set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.ingest(\"customer_transactions\", customer_features)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 8. Create a batch retrieval query"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to retrieve historical feature data, the user must provide an entity_rows dataframe. This dataframe contains a combination of timestamps and entities. In this case, the user must provide both customer_ids and timestamps. \n",
    "\n",
    "We will randomly generate timestamps over the last 30 days, and assign customer_ids to them. When these entity rows are sent to the Feast Serving API to retrieve feature values, along with a list of feature ids, Feast is then able to attach the correct feature values to each entity row. The one exception is if the feature values fall outside of the maximum age window."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "event_timestamps = [datetime.utcnow().replace(tzinfo=utc) - timedelta(days=randrange(15), hours=randrange(24), minutes=randrange(60)) for day in range(30)]\n",
    "\n",
    "entity_rows = pd.DataFrame(\n",
    "    {\n",
    "        \"datetime\": event_timestamps,\n",
    "        \"customer_id\": [customers[idx % len(customers)] for idx in range(len(event_timestamps))],\n",
    "    }\n",
    ")\n",
    "\n",
    "print(entity_rows.head(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 9. Retrieve historical/batch features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "job = client.get_batch_features(\n",
    "                            feature_refs=[\n",
    "                                f\"daily_transactions\", \n",
    "                                f\"total_transactions\", \n",
    "                               ],\n",
    "                            entity_rows=entity_rows\n",
    "                         )\n",
    "df = job.to_dataframe()\n",
    "print(df.head(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 10. Retrieve online features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The process of retrieving features from the online API is very similar to that of the batch API. The only major difference is that users do not have to provide timestamps (only the latest features are returned, as long as they are within the maximum age window)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "online_client = Client(core_url=CORE_URL, serving_url=ONLINE_SERVING_URL)\n",
    "online_client.set_project(\"customer_project\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The example below retrieves online features for a single customer: \"1001\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "online_features = online_client.get_online_features(\n",
    "    feature_refs=[\n",
    "        f\"daily_transactions\",\n",
    "        f\"total_transactions\",\n",
    "    ],\n",
    "    entity_rows=[\n",
    "        GetOnlineFeaturesRequest.EntityRow(\n",
    "            fields={\n",
    "                \"customer_id\": Value(\n",
    "                    int64_val=1001)\n",
    "            }\n",
    "        )\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(online_features)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.7.3"
  },
  "pycharm": {
   "stem_cell": {
    "cell_type": "raw",
    "metadata": {
     "collapsed": false
    },
    "source": []
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}