{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Overview\n",
    "\n",
    "In this tutorial, we'll use Feast to generate training data and power online model inference for a \n",
    "ride-sharing driver satisfaction prediction model. Feast solves several common issues in this flow:\n",
    "\n",
    "1. **Training-serving skew and complex data joins:** Feature values often exist across multiple tables. Joining \n",
    "   these datasets can be complicated, slow, and error-prone.\n",
    "   * Feast joins these tables with battle-tested logic that ensures _point-in-time_ correctness so future feature \n",
    "     values do not leak to models.\n",
    "2. **Online feature availability:** At inference time, models often need access to features that aren't readily \n",
    "   available and need to be precomputed from other data sources.\n",
    "   * Feast manages deployment to a variety of online stores (e.g. DynamoDB, Redis, Google Cloud Datastore) and \n",
    "     ensures necessary features are consistently _available_ and _freshly computed_ at inference time.\n",
    "3. **Feature and model versioning:** Different teams within an organization are often unable to reuse \n",
    "   features across projects, resulting in duplicate feature creation logic. Models have data dependencies that need \n",
    "   to be versioned, for example when running A/B tests on model versions.\n",
    "   * Feast enables discovery of and collaboration on previously used features and enables versioning of sets of \n",
    "     features (via _feature services_).\n",
    "   * _(Experimental)_ Feast enables light-weight feature transformations so users can re-use transformation logic \n",
    "     across online / offline use cases and across models.\n",
    "\n",
    "We will:\n",
    "1. Deploy a local feature store with a **Parquet file offline store** and **Sqlite online store**.\n",
    "2. Build a training dataset using our time series features from our **Parquet files**.\n",
    "3. Materialize feature values from the offline store into the online store.\n",
    "4. Read the latest features from the online store for inference."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 1: Inspect a feature repository\n",
    "\n",
    "A feature repository is a directory that contains the configuration of the feature store and individual features. This configuration is written as code (Python/YAML) and it's highly recommended that teams track it centrally using git. See [Feature Repository](https://docs.feast.dev/reference/feature-repository) for a detailed explanation of feature repositories.\n",
    "\n",
    "The easiest way to create a new feature repository to use the `feast init` command. This creates a scaffolding with initial demo data.\n",
    "\n",
    "### Demo data scenario \n",
    "- We have surveyed some drivers for how satisfied they are with their experience in a ride-sharing app. \n",
    "- We want to generate predictions for driver satisfaction for the rest of the users so we can reach out to potentially dissatisfied users."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 1a: Inspecting the feature repository\n",
    "\n",
    "Let's take a look at the demo repo itself. It breaks down into\n",
    "\n",
    "\n",
    "* `data/` contains raw demo parquet data\n",
    "* `definition.py` contains demo feature definitions\n",
    "* `feature_store.yaml` contains a demo setup configuring where data sources are\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!cd /mnt/shared/feast-store\n",
    "!ls -R"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 1b: Inspecting the project configuration\n",
    "Let's inspect the setup of the project in `feature_store.yaml`. \n",
    "\n",
    "The key line defining the overall architecture of the feature store is the **provider**. \n",
    "\n",
    "The provider value sets default offline and online stores. \n",
    "* The offline store provides the compute layer to process historical data (for generating training data & feature \n",
    "  values for serving). \n",
    "* The online store is a low latency store of the latest feature values (for powering real-time inference).\n",
    "\n",
    "Valid values for `provider` in `feature_store.yaml` are:\n",
    "\n",
    "* local: use file source with SQLite/Redis\n",
    "* gcp: use BigQuery/Snowflake with Google Cloud Datastore/Redis\n",
    "* aws: use Redshift/Snowflake with DynamoDB/Redis\n",
    "\n",
    "Note that there are many other offline / online stores Feast works with, including Azure, Hive, Trino, and PostgreSQL via community plugins. See https://docs.feast.dev/roadmap for all supported connectors.\n",
    "\n",
    "A custom setup can also be made by following [Customizing Feast](https://docs.feast.dev/v/master/how-to-guides/customizing-feast)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!cat /mnt/shared/feast-store/feature_store.yaml"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "FnMlk4zshywp"
   },
   "source": [
    "### Inspecting the raw data\n",
    "\n",
    "The raw feature data we have in this demo is stored in a local parquet file. The dataset captures hourly stats of a driver in a ride-sharing app."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from pathlib import Path\n",
    "\n",
    "pd.read_parquet(f\"{str(Path.home())}/user/Feast/data/driver_stats.parquet\", engine=\"pyarrow\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Register feature definitions and deploy your feature store\n",
    "\n",
    "`feast apply` scans python files in the current directory for feature/entity definitions and deploys infrastructure according to `feature_store.yaml`.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 2a: Inspecting feature definitions\n",
    "Now we run `feast apply` to register the feature views and entities defined in `definition.py`, and sets up SQLite online store tables. Note that we had previously specified SQLite as the online store in `feature_store.yaml` by specifying a `local` provider."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Step 2a: Inspecting feature definitions\n",
    "!cat definitions.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 2b: Applying feature definitions\n",
    "Now we run `feast apply` to register the feature views and entities defined in `definitions.py`, and sets up SQLite online store tables. Note that we had previously specified SQLite as the online store in `feature_store.yaml` by specifying a `local` provider."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from definitions import driver, driver_stats_source, driver_stats_fv, driver_stats_push_source, input_request, driver_activity_v1, driver_activity_v2, driver_activity_v3, driver_stats_fresh_fv\n",
    "from definitions import transformed_conv_rate\n",
    "from pprint import pprint\n",
    "from feast import FeatureStore, Entity, FeatureView, Feature, ValueType, FileSource, RepoConfig\n",
    "\n",
    "from datetime import datetime, timedelta\n",
    "import pandas as pd\n",
    "\n",
    "Store = \"/mnt/shared/feast-store\"\n",
    "fs = FeatureStore(repo_path=Store)\n",
    "\n",
    "fs.apply([driver, driver_stats_source, driver_stats_fv, driver_stats_push_source, input_request, driver_activity_v1, driver_activity_v2, transformed_conv_rate, driver_activity_v3, driver_stats_fresh_fv])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Generating training data or powering batch scoring models\n",
    "\n",
    "To train a model, we need features and labels. Often, this label data is stored separately (e.g. you have one table storing user survey results and another set of tables with feature values). Feast can help generate the features that map to these labels.\n",
    "\n",
    "Feast needs a list of **entities** (e.g. driver ids) and **timestamps**. Feast will intelligently join relevant \n",
    "tables to create the relevant feature vectors. There are two ways to generate this list:\n",
    "1. The user can query that table of labels with timestamps and pass that into Feast as an _entity dataframe_ for \n",
    "training data generation. \n",
    "2. The user can also query that table with a *SQL query* which pulls entities. See the documentation on [feature retrieval](https://docs.feast.dev/getting-started/concepts/feature-retrieval) for details    \n",
    "\n",
    "* Note that we include timestamps because we want the features for the same driver at various timestamps to be used in a model.\n",
    "\n",
    "### Step 3a: Generating training data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "import pandas as pd\n",
    "\n",
    "from feast import FeatureStore\n",
    "\n",
    "# The entity dataframe is the dataframe we want to enrich with feature values\n",
    "# Note: see https://docs.feast.dev/getting-started/concepts/feature-retrieval for more details on how to retrieve\n",
    "# for all entities in the offline store instead\n",
    "entity_df = pd.DataFrame.from_dict(\n",
    "    {\n",
    "        # entity's join key -> entity values\n",
    "        \"driver_id\": [1001, 1002, 1003],\n",
    "        # \"event_timestamp\" (reserved key) -> timestamps\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",
    "        ],\n",
    "        # (optional) label name -> label values. Feast does not process these\n",
    "        \"label_driver_reported_satisfaction\": [1, 5, 3],\n",
    "        # values we're using for an on-demand transformation\n",
    "        \"val_to_add\": [1, 2, 3],\n",
    "        \"val_to_add_2\": [10, 20, 30],\n",
    "    }\n",
    ")\n",
    "\n",
    "training_df = fs.get_historical_features(\n",
    "    entity_df=entity_df,\n",
    "    features=[\n",
    "        \"driver_hourly_stats:conv_rate\",\n",
    "        \"driver_hourly_stats:acc_rate\",\n",
    "        \"driver_hourly_stats:avg_daily_trips\",\n",
    "        \"transformed_conv_rate:conv_rate_plus_val1\",\n",
    "        \"transformed_conv_rate:conv_rate_plus_val2\",\n",
    "    ],\n",
    ").to_df()\n",
    "\n",
    "print(\"----- Feature schema -----\\n\")\n",
    "print(training_df.info())\n",
    "\n",
    "print()\n",
    "print(\"----- Example features -----\\n\")\n",
    "print(training_df.head())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3b: Run offline inference (batch scoring)\n",
    "To power a batch model, we primarily need to generate features with the `get_historical_features` call, but using the current timestamp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# entity_df[\"event_timestamp\"] = pd.to_datetime(\"now\", utc=True)\n",
    "training_df = fs.get_historical_features(\n",
    "    entity_df=entity_df,\n",
    "    features=[\n",
    "        \"driver_hourly_stats:conv_rate\",\n",
    "        \"driver_hourly_stats:acc_rate\",\n",
    "        \"driver_hourly_stats:avg_daily_trips\",\n",
    "        \"transformed_conv_rate:conv_rate_plus_val1\",\n",
    "        \"transformed_conv_rate:conv_rate_plus_val2\",\n",
    "    ],\n",
    ").to_df()\n",
    "\n",
    "print(\"\\n----- Example features -----\\n\")\n",
    "print(training_df.head())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Load features into your online store"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 4a: Using `materialize_incremental`\n",
    "\n",
    "We now serialize the latest values of features since the beginning of time to prepare for serving (note: `materialize_incremental` serializes all new features since the last `materialize` call).\n",
    "\n",
    "While running the cell if we get an error that online store is not populated, uncomment the commented line in the below cell and re run the cell\n",
    "\nRefer to [documentation](https://docs.feast.dev/how-to-guides/feast-snowflake-gcp-aws/load-data-into-the-online-store) page on the materialize and materialize_incremental usage.\n",
    "\nAn alternative to using the CLI command is to use Python:\n",
    "\n",
    "```bash\n",
    "CURRENT_TIME=$(date -u +\"%Y-%m-%dT%H:%M:%S\")\n",
    "feast materialize-incremental $CURRENT_TIME\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "fs.materialize_incremental(datetime.now())\n",
    "#fs.materialize(start_date=datetime(2021, 4, 12),end_date=datetime.now())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 4b: Inspect materialized features\n",
    "\n",
    "Note that now there are `online_store.db` and `registry.db`, which store the materialized features and schema information, respectively."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"--- Data directory ---\")\n",
    "!ls /mnt/shared/feast-store/data\n",
    "\n",
    "import sqlite3\n",
    "import pandas as pd\n",
    "con = sqlite3.connect(\"/mnt/shared/feast-store/data/online.db\")\n",
    "print(\"\\n--- Schema of online store ---\")\n",
    "print(\n",
    "    pd.read_sql_query(\n",
    "        \"SELECT * FROM ezaf_feast_repo_driver_hourly_stats\", con).columns.tolist())\n",
    "con.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Fetching real-time feature vectors for online inference"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At inference time, we need to quickly read the latest feature values for different drivers (which otherwise might have existed only in batch sources) from the online feature store using `get_online_features()`. These feature vectors can then be fed to the model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pprint import pprint\n",
    "from feast import FeatureStore\n",
    "\n",
    "feature_vector = fs.get_online_features(\n",
    "    features=[\n",
    "        \"driver_hourly_stats:acc_rate\",\n",
    "        \"driver_hourly_stats:avg_daily_trips\",\n",
    "        \"transformed_conv_rate:conv_rate_plus_val1\",\n",
    "        \"transformed_conv_rate:conv_rate_plus_val2\",\n",
    "    ],\n",
    "    entity_rows=[\n",
    "        # {join_key: entity_value}\n",
    "        {\n",
    "            \"driver_id\": 1001,\n",
    "            \"val_to_add\": 1000,\n",
    "            \"val_to_add_2\": 2000,\n",
    "        },\n",
    "        {\n",
    "            \"driver_id\": 1002,\n",
    "            \"val_to_add\": 1001,\n",
    "            \"val_to_add_2\": 2002,\n",
    "        },\n",
    "    ],\n",
    ").to_dict()\n",
    "\n",
    "pprint(feature_vector)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fetching features using feature services\n",
    "You can also use feature services to manage multiple features, and decouple feature view definitions and the features needed by end applications. The feature store can also be used to fetch either online or historical features using the same api below. More information can be found [here](https://docs.feast.dev/getting-started/concepts/feature-retrieval).\n",
    "\n",
    " The `driver_activity_v1` feature service pulls all features from the `driver_hourly_stats` feature view:\n",
    "\n",
    "```python\n",
    "driver_stats_fs = FeatureService(\n",
    "    name=\"driver_activity_v1\", features=[driver_hourly_stats_view]\n",
    ")\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from feast import FeatureStore\n",
    "\n",
    "feature_service = fs.get_feature_service(\"driver_activity_v1\")\n",
    "feature_vector = fs.get_online_features(\n",
    "    features=feature_service,\n",
    "    entity_rows=[\n",
    "        # {join_key: entity_value}\n",
    "        {\n",
    "            \"driver_id\": 1001,\n",
    "            \"val_to_add\": 1000,\n",
    "            \"val_to_add_2\": 2000,\n",
    "        },\n",
    "        {\n",
    "            \"driver_id\": 1002,\n",
    "            \"val_to_add\": 1001,\n",
    "            \"val_to_add_2\": 2002,\n",
    "        },\n",
    "    ],\n",
    ").to_dict()\n",
    "pprint(feature_vector)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 6: Making streaming features available in Feast\n",
    "Feast does not directly ingest from streaming sources. Instead, Feast relies on a push-based model to push features into Feast. You can write a streaming pipeline that generates features, which can then be pushed to the offline store, the online store, or both (depending on your needs).\n",
    "\n",
    "This relies on the `PushSource` defined above. Pushing to this source will populate all dependent feature views with the pushed feature values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from feast.data_source import PushMode\n",
    "\n",
    "print(\"\\n--- Simulate a stream event ingestion of the hourly stats df ---\")\n",
    "event_df = pd.DataFrame.from_dict(\n",
    "    {\n",
    "        \"driver_id\": [1001],\n",
    "        \"event_timestamp\": [\n",
    "            datetime(2021, 5, 13, 10, 59, 42),\n",
    "        ],\n",
    "        \"created\": [\n",
    "            datetime(2021, 5, 13, 10, 59, 42),\n",
    "        ],\n",
    "        \"conv_rate\": [1.0],\n",
    "        \"acc_rate\": [1.0],\n",
    "        \"avg_daily_trips\": [1000],\n",
    "    }\n",
    ")\n",
    "print(event_df)\n",
    "fs.push(\"driver_stats_push_source\", event_df, to=PushMode.ONLINE_AND_OFFLINE)"
   ]
  }
 ],
 "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.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}