Add geo features to module 2

adchia · adchia · commit fc0788206000 · 2022-05-17T11:37:05.000-04:00
Signed-off-by: Danny Chiao &lt;danny@tecton.ai&gt;
diff --git a/module_2/README.md b/module_2/README.md
@@ -4,4 +4,27 @@ TODO
 - Define request data
 - Define on demand transforms
   - Note that this can also transforms pushed features (e.g. stream features)
-  - Note that this can combine multiple feature views and request data
+  - Note that this can combine multiple feature views and request data
+
+
+<h1>Module 2: On demand transformations</h1>
+
+In this module, we introduce the concept of on demand transforms. These are transformations that execute on-the-fly and accept as input other feature views or request data.
+
+We  and focus on building features for online serving, and keeping them fresh with a combination of batch feature materialization and stream feature ingestion. We'll be roughly working towards the following:
+
+- **Data sources**: Kafka + File source
+- **Online store**: Redis
+- **Use case**: Predicting churn for drivers in real time.
+
+<img src="architecture.png" width=750>
+
+<h2>Table of Contents</h2>
+
+# Workshop
+## Step 1: Install Feast
+
+First, we install Feast with Spark and Redis support:
+```bash
+pip install "feast[spark,redis]"
+```
diff --git a/module_2/data/driver_stats_lat_lon.parquet b/module_2/data/driver_stats_lat_lon.parquet
diff --git a/module_2/data/gen_lat_lon.py b/module_2/data/gen_lat_lon.py
@@ -0,0 +1,18 @@
+import numpy as np
+import math
+
+import pandas as pd
+
+
+def gen_lat_lon():
+    df = pd.read_parquet("driver_stats.parquet")
+
+    latitude = np.arccos(np.random.random(size=(1807, 1)) * 2 - 1)
+    longitude = np.random.random(size=(1807, 1)) * math.pi * 2
+    df["lat"] = latitude
+    df["lon"] = longitude
+    df.to_parquet("driver_stats_lat_lon.parquet")
+
+
+if __name__ == "__main__":
+    gen_lat_lon()
diff --git a/module_2/feature_repo/data_sources.py b/module_2/feature_repo/data_sources.py
@@ -4,11 +4,11 @@
     PushSource,
     RequestSource,
 )
-from feast.types import Int64
+from feast.types import Int64, Float32
 
 driver_stats = FileSource(
     name="driver_stats_source",
-    path="../data/driver_stats.parquet",  # Should be a remote path in reality for re-use
+    path="../data/driver_stats_lat_lon.parquet",  # Should be a remote path in reality for re-use
     timestamp_field="event_timestamp",
     created_timestamp_column="created",
     description="A table describing the stats of a driver based on hourly logs",
@@ -22,10 +22,12 @@
 
 # Define a request data source which encodes features / information only
 # available at request time (e.g. part of the user initiated HTTP request)
-input_request = RequestSource(
-    name="vals_to_add",
+driver_request = RequestSource(
+    name="driver_request",
     schema=[
         Field(name="val_to_add", dtype=Int64),
         Field(name="val_to_add_2", dtype=Int64),
+        Field(name="lat", dtype=Float32),
+        Field(name="lon", dtype=Float32),
     ],
 )
diff --git a/module_2/feature_repo/feature_services.py b/module_2/feature_repo/feature_services.py
@@ -17,3 +17,16 @@
     ],
     owner="test3@gmail.com",
 )
+
+feature_service = FeatureService(
+    name="model_v3",
+    features=[
+        driver_hourly_stats_view,
+        driver_daily_features_view,
+        transformed_conv_rate,
+        avg_hourly_miles_driven,
+        location_features,
+    ],
+    owner="test3@gmail.com",
+)
+
diff --git a/module_2/feature_repo/features.py b/module_2/feature_repo/features.py
@@ -6,7 +6,7 @@
     Field,
 )
 from feast.on_demand_feature_view import on_demand_feature_view
-from feast.types import Float32, Float64
+from feast.types import Float32, Float64, String
 
 from data_sources import *
 from entities import *
@@ -19,6 +19,8 @@
     schema=[
         Field(name="conv_rate", dtype=Float32),
         Field(name="acc_rate", dtype=Float32),
+        Field(name="lat", dtype=Float32),
+        Field(name="lon", dtype=Float32),
     ],
     online=True,
     source=driver_stats,
@@ -41,16 +43,38 @@
 # Define an on demand feature view which can generate new features based on
 # existing feature views and RequestSource features
 @on_demand_feature_view(
-    sources=[driver_hourly_stats_view, driver_daily_features_view, input_request,],
+    sources=[driver_hourly_stats_view, driver_request],
     schema=[
         Field(name="conv_rate_plus_val1", dtype=Float64),
         Field(name="conv_rate_plus_val2", dtype=Float64),
-        Field(name="daily_miles_driven_normalized", dtype=Float64),
     ],
 )
 def transformed_conv_rate(inputs: pd.DataFrame) -> pd.DataFrame:
     df = pd.DataFrame()
     df["conv_rate_plus_val1"] = inputs["conv_rate"] + inputs["val_to_add"]
     df["conv_rate_plus_val2"] = inputs["conv_rate"] + inputs["val_to_add_2"]
-    df["daily_miles_driven_normalized"] = inputs["daily_miles_driven"] / 1000
+    return df
+
+
+@on_demand_feature_view(
+    sources=[driver_daily_features_view],
+    schema=[Field(name="avg_hourly_miles_driven", dtype=Float64),],
+)
+def avg_hourly_miles_driven(inputs: pd.DataFrame) -> pd.DataFrame:
+    df = pd.DataFrame()
+    df["avg_hourly_miles_driven"] = inputs["daily_miles_driven"] / 24
+    return df
+
+
+@on_demand_feature_view(
+    sources=[driver_daily_features_view, driver_request],
+    schema=[Field(name=f"geohash_{i}", dtype=String) for i in range(1, 7)],
+)
+def location_features(inputs: pd.DataFrame) -> pd.DataFrame:
+    import pygeohash as gh
+
+    df = pd.DataFrame()
+    geohash = df.apply(lambda x: gh.encode(x.lat, x.lon), axis=1)
+    for i in range(1, 7):
+        df[f"geohash_{i}"] = geohash.str[:i]
     return df
