Fixing some bugs in example feature repo for spark:

- creating parquet files with timestamps in us units instead of ns (pyspark has problems reading ns timestamps)
- removing on-the-fly features that aren't declared before
- settings join keys for entities.

Author: Felix-neko

sdk/python/feast/templates/spark/bootstrap.py

@@ -21,6 +21,7 @@ def bootstrap():
     driver_stats_df.to_parquet(
         path=str(data_path / "driver_hourly_stats.parquet"),
         allow_truncated_timestamps=True,
+        coerce_timestamps="us",
     )
 
     customer_entities = [201, 202, 203]
@@ -30,6 +31,7 @@ def bootstrap():
     customer_profile_df.to_parquet(
         path=str(data_path / "customer_daily_profile.parquet"),
         allow_truncated_timestamps=True,
+        coerce_timestamps="us",
     )
 
 

sdk/python/feast/templates/spark/feature_repo/example_repo.py

@@ -16,14 +16,8 @@
 
 
 # Entity definitions
-driver = Entity(
-    name="driver",
-    description="driver id",
-)
-customer = Entity(
-    name="customer",
-    description="customer id",
-)
+driver = Entity(name="driver", description="driver id", join_keys=["driver_id"])
+customer = Entity(name="customer", description="customer id", join_keys=["customer_id"])
 
 # Sources
 driver_hourly_stats = SparkSource(

sdk/python/feast/templates/spark/feature_repo/test_workflow.py

@@ -82,8 +82,6 @@ def fetch_historical_features_entity_df(store: FeatureStore, for_batch_scoring:
             "driver_hourly_stats:conv_rate",
             "driver_hourly_stats:acc_rate",
             "driver_hourly_stats:avg_daily_trips",
-            "transformed_conv_rate:conv_rate_plus_val1",
-            "transformed_conv_rate:conv_rate_plus_val2",
         ],
     ).to_df()
     print(training_df.head())
@@ -109,8 +107,6 @@ def fetch_online_features(store, use_feature_service: bool):
         features_to_fetch = [
             "driver_hourly_stats:acc_rate",
             "driver_hourly_stats:avg_daily_trips",
-            "transformed_conv_rate:conv_rate_plus_val1",
-            "transformed_conv_rate:conv_rate_plus_val2",
         ]
     returned_features = store.get_online_features(
         features=features_to_fetch,