docs/examples/nyc_taxi.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "443ca22f",
   "metadata": {},
   "source": [
    "[![Open In\n",
    "Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/alibaba/feathub/blob/master/docs/examples/nyc_taxi.ipynb)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "2841dbb4",
   "metadata": {},
   "source": [
    "# NYC Taxi Demo"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "009fc726",
   "metadata": {},
   "source": [
    "This demo shows how to use FeatHub to achieve the following objectives:\n",
    "\n",
    "- Define, extract, transform and materialize features into feature stores.\n",
    "- Transform features into Pandas dataframe for offline training.\n",
    "- Materialize features into online feature store.\n",
    "- Fetch features with on-demand feature transformation from online feature store\n",
    "  for online feature serving.\n",
    "\n",
    "This demo shows these FeatHub capabilities by walking you an example, that\n",
    "trains a GradientBoostingRegressor model on the NYC Taxi Records\n",
    "[dataset](https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page) to\n",
    "predict taxi fares, evaluates the prediction accuracy, and materialize features\n",
    "into online feature store for online feature serving.\n",
    "\n",
    "Please feel free to view this example interactively with Colab by clicking the\n",
    "badge at the top left corner of this notebook."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "6a1ba839",
   "metadata": {},
   "source": [
    "## Install dependencies\n",
    "\n",
    "Execute the following cells to install dependencies. **If the notebook is\n",
    "executed in Colab, restart the runtime after the following cells are executed,\n",
    "in order to make sure Python 3.7 is correctly configured to execute the Python\n",
    "cells.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "caac7a87",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "\n",
    "# Get the Python version string\n",
    "python_version=$(python -c \"import sys; print('.'.join(map(str, sys.version_info[:2])))\")\n",
    "\n",
    "# Define the allowed versions\n",
    "allowed_versions=(\"3.7\" \"3.8\" \"3.9\")\n",
    "\n",
    "# Check if the Python version is allowed\n",
    "if [[ ! \" ${allowed_versions[*]} \" =~ \" $python_version \" ]]; then\n",
    "    # install python 3.7\n",
    "    sudo apt-get update -y\n",
    "    sudo apt-get install python3.7 python3-pip python3.7-distutils python3-apt\n",
    "\n",
    "    # change alternatives\n",
    "    sudo update-alternatives --install /usr/bin/python3 python3 /usr/bin/python3.7 0\n",
    "    sudo update-alternatives --set python3 /usr/bin/python3.7\n",
    "fi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "df5d18a6",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%bash\n",
    "feathub_dependencies=`pip list | grep feathub`\n",
    "if [[ -z \"$feathub_dependencies\" ]]; then\n",
    "    pip install \"feathub-nightly[flink]\"\n",
    "fi"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "561ba2d0",
   "metadata": {},
   "source": [
    "## Download resource files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "57ce8991",
   "metadata": {},
   "outputs": [],
   "source": [
    "!wget -nc https://raw.githubusercontent.com/alibaba/feathub/master/python/feathub/examples/sample_data.csv"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "80213d9d",
   "metadata": {},
   "source": [
    "## Import Python dependencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "990de1e0",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "from datetime import timedelta, datetime\n",
    "from math import sqrt\n",
    "\n",
    "from feathub.table.schema import Schema\n",
    "\n",
    "from sklearn.metrics import mean_squared_error\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.ensemble import GradientBoostingRegressor\n",
    "\n",
    "from feathub.feature_tables.sources.file_system_source import FileSystemSource\n",
    "from feathub.feature_tables.sinks.memory_store_sink import MemoryStoreSink\n",
    "from feathub.feature_views.feature import Feature\n",
    "from feathub.feature_views.on_demand_feature_view import OnDemandFeatureView\n",
    "from feathub.common import types\n",
    "from feathub.feature_tables.sources.memory_store_source import MemoryStoreSource\n",
    "from feathub.feathub_client import FeathubClient\n",
    "from feathub.feature_views.transforms.over_window_transform import OverWindowTransform\n",
    "from feathub.feature_views.derived_feature_view import DerivedFeatureView"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "a673465d",
   "metadata": {},
   "source": [
    "## Initialize FeatHub client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2b14645d",
   "metadata": {},
   "outputs": [],
   "source": [
    "client = FeathubClient(\n",
    "    props={\n",
    "        \"processor\": {\n",
    "            \"type\": \"flink\",\n",
    "            \"flink\": {\n",
    "                \"master\": \"local\",\n",
    "            },\n",
    "        },\n",
    "        \"registry\": {\n",
    "            \"type\": \"local\",\n",
    "            \"local\": {\n",
    "                \"namespace\": \"default\",\n",
    "            },\n",
    "        },\n",
    "        \"feature_service\": {\n",
    "            \"type\": \"local\",\n",
    "            \"local\": {},\n",
    "        },\n",
    "    }\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "1461c1eb",
   "metadata": {},
   "source": [
    "## Specify source dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8bdc9866",
   "metadata": {},
   "outputs": [],
   "source": [
    "source_file_path = \"sample_data.csv\"\n",
    "\n",
    "schema = (\n",
    "    Schema.new_builder()\n",
    "    .column(\"trip_id\", types.Int64)\n",
    "    .column(\"VendorID\", types.Float64)\n",
    "    .column(\"lpep_pickup_datetime\", types.String)\n",
    "    .column(\"lpep_dropoff_datetime\", types.String)\n",
    "    .column(\"store_and_fwd_flag\", types.String)\n",
    "    .column(\"RatecodeID\", types.Float64)\n",
    "    .column(\"PULocationID\", types.Int64)\n",
    "    .column(\"DOLocationID\", types.Int64)\n",
    "    .column(\"passenger_count\", types.Float64)\n",
    "    .column(\"trip_distance\", types.Float64)\n",
    "    .column(\"fare_amount\", types.Float64)\n",
    "    .column(\"extra\", types.Float64)\n",
    "    .column(\"mta_tax\", types.Float64)\n",
    "    .column(\"tip_amount\", types.Float64)\n",
    "    .column(\"tolls_amount\", types.Float64)\n",
    "    .column(\"ehail_fee\", types.Float64)\n",
    "    .column(\"improvement_surcharge\", types.Float64)\n",
    "    .column(\"total_amount\", types.Float64)\n",
    "    .column(\"payment_type\", types.Float64)\n",
    "    .column(\"trip_type\", types.Float64)\n",
    "    .column(\"congestion_surcharge\", types.Float64)\n",
    "    .build()\n",
    ")\n",
    "\n",
    "source = FileSystemSource(\n",
    "    name=\"source_1\",\n",
    "    path=source_file_path,\n",
    "    data_format=\"csv\",\n",
    "    schema=schema,\n",
    "    timestamp_field=\"lpep_dropoff_datetime\",\n",
    "    timestamp_format=\"%Y-%m-%d %H:%M:%S\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "da8a141b",
   "metadata": {},
   "source": [
    "## Define features as transformations on the source dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6af3767",
   "metadata": {},
   "outputs": [],
   "source": [
    "f_trip_time_duration = Feature(\n",
    "    name=\"f_trip_time_duration\",\n",
    "    transform=\"UNIX_TIMESTAMP(lpep_dropoff_datetime) - \"\n",
    "    \"UNIX_TIMESTAMP(lpep_pickup_datetime)\",\n",
    ")\n",
    "\n",
    "f_location_avg_fare = Feature(\n",
    "    name=\"f_location_avg_fare\",\n",
    "    transform=OverWindowTransform(\n",
    "        expr=\"fare_amount\",\n",
    "        agg_func=\"AVG\",\n",
    "        group_by_keys=[\"DOLocationID\"],\n",
    "        window_size=timedelta(days=90),\n",
    "    ),\n",
    ")\n",
    "\n",
    "f_location_max_fare = Feature(\n",
    "    name=\"f_location_max_fare\",\n",
    "    transform=OverWindowTransform(\n",
    "        expr=\"fare_amount\",\n",
    "        agg_func=\"MAX\",\n",
    "        group_by_keys=[\"DOLocationID\"],\n",
    "        window_size=timedelta(days=90),\n",
    "    ),\n",
    ")\n",
    "\n",
    "f_location_total_fare_cents = Feature(\n",
    "    name=\"f_location_total_fare_cents\",\n",
    "    transform=OverWindowTransform(\n",
    "        expr=\"fare_amount * 100\",\n",
    "        agg_func=\"SUM\",\n",
    "        group_by_keys=[\"DOLocationID\"],\n",
    "        window_size=timedelta(days=90),\n",
    "    ),\n",
    ")\n",
    "\n",
    "feature_view_1 = DerivedFeatureView(\n",
    "    name=\"feature_view_1\",\n",
    "    source=source,\n",
    "    features=[\n",
    "        f_trip_time_duration,\n",
    "        f_location_avg_fare,\n",
    "        f_location_max_fare,\n",
    "        f_location_total_fare_cents,\n",
    "    ],\n",
    "    keep_source_fields=True,\n",
    ")\n",
    "\n",
    "f_trip_time_rounded = Feature(\n",
    "    name=\"f_trip_time_rounded\",\n",
    "    transform=\"f_trip_time_duration / 10\",\n",
    "    input_features=[f_trip_time_duration],\n",
    ")\n",
    "\n",
    "f_is_long_trip_distance = Feature(\n",
    "    name=\"f_is_long_trip_distance\",\n",
    "    transform=\"trip_distance > 30\",\n",
    ")\n",
    "\n",
    "feature_view_2 = DerivedFeatureView(\n",
    "    name=\"feature_view_2\",\n",
    "    source=\"feature_view_1\",\n",
    "    features=[\n",
    "        \"f_location_avg_fare\",\n",
    "        f_trip_time_rounded,\n",
    "        f_is_long_trip_distance,\n",
    "        \"f_location_total_fare_cents\",\n",
    "    ],\n",
    "    keep_source_fields=True,\n",
    ")\n",
    "\n",
    "_ = client.build_features(feature_descriptors=[feature_view_1, feature_view_2])"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "27479e79",
   "metadata": {},
   "source": [
    "## Transform features into Pandas DataFrame for offline training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dad15084",
   "metadata": {},
   "outputs": [],
   "source": [
    "final_df = client.get_features(feature_view_2).to_pandas()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "c3571381",
   "metadata": {},
   "source": [
    "## Train a model using the dataset and evaluate the model accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eae9316d",
   "metadata": {},
   "outputs": [],
   "source": [
    "final_df.drop(\n",
    "    [\"lpep_pickup_datetime\", \"lpep_dropoff_datetime\", \"store_and_fwd_flag\"],\n",
    "    axis=1,\n",
    "    inplace=True,\n",
    "    errors=\"ignore\",\n",
    ")\n",
    "\n",
    "final_df.fillna(0, inplace=True)\n",
    "final_df[\"fare_amount\"] = final_df[\"fare_amount\"].astype(\"float64\")\n",
    "\n",
    "train_x, test_x, train_y, test_y = train_test_split(\n",
    "    final_df.drop([\"fare_amount\"], axis=1),\n",
    "    final_df[\"fare_amount\"],\n",
    "    test_size=0.2,\n",
    "    random_state=42,\n",
    ")\n",
    "model = GradientBoostingRegressor()\n",
    "model.fit(train_x, train_y)\n",
    "\n",
    "y_predict = model.predict(test_x)\n",
    "\n",
    "y_actual = test_y.values.flatten().tolist()\n",
    "rmse = sqrt(mean_squared_error(y_actual, y_predict))\n",
    "\n",
    "sum_actuals = sum_errors = 0\n",
    "\n",
    "for actual_val, predict_val in zip(y_actual, y_predict):\n",
    "    abs_error = actual_val - predict_val\n",
    "    if abs_error < 0:\n",
    "        abs_error = abs_error * -1\n",
    "\n",
    "    sum_errors = sum_errors + abs_error\n",
    "    sum_actuals = sum_actuals + actual_val\n",
    "\n",
    "mean_abs_percent_error = sum_errors / sum_actuals\n",
    "\n",
    "print(f\"Model MSE {rmse}\")\n",
    "print(f\"Model MAPE {mean_abs_percent_error}\")\n",
    "print(f\"Model Accuracy: {1 - mean_abs_percent_error}\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "b61a0eee",
   "metadata": {},
   "source": [
    "## Materialize features into online feature store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aaa75ad4",
   "metadata": {},
   "outputs": [],
   "source": [
    "sink = MemoryStoreSink(table_name=\"table_name_1\")\n",
    "selected_features = DerivedFeatureView(\n",
    "    name=\"feature_view_3\",\n",
    "    source=\"feature_view_2\",\n",
    "    features=[\"f_location_avg_fare\", \"f_location_max_fare\"],\n",
    ")\n",
    "client.build_features([selected_features])\n",
    "\n",
    "job = client.materialize_features(\n",
    "    feature_descriptor=selected_features,\n",
    "    sink=sink,\n",
    "    start_datetime=datetime(2020, 1, 1),\n",
    "    end_datetime=datetime(2020, 5, 20),\n",
    "    allow_overwrite=True,\n",
    ")\n",
    "job.wait(timeout_ms=10000)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "bf75973a",
   "metadata": {},
   "source": [
    "## Fetch features from online feature store with on-demand transformations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "25103ffb",
   "metadata": {},
   "outputs": [],
   "source": [
    "source = MemoryStoreSource(\n",
    "    name=\"online_store_source\",\n",
    "    keys=[\"DOLocationID\"],\n",
    "    table_name=\"table_name_1\",\n",
    ")\n",
    "on_demand_feature_view = OnDemandFeatureView(\n",
    "    name=\"on_demand_feature_view\",\n",
    "    features=[\n",
    "        \"online_store_source.f_location_avg_fare\",\n",
    "        \"online_store_source.f_location_max_fare\",\n",
    "        Feature(\n",
    "            name=\"max_avg_ratio\",\n",
    "            transform=\"f_location_max_fare / f_location_avg_fare\",\n",
    "        ),\n",
    "    ],\n",
    "    request_schema=Schema.new_builder().column(\"DOLocationID\", types.Int64).build(),\n",
    ")\n",
    "client.build_features([source, on_demand_feature_view])\n",
    "\n",
    "request_df = pd.DataFrame(np.array([[247]]), columns=[\"DOLocationID\"])\n",
    "online_features = client.get_online_features(\n",
    "    request_df=request_df,\n",
    "    feature_view=on_demand_feature_view,\n",
    ")\n",
    "\n",
    "print(online_features)"
   ]
  }
 ],
 "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.7.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}