feast_feature_store_entity_registration_and_online_inference.py

1import pandas as pd
2from datetime import datetime
3from feast import FeatureStore, Entity, FeatureView, Field, FileSource
4from feast.types import Int64, Float32
5
6# 1. Define the data source
7# Note: For this example, we assume a parquet file exists at 'data/driver_stats.parquet'
8driver_stats_source = FileSource(
9    path="data/driver_stats.parquet",
10    timestamp_field="event_timestamp",
11    created_timestamp_column="created",
12)
13
14# 2. Define an entity (e.g., a driver)
15driver = Entity(name="driver", join_keys=["driver_id"])
16
17# 3. Define the Feature View
18driver_stats_fv = FeatureView(
19    name="driver_hourly_stats",
20    entities=[driver],
21    ttl=pd.Timedelta(days=1),
22    schema=[
23        Field(name="conv_rate", dtype=Float32),
24        Field(name="acc_rate", dtype=Float32),
25        Field(name="avg_daily_trips", dtype=Int64),
26    ],
27    online=True,
28    source=driver_stats_source,
29)
30
31# 4. Initialize the Feature Store (points to current directory)
32store = FeatureStore(repo_path=".")
33
34# 5. Apply (register) the definitions to the registry
35store.apply([driver, driver_stats_fv])
36
37# 6. Fetch online features for inference
38entity_rows = [
39    {"driver_id": 1001},
40    {"driver_id": 1002},
41]
42
43features = store.get_online_features(
44    features=[
45        "driver_hourly_stats:conv_rate",
46        "driver_hourly_stats:acc_rate",
47        "driver_hourly_stats:avg_daily_trips",
48    ],
49    entity_rows=entity_rows,
50).to_dict()
51
52# Print results
53print(features)