ate this documentation as we develop this feature.Getting features from a registered and active feature store is possible by creating your own step that interfaces into the feature store:

from datetime import datetime

from typing import Any, Dict, List, Union

import pandas as pd

from zenml import step

from zenml.client import Client

@step

def get_historical_features(

entity_dict: Union[Dict[str, Any], str],

features: List[str],

full_feature_names: bool = False

) -> pd.DataFrame:

"""Feast Feature Store historical data step

Returns:

The historical features as a DataFrame.

"""

feature_store = Client().active_stack.feature_store

if not feature_store:

raise DoesNotExistException(

"The Feast feature store component is not available. "

"Please make sure that the Feast stack component is registered as part of your current active stack."

params.entity_dict["event_timestamp"] = [

datetime.fromisoformat(val)

for val in entity_dict["event_timestamp"]

entity_df = pd.DataFrame.from_dict(entity_dict)

return feature_store.get_historical_features(

entity_df=entity_df,

features=features,

full_feature_names=full_feature_names,

entity_dict = {

"driver_id": [1001, 1002, 1003],

"label_driver_reported_satisfaction": [1, 5, 3],

"event_timestamp": [

datetime(2021, 4, 12, 10, 59, 42).isoformat(),

datetime(2021, 4, 12, 8, 12, 10).isoformat(),

datetime(2021, 4, 12, 16, 40, 26).isoformat(),

],

"val_to_add": [1, 2, 3],

"val_to_add_2": [10, 20, 30],

features = [

"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",

@pipeline

def my_pipeline():

my_features = get_historical_features(entity_dict, features)

...