diff --git a/README.md b/README.md index 4fe0c11083f..2e3c7b9086e 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,7 @@ store = FeatureStore(repo_path=".") training_df = store.get_historical_features( entity_df=entity_df, - feature_refs = [ + features = [ 'driver_hourly_stats:conv_rate', 'driver_hourly_stats:acc_rate', 'driver_hourly_stats:avg_daily_trips' @@ -101,7 +101,7 @@ from feast import FeatureStore store = FeatureStore(repo_path=".") feature_vector = store.get_online_features( - feature_refs=[ + features=[ 'driver_hourly_stats:conv_rate', 'driver_hourly_stats:acc_rate', 'driver_hourly_stats:avg_daily_trips' diff --git a/docs/concepts/data-model-and-concepts.md b/docs/concepts/data-model-and-concepts.md index d1dcad56e5c..396c9187205 100644 --- a/docs/concepts/data-model-and-concepts.md +++ b/docs/concepts/data-model-and-concepts.md @@ -16,7 +16,7 @@ Feature references are used for the retrieval of features from Feast: ```python online_features = fs.get_online_features( - feature_refs=[ + features=[ 'driver_locations:lon', 'drivers_activity:trips_today' ], @@ -53,7 +53,7 @@ A collection of entity rows. Entity dataframes are the "left table" that is enri ```python training_df = store.get_historical_features( entity_df=entity_df, - feature_refs = [ + features = [ 'drivers_activity:trips_today' 'drivers_activity:rating' ], diff --git a/docs/concepts/feature-service.md b/docs/concepts/feature-service.md new file mode 100644 index 00000000000..791b6345fbc --- /dev/null +++ b/docs/concepts/feature-service.md @@ -0,0 +1,27 @@ +# Feature Service + +A feature service is an object that represents a logical group of features from one or more [feature views](feature-view.md). +Feature Services allows features from within a feature view to be used as needed by an ML model. Users can expect to create one feature service per model, allowing for tracking of the features used by models. + +{% tabs %} +{% tab title="driver\_trips\_feature\_service.py" %} +```python +from driver_ratings_feature_view import driver_ratings_fv +from driver_trips_feature_view import driver_stats_fv + +driver_stats_fs = FeatureService( + name="driver_activity", + features=[driver_stats_fv, driver_ratings_fv[["lifetime_rating"]]] +) +``` +{% endtab %} +{% endtabs %} + +Feature services are used during + +* The generation of training datasets when querying feature views in order to find historical feature values. A single training dataset may consist of features from multiple feature views. +* Retrieval of features from the online store. The features retrieved from the online store may also belong to multiple feature views. + +{% hint style="info" %} +Applying a feature service does not result in an actual service being deployed. +{% endhint %} diff --git a/docs/feast-on-kubernetes/user-guide/getting-training-features.md b/docs/feast-on-kubernetes/user-guide/getting-training-features.md index 8bac2db5a7b..e0f52a8cd96 100644 --- a/docs/feast-on-kubernetes/user-guide/getting-training-features.md +++ b/docs/feast-on-kubernetes/user-guide/getting-training-features.md @@ -8,7 +8,7 @@ Below is an example of the process required to produce a training dataset: ```python # Feature references with target feature -feature_refs = [ +features = [ "driver_trips:average_daily_rides", "driver_trips:maximum_daily_rides", "driver_trips:rating", @@ -24,7 +24,7 @@ entity_source = FileSource( # Retrieve historical dataset from Feast. historical_feature_retrieval_job = client.get_historical_features( - feature_refs=feature_refs, + features=features, entity_rows=entity_source ) diff --git a/docs/getting-started/build-a-training-dataset.md b/docs/getting-started/build-a-training-dataset.md index 6e7eb8833ff..3a673a6afaf 100644 --- a/docs/getting-started/build-a-training-dataset.md +++ b/docs/getting-started/build-a-training-dataset.md @@ -61,7 +61,7 @@ from feast import FeatureStore fs = FeatureStore(repo_path="path/to/your/feature/repo") training_df = fs.get_historical_features( - feature_refs=[ + features=[ "driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate" ], diff --git a/docs/getting-started/read-features-from-the-online-store.md b/docs/getting-started/read-features-from-the-online-store.md index 37c203e0ee8..d5abfd00ea9 100644 --- a/docs/getting-started/read-features-from-the-online-store.md +++ b/docs/getting-started/read-features-from-the-online-store.md @@ -19,7 +19,7 @@ Please ensure that you have materialized \(loaded\) your feature values into the Create a list of features that you would like to retrieve. This list typically comes from the model training step and should accompany the model binary. ```python -feature_refs = [ +features = [ "driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate" ] @@ -32,7 +32,7 @@ Next, we will create a feature store object and call `get_online_features()` whi ```python fs = FeatureStore(repo_path="path/to/feature/repo") online_features = fs.get_online_features( - feature_refs=feature_refs, + features=features, entity_rows=[ {"driver_id": 1001}, {"driver_id": 1002}] diff --git a/docs/quickstart.md b/docs/quickstart.md index 866f96f3025..d41dc7b09d6 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -69,7 +69,7 @@ store = FeatureStore(repo_path=".") training_df = store.get_historical_features( entity_df=entity_df, - feature_refs=[ + features=[ "driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate", "driver_hourly_stats:avg_daily_trips", @@ -105,7 +105,7 @@ from feast import FeatureStore store = FeatureStore(repo_path=".") feature_vector = store.get_online_features( - feature_refs=[ + features=[ "driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate", "driver_hourly_stats:avg_daily_trips", diff --git a/docs/user-guide/getting-training-features.md b/docs/user-guide/getting-training-features.md index a3897b10190..b9d0b050f29 100644 --- a/docs/user-guide/getting-training-features.md +++ b/docs/user-guide/getting-training-features.md @@ -8,7 +8,7 @@ Below is an example of the process required to produce a training dataset: ```python # Feature references with target feature -feature_refs = [ +features = [ "driver_trips:average_daily_rides", "driver_trips:maximum_daily_rides", "driver_trips:rating", @@ -24,7 +24,7 @@ entity_source = FileSource( # Retrieve historical dataset from Feast. historical_feature_retrieval_job = client.get_historical_features( - feature_refs=feature_refs, + features=features, entity_rows=entity_source ) diff --git a/protos/feast/core/Registry.proto b/protos/feast/core/Registry.proto index bb32d6fc3c9..6900a5b1e74 100644 --- a/protos/feast/core/Registry.proto +++ b/protos/feast/core/Registry.proto @@ -22,6 +22,7 @@ option java_outer_classname = "RegistryProto"; option go_package = "github.com/feast-dev/feast/sdk/go/protos/feast/core"; import "feast/core/Entity.proto"; +import "feast/core/FeatureService.proto"; import "feast/core/FeatureTable.proto"; import "feast/core/FeatureView.proto"; import "google/protobuf/timestamp.proto"; @@ -30,8 +31,10 @@ message Registry { repeated Entity entities = 1; repeated FeatureTable feature_tables = 2; repeated FeatureView feature_views = 6; + repeated FeatureService feature_services = 7; string registry_schema_version = 3; // to support migrations; incremented when schema is changed string version_id = 4; // version id, random string generated on each update of the data; now used only for debugging purposes google.protobuf.Timestamp last_updated = 5; + } diff --git a/sdk/python/feast/__init__.py b/sdk/python/feast/__init__.py index 430fd9f715b..4fdd895fea0 100644 --- a/sdk/python/feast/__init__.py +++ b/sdk/python/feast/__init__.py @@ -10,6 +10,7 @@ from .data_source import KafkaSource, KinesisSource, SourceType from .entity import Entity from .feature import Feature +from .feature_service import FeatureService from .feature_store import FeatureStore from .feature_table import FeatureTable from .feature_view import FeatureView @@ -34,6 +35,7 @@ "KafkaSource", "KinesisSource", "Feature", + "FeatureService", "FeatureStore", "FeatureTable", "FeatureView", diff --git a/sdk/python/feast/cli.py b/sdk/python/feast/cli.py index a1fe6cd7958..a7028640a48 100644 --- a/sdk/python/feast/cli.py +++ b/sdk/python/feast/cli.py @@ -129,6 +129,63 @@ def entity_list(ctx: click.Context): print(tabulate(table, headers=["NAME", "DESCRIPTION", "TYPE"], tablefmt="plain")) +@cli.group(name="feature-services") +def feature_services_cmd(): + """ + Access feature services + """ + pass + + +@feature_services_cmd.command("describe") +@click.argument("name", type=click.STRING) +@click.pass_context +def feature_service_describe(ctx: click.Context, name: str): + """ + Describe a feature service + """ + repo = ctx.obj["CHDIR"] + cli_check_repo(repo) + store = FeatureStore(repo_path=str(repo)) + + try: + feature_service = store.get_feature_service(name) + except FeastObjectNotFoundException as e: + print(e) + exit(1) + + print( + yaml.dump( + yaml.safe_load(str(feature_service)), + default_flow_style=False, + sort_keys=False, + ) + ) + + +@feature_services_cmd.command(name="list") +@click.pass_context +def feature_service_list(ctx: click.Context): + """ + List all feature services + """ + repo = ctx.obj["CHDIR"] + cli_check_repo(repo) + store = FeatureStore(repo_path=str(repo)) + feature_services = [] + for feature_service in store.list_feature_services(): + feature_names = [] + for projection in feature_service.features: + feature_names.extend( + [f"{projection.name}:{feature.name}" for feature in projection.features] + ) + feature_services.append([feature_service.name, ", ".join(feature_names)]) + + from tabulate import tabulate + + print(tabulate(feature_services, headers=["NAME", "FEATURES"], tablefmt="plain")) + + @cli.group(name="feature-views") def feature_views_cmd(): """ diff --git a/sdk/python/feast/errors.py b/sdk/python/feast/errors.py index 61fee8f9187..bdcba937926 100644 --- a/sdk/python/feast/errors.py +++ b/sdk/python/feast/errors.py @@ -22,6 +22,16 @@ def __init__(self, name, project=None): super().__init__(f"Entity {name} does not exist") +class FeatureServiceNotFoundException(FeastObjectNotFoundException): + def __init__(self, name, project=None): + if project: + super().__init__( + f"Feature service {name} does not exist in project {project}" + ) + else: + super().__init__(f"Feature service {name} does not exist") + + class FeatureViewNotFoundException(FeastObjectNotFoundException): def __init__(self, name, project=None): if project: diff --git a/sdk/python/feast/feature.py b/sdk/python/feast/feature.py index 1eec6d6bd84..c7bcac98ed0 100644 --- a/sdk/python/feast/feature.py +++ b/sdk/python/feast/feature.py @@ -52,6 +52,14 @@ def __eq__(self, other): def __lt__(self, other): return self.name < other.name + def __repr__(self): + # return string representation of the reference + return self.name + + def __str__(self): + # readable string of the reference + return f"Feature<{self.__repr__()}>" + @property def name(self): """ diff --git a/sdk/python/feast/feature_service.py b/sdk/python/feast/feature_service.py index aaff436c283..52df88cdf58 100644 --- a/sdk/python/feast/feature_service.py +++ b/sdk/python/feast/feature_service.py @@ -1,5 +1,7 @@ from datetime import datetime -from typing import List, Optional, Union +from typing import Dict, List, Optional, Union + +from google.protobuf.json_format import MessageToJson from feast.feature_table import FeatureTable from feast.feature_view import FeatureView @@ -14,8 +16,14 @@ class FeatureService: + """ + A feature service is a logical grouping of features for retrieval (training or serving). + The features grouped by a feature service may come from any number of feature views. + """ + name: str features: List[FeatureViewProjection] + tags: Dict[str, str] created_timestamp: Optional[datetime] = None last_updated_timestamp: Optional[datetime] = None @@ -23,7 +31,17 @@ def __init__( self, name: str, features: List[Union[FeatureTable, FeatureView, FeatureViewProjection]], + tags: Optional[Dict[str, str]] = None, ): + """ + Create a new Feature Service object. + + Args: + name: A unique name for the Feature Service. + features: A list of Features that are grouped as part of this FeatureService. + The list may contain Feature Views, Feature Tables, or a subset of either. + tags: A dictionary of key-value pairs used for organizing Feature Services. + """ self.name = name self.features = [] for feature in features: @@ -33,9 +51,32 @@ def __init__( self.features.append(feature) else: raise ValueError(f"Unexpected type: {type(feature)}") + self.tags = tags or {} + self.created_timestamp = None + self.last_updated_timestamp = None + + def __repr__(self): + items = (f"{k} = {v}" for k, v in self.__dict__.items()) + return f"<{self.__class__.__name__}({', '.join(items)})>" + + def __str__(self): + return str(MessageToJson(self.to_proto())) + + def __hash__(self): + return hash(self.name) def __eq__(self, other): - pass + if not isinstance(other, FeatureService): + raise TypeError( + "Comparisons should only involve FeatureService class objects." + ) + if self.tags != other.tags or self.name != other.name: + return False + + if sorted(self.features) != sorted(other.features): + return False + + return True @staticmethod def from_proto(feature_service_proto: FeatureServiceProto): @@ -45,6 +86,7 @@ def from_proto(feature_service_proto: FeatureServiceProto): FeatureViewProjection.from_proto(fp) for fp in feature_service_proto.spec.features ], + tags=dict(feature_service_proto.spec.tags), ) if feature_service_proto.meta.HasField("created_timestamp"): @@ -58,7 +100,7 @@ def from_proto(feature_service_proto: FeatureServiceProto): return fs - def to_proto(self): + def to_proto(self) -> FeatureServiceProto: meta = FeatureServiceMeta() if self.created_timestamp: meta.created_timestamp.FromDatetime(self.created_timestamp) @@ -77,6 +119,9 @@ def to_proto(self): spec.features.append(feature_ref.to_proto()) + if self.tags: + spec.tags.update(self.tags) + feature_service_proto = FeatureServiceProto(spec=spec, meta=meta) return feature_service_proto diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 80a071c97cd..cd41f783b67 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -15,7 +15,7 @@ from collections import Counter, OrderedDict, defaultdict from datetime import datetime, timedelta from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any, Dict, Iterable, List, Optional, Tuple, Union import pandas as pd from colorama import Fore, Style @@ -24,6 +24,7 @@ from feast import utils from feast.entity import Entity from feast.errors import FeatureNameCollisionError, FeatureViewNotFoundException +from feast.feature_service import FeatureService from feast.feature_table import FeatureTable from feast.feature_view import FeatureView from feast.inference import ( @@ -130,6 +131,17 @@ def list_entities(self, allow_cache: bool = False) -> List[Entity]: return self._registry.list_entities(self.project, allow_cache=allow_cache) + @log_exceptions_and_usage + def list_feature_services(self) -> List[FeatureService]: + """ + Retrieve a list of feature services from the registry + + Returns: + List of feature services + """ + + return self._registry.list_feature_services(self.project) + @log_exceptions_and_usage def list_feature_views(self) -> List[FeatureView]: """ @@ -156,6 +168,21 @@ def get_entity(self, name: str) -> Entity: return self._registry.get_entity(name, self.project) + @log_exceptions_and_usage + def get_feature_service(self, name: str) -> FeatureService: + """ + Retrieves a feature service. + + Args: + name: Name of FeatureService + + Returns: + Returns either the specified FeatureService, or raises an exception if + none is found + """ + + return self._registry.get_feature_service(name, self.project) + @log_exceptions_and_usage def get_feature_view(self, name: str) -> FeatureView: """ @@ -182,9 +209,34 @@ def delete_feature_view(self, name: str): return self._registry.delete_feature_view(name, self.project) + def _get_features( + self, + features: Union[List[str], FeatureService], + feature_refs: Optional[List[str]], + ) -> List[str]: + _features = features or feature_refs + if not _features: + raise ValueError("No features specified for retrieval") + + _feature_refs: List[str] + if isinstance(_features, FeatureService): + # Get the latest value of the feature service, in case the object passed in has been updated underneath us. + _feature_refs = _get_feature_refs_from_feature_services( + self.get_feature_service(_features.name) + ) + else: + _feature_refs = _features + return _feature_refs + @log_exceptions_and_usage def apply( - self, objects: Union[Entity, FeatureView, List[Union[FeatureView, Entity]]] + self, + objects: Union[ + Entity, + FeatureView, + FeatureService, + List[Union[FeatureView, Entity, FeatureService]], + ], ): """Register objects to metadata store and update related infrastructure. @@ -194,8 +246,7 @@ def apply( operations are idempotent, meaning they can safely be rerun. Args: - objects (List[Union[FeatureView, Entity]]): A list of FeatureView or Entity objects that should be - registered + objects: A single object, or a list of objects that should be registered with the Feature Store. Examples: Register a single Entity and FeatureView. @@ -218,12 +269,14 @@ def apply( # TODO: Add locking - if isinstance(objects, Entity) or isinstance(objects, FeatureView): + if not isinstance(objects, Iterable): objects = [objects] + assert isinstance(objects, list) views_to_update = [ob for ob in objects if isinstance(ob, FeatureView)] entities_to_update = [ob for ob in objects if isinstance(ob, Entity)] + services_to_update = [ob for ob in objects if isinstance(ob, FeatureService)] # Make inferences update_entities_with_inferred_types_from_feature_views( @@ -237,13 +290,17 @@ def apply( for view in views_to_update: view.infer_features_from_input_source(self.config) - if len(views_to_update) + len(entities_to_update) != len(objects): + if len(views_to_update) + len(entities_to_update) + len( + services_to_update + ) != len(objects): raise ValueError("Unknown object type provided as part of apply() call") for view in views_to_update: self._registry.apply_feature_view(view, project=self.project, commit=False) for ent in entities_to_update: self._registry.apply_entity(ent, project=self.project, commit=False) + for feature_service in services_to_update: + self._registry.apply_feature_service(feature_service, project=self.project) self._registry.commit() self._get_provider().update_infra( @@ -273,7 +330,8 @@ def teardown(self): def get_historical_features( self, entity_df: Union[pd.DataFrame, str], - feature_refs: List[str], + features: Union[List[str], FeatureService], + feature_refs: Optional[List[str]] = None, full_feature_names: bool = False, ) -> RetrievalJob: """Enrich an entity dataframe with historical feature values for either training or batch scoring. @@ -294,8 +352,9 @@ def get_historical_features( columns (e.g., customer_id, driver_id) on which features need to be joined, as well as a event_timestamp column used to ensure point-in-time correctness. Either a Pandas DataFrame can be provided or a string SQL query. The query must be of a format supported by the configured offline store (e.g., BigQuery) - feature_refs: A list of features that should be retrieved from the offline store. Feature references are of - the format "feature_view:feature", e.g., "customer_fv:daily_transactions". + features: A list of features, that should be retrieved from the offline store. + Either a list of string feature references can be provided or a FeatureService object. + Feature references are of the format "feature_view:feature", e.g., "customer_fv:daily_transactions". full_feature_names: A boolean that provides the option to add the feature view prefixes to the feature names, changing them from the format "feature" to "feature_view__feature" (e.g., "daily_transactions" changes to "customer_fv__daily_transactions"). By default, this value is set to False. @@ -311,24 +370,29 @@ def get_historical_features( >>> fs = FeatureStore(config=RepoConfig(provider="gcp")) >>> retrieval_job = fs.get_historical_features( >>> entity_df="SELECT event_timestamp, order_id, customer_id from gcp_project.my_ds.customer_orders", - >>> feature_refs=["customer:age", "customer:avg_orders_1d", "customer:avg_orders_7d"], - >>> ) + >>> features=["customer:age", "customer:avg_orders_1d", "customer:avg_orders_7d"] + >>> ) >>> feature_data = retrieval_job.to_df() >>> model.fit(feature_data) # insert your modeling framework here. """ - all_feature_views = self._registry.list_feature_views(project=self.project) - _validate_feature_refs(feature_refs, full_feature_names) + _feature_refs = self._get_features(features, feature_refs) + + print(f"_feature_refs: {_feature_refs}") + + all_feature_views = self._registry.list_feature_views(project=self.project) feature_views = list( - view for view, _ in _group_feature_refs(feature_refs, all_feature_views) + view for view, _ in _group_feature_refs(_feature_refs, all_feature_views) ) + _validate_feature_refs(_feature_refs, full_feature_names) + provider = self._get_provider() job = provider.get_historical_features( self.config, feature_views, - feature_refs, + _feature_refs, entity_df, self._registry, self.project, @@ -500,8 +564,9 @@ def tqdm_builder(length): @log_exceptions_and_usage def get_online_features( self, - feature_refs: List[str], + features: Union[List[str], FeatureService], entity_rows: List[Dict[str, Any]], + feature_refs: Optional[List[str]] = None, full_feature_names: bool = False, ) -> OnlineResponse: """ @@ -516,7 +581,7 @@ def get_online_features( infinity (cache forever). Args: - feature_refs: List of feature references that will be returned for each entity. + features: List of feature references that will be returned for each entity. Each feature reference should have the following format: "feature_table:feature" where "feature_table" & "feature" refer to the feature and feature table names respectively. @@ -538,6 +603,8 @@ def get_online_features( {'sales:daily_transactions': [1.1,1.2], 'sales:customer_id': [0,1]} """ + _feature_refs = self._get_features(features, feature_refs) + provider = self._get_provider() entities = self.list_entities(allow_cache=True) entity_name_to_join_key_map = {} @@ -570,8 +637,8 @@ def get_online_features( project=self.project, allow_cache=True ) - _validate_feature_refs(feature_refs, full_feature_names) - grouped_refs = _group_feature_refs(feature_refs, all_feature_views) + _validate_feature_refs(_feature_refs, full_feature_names) + grouped_refs = _group_feature_refs(_feature_refs, all_feature_views) for table, requested_features in grouped_refs: entity_keys = _get_table_entity_keys( table, union_of_entity_keys, entity_name_to_join_key_map @@ -616,7 +683,7 @@ def get_online_features( def _entity_row_to_key(row: GetOnlineFeaturesRequestV2.EntityRow) -> EntityKeyProto: names, values = zip(*row.fields.items()) - return EntityKeyProto(join_keys=names, entity_values=values) # type: ignore + return EntityKeyProto(join_keys=names, entity_values=values) def _entity_row_to_field_values( @@ -655,7 +722,7 @@ def _validate_feature_refs(feature_refs: List[str], full_feature_names: bool = F def _group_feature_refs( - feature_refs: List[str], all_feature_views: List[FeatureView] + features: Union[List[str], FeatureService], all_feature_views: List[FeatureView] ) -> List[Tuple[FeatureView, List[str]]]: """ Get list of feature views and corresponding feature names based on feature references""" @@ -665,12 +732,18 @@ def _group_feature_refs( # view name to feature names views_features = defaultdict(list) - for ref in feature_refs: - view_name, feat_name = ref.split(":") - - if view_name not in view_index: - raise FeatureViewNotFoundException(view_name) - views_features[view_name].append(feat_name) + if isinstance(features, list) and isinstance(features[0], str): + for ref in features: + view_name, feat_name = ref.split(":") + if view_name not in view_index: + raise FeatureViewNotFoundException(view_name) + views_features[view_name].append(feat_name) + elif isinstance(features, FeatureService): + for feature_projection in features.features: + projected_features = feature_projection.features + views_features[feature_projection.name].extend( + [f.name for f in projected_features] + ) result = [] for view_name, feature_names in views_features.items(): @@ -678,6 +751,17 @@ def _group_feature_refs( return result +def _get_feature_refs_from_feature_services( + feature_service: FeatureService, +) -> List[str]: + feature_refs = [] + for projection in feature_service.features: + feature_refs.extend( + [f"{projection.name}:{f.name}" for f in projection.features] + ) + return feature_refs + + def _get_table_entity_keys( table: FeatureView, entity_keys: List[EntityKeyProto], join_key_map: Dict[str, str], ) -> List[EntityKeyProto]: diff --git a/sdk/python/feast/registry.py b/sdk/python/feast/registry.py index 324901057ab..7f97589bce1 100644 --- a/sdk/python/feast/registry.py +++ b/sdk/python/feast/registry.py @@ -24,11 +24,13 @@ from feast.entity import Entity from feast.errors import ( EntityNotFoundException, + FeatureServiceNotFoundException, FeatureTableNotFoundException, FeatureViewNotFoundException, S3RegistryBucketForbiddenAccess, S3RegistryBucketNotExist, ) +from feast.feature_service import FeatureService from feast.feature_table import FeatureTable from feast.feature_view import FeatureView from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto @@ -70,7 +72,8 @@ def __init__(self, registry_path: str, repo_path: Path, cache_ttl: timedelta): ) else: raise Exception( - f"Registry path {registry_path} has unsupported scheme {uri.scheme}. Supported schemes are file and gs." + f"Registry path {registry_path} has unsupported scheme {uri.scheme}. " + f"Supported schemes are file and gs." ) self.cached_registry_proto_ttl = cache_ttl @@ -127,6 +130,79 @@ def list_entities(self, project: str, allow_cache: bool = False) -> List[Entity] entities.append(Entity.from_proto(entity_proto)) return entities + def apply_feature_service( + self, feature_service: FeatureService, project: str, commit: bool = True + ): + """ + Registers a single feature service with Feast + + Args: + feature_service: A feature service that will be registered + project: Feast project that this entity belongs to + """ + feature_service_proto = feature_service.to_proto() + feature_service_proto.spec.project = project + + registry = self._prepare_registry_for_changes() + + for idx, existing_feature_service_proto in enumerate(registry.feature_services): + if ( + existing_feature_service_proto.spec.name + == feature_service_proto.spec.name + and existing_feature_service_proto.spec.project == project + ): + del registry.feature_services[idx] + registry.feature_services.append(feature_service_proto) + if commit: + self.commit() + + def list_feature_services( + self, project: str, allow_cache: bool = False + ) -> List[FeatureService]: + """ + Retrieve a list of feature services from the registry + + Args: + allow_cache: Whether to allow returning entities from a cached registry + project: Filter entities based on project name + + Returns: + List of feature services + """ + + registry = self._get_registry_proto(allow_cache=allow_cache) + feature_services = [] + for feature_service_proto in registry.feature_services: + if feature_service_proto.spec.project == project: + feature_services.append( + FeatureService.from_proto(feature_service_proto) + ) + return feature_services + + def get_feature_service( + self, name: str, project: str, allow_cache: bool = False + ) -> FeatureService: + """ + Retrieves a feature service. + + Args: + name: Name of feature service + project: Feast project that this feature service belongs to + + Returns: + Returns either the specified feature service, or raises an exception if + none is found + """ + registry = self._get_registry_proto(allow_cache=allow_cache) + + for feature_service_proto in registry.feature_services: + if ( + feature_service_proto.spec.project == project + and feature_service_proto.spec.name == name + ): + return FeatureService.from_proto(feature_service_proto) + raise FeatureServiceNotFoundException(name, project=project) + def get_entity(self, name: str, project: str, allow_cache: bool = False) -> Entity: """ Retrieves an entity. @@ -333,6 +409,24 @@ def get_feature_view(self, name: str, project: str) -> FeatureView: return FeatureView.from_proto(feature_view_proto) raise FeatureViewNotFoundException(name, project) + def delete_feature_service(self, name: str, project: str): + """ + Deletes a feature service or raises an exception if not found. + + Args: + name: Name of feature service + project: Feast project that this feature service belongs to + """ + registry_proto = self._get_registry_proto() + for idx, feature_service_proto in enumerate(registry_proto.feature_services): + if ( + feature_service_proto.spec.name == name + and feature_service_proto.spec.project == project + ): + del registry_proto.feature_services[idx] + return feature_service_proto + raise FeatureServiceNotFoundException(name, project) + def delete_feature_table(self, name: str, project: str, commit: bool = True): """ Deletes a feature table or raises an exception if not found. diff --git a/sdk/python/feast/repo_operations.py b/sdk/python/feast/repo_operations.py index 3474424119b..683ab3bc8d8 100644 --- a/sdk/python/feast/repo_operations.py +++ b/sdk/python/feast/repo_operations.py @@ -12,6 +12,7 @@ from click.exceptions import BadParameter from feast import Entity, FeatureTable +from feast.feature_service import FeatureService from feast.feature_store import FeatureStore from feast.feature_view import FeatureView from feast.inference import ( @@ -37,6 +38,7 @@ class ParsedRepo(NamedTuple): feature_tables: List[FeatureTable] feature_views: List[FeatureView] entities: List[Entity] + feature_services: List[FeatureService] def read_feastignore(repo_root: Path) -> List[str]: @@ -94,7 +96,9 @@ def get_repo_files(repo_root: Path) -> List[Path]: def parse_repo(repo_root: Path) -> ParsedRepo: """ Collect feature table definitions from feature repo """ - res = ParsedRepo(feature_tables=[], entities=[], feature_views=[]) + res = ParsedRepo( + feature_tables=[], entities=[], feature_views=[], feature_services=[] + ) for repo_file in get_repo_files(repo_root): module_path = py_path_to_module(repo_file, repo_root) @@ -108,9 +112,35 @@ def parse_repo(repo_root: Path) -> ParsedRepo: res.feature_views.append(obj) elif isinstance(obj, Entity): res.entities.append(obj) + elif isinstance(obj, FeatureService): + res.feature_services.append(obj) return res +def apply_feature_services(registry: Registry, project: str, repo: ParsedRepo): + from colorama import Fore, Style + + # Determine which feature services should be deleted. + existing_feature_services = registry.list_feature_services(project) + for feature_service in repo.feature_services: + if feature_service in existing_feature_services: + existing_feature_services.remove(feature_service) + + # The remaining features services in the list should be deleted. + for feature_service_to_delete in existing_feature_services: + registry.delete_feature_service(feature_service_to_delete.name, project) + click.echo( + f"Deleted feature service {Style.BRIGHT + Fore.GREEN}{feature_service_to_delete.name}{Style.RESET_ALL} " + f"from registry" + ) + + for feature_service in repo.feature_services: + registry.apply_feature_service(feature_service, project=project) + click.echo( + f"Registered feature service {Style.BRIGHT + Fore.GREEN}{feature_service.name}{Style.RESET_ALL}" + ) + + @log_exceptions_and_usage def apply_total(repo_config: RepoConfig, repo_path: Path, skip_source_validation: bool): from colorama import Fore, Style @@ -192,7 +222,7 @@ def apply_total(repo_config: RepoConfig, repo_path: Path, skip_source_validation f"Deleted feature view {Style.BRIGHT + Fore.GREEN}{registry_view.name}{Style.RESET_ALL} from registry" ) - # Create views that should + # Create views that should exist for view in repo.feature_views: registry.apply_feature_view(view, project, commit=False) click.echo( @@ -200,6 +230,8 @@ def apply_total(repo_config: RepoConfig, repo_path: Path, skip_source_validation ) registry.commit() + apply_feature_services(registry, project, repo) + infra_provider = get_provider(repo_config, repo_path) all_to_delete: List[Union[FeatureTable, FeatureView]] = [] diff --git a/sdk/python/feast/templates/aws/test.py b/sdk/python/feast/templates/aws/test.py index cc2cf7e984a..f2dea7212ec 100644 --- a/sdk/python/feast/templates/aws/test.py +++ b/sdk/python/feast/templates/aws/test.py @@ -27,8 +27,7 @@ def main(): # Retrieve features from the online store (DynamoDB) online_features = fs.get_online_features( - feature_refs=feature_refs, - entity_rows=[{"driver_id": 1001}, {"driver_id": 1002}], + features=feature_refs, entity_rows=[{"driver_id": 1001}, {"driver_id": 1002}], ).to_dict() print(pd.DataFrame.from_dict(online_features)) diff --git a/sdk/python/feast/templates/gcp/test.py b/sdk/python/feast/templates/gcp/test.py index aee89500777..538334044bf 100644 --- a/sdk/python/feast/templates/gcp/test.py +++ b/sdk/python/feast/templates/gcp/test.py @@ -18,7 +18,7 @@ def main(): fs.apply([driver, driver_stats_fv]) # Select features - feature_refs = ["driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate"] + features = ["driver_hourly_stats:conv_rate", "driver_hourly_stats:acc_rate"] # Create an entity dataframe. This is the dataframe that will be enriched with historical features entity_df = pd.DataFrame( @@ -39,7 +39,7 @@ def main(): # Retrieve historical features by joining the entity dataframe to the BigQuery table source training_df = fs.get_historical_features( - feature_refs=feature_refs, entity_df=entity_df + features=features, entity_df=entity_df ).to_df() print() @@ -54,8 +54,7 @@ def main(): # Retrieve features from the online store (Firestore) online_features = fs.get_online_features( - feature_refs=feature_refs, - entity_rows=[{"driver_id": 1001}, {"driver_id": 1002}], + features=features, entity_rows=[{"driver_id": 1001}, {"driver_id": 1002}], ).to_dict() print() diff --git a/sdk/python/tests/example_repos/example_feature_repo_1.py b/sdk/python/tests/example_repos/example_feature_repo_1.py index 808e8ed54ed..e42e557bfe3 100644 --- a/sdk/python/tests/example_repos/example_feature_repo_1.py +++ b/sdk/python/tests/example_repos/example_feature_repo_1.py @@ -1,6 +1,13 @@ from datetime import timedelta -from feast import BigQuerySource, Entity, Feature, FeatureView, ValueType +from feast import ( + BigQuerySource, + Entity, + Feature, + FeatureService, + FeatureView, + ValueType, +) driver_locations_source = BigQuerySource( table_ref="feast-oss.public.drivers", @@ -65,3 +72,10 @@ input=customer_driver_combined_source, tags={}, ) + + +all_drivers_feature_service = FeatureService( + name="driver_locations_service", + features=[driver_locations], + tags={"release": "production"}, +) diff --git a/sdk/python/tests/integration/materialization/test_offline_online_store_consistency.py b/sdk/python/tests/integration/materialization/test_offline_online_store_consistency.py index dd8289710d5..a8efdfcfb55 100644 --- a/sdk/python/tests/integration/materialization/test_offline_online_store_consistency.py +++ b/sdk/python/tests/integration/materialization/test_offline_online_store_consistency.py @@ -349,7 +349,7 @@ def check_offline_and_online_features( entity_df=pd.DataFrame.from_dict( {"driver_id": [driver_id], "event_timestamp": [event_timestamp]} ), - feature_refs=[f"{fv.name}:value"], + features=[f"{fv.name}:value"], full_feature_names=full_feature_names, ).to_df() diff --git a/sdk/python/tests/integration/offline_store/test_historical_retrieval.py b/sdk/python/tests/integration/offline_store/test_historical_retrieval.py index d2ea0798a24..af043359e4b 100644 --- a/sdk/python/tests/integration/offline_store/test_historical_retrieval.py +++ b/sdk/python/tests/integration/offline_store/test_historical_retrieval.py @@ -14,7 +14,7 @@ from pytz import utc import feast.driver_test_data as driver_data -from feast import BigQuerySource, FileSource, RepoConfig, errors, utils +from feast import BigQuerySource, FeatureService, FileSource, RepoConfig, errors, utils from feast.entity import Entity from feast.errors import FeatureNameCollisionError from feast.feature import Feature @@ -95,6 +95,10 @@ def stage_customer_daily_profile_parquet_source(directory, df): ) +def feature_service(name: str, views) -> FeatureService: + return FeatureService(name, views) + + def stage_customer_daily_profile_bigquery_source(df, table_id): client = bigquery.Client() job_config = bigquery.LoadJobConfig() @@ -292,6 +296,18 @@ def test_historical_features_from_parquet_sources( temp_dir, customer_df ) customer_fv = create_customer_daily_profile_feature_view(customer_source) + + customer_fs = feature_service( + "customer_feature_service", + [ + customer_fv[ + ["current_balance", "avg_passenger_count", "lifetime_trip_count"] + ], + driver_fv[["conv_rate", "avg_daily_trips"]], + ], + ) + print(f"Customer fs features: {customer_fs.features}") + driver = Entity(name="driver", join_key="driver_id", value_type=ValueType.INT64) customer = Entity(name="customer_id", value_type=ValueType.INT64) @@ -306,11 +322,11 @@ def test_historical_features_from_parquet_sources( ) ) - store.apply([driver, customer, driver_fv, customer_fv]) + store.apply([driver, customer, driver_fv, customer_fv, customer_fs]) job = store.get_historical_features( entity_df=orders_df, - feature_refs=[ + features=[ "driver_stats:conv_rate", "driver_stats:avg_daily_trips", "customer_profile:current_balance", @@ -335,15 +351,35 @@ def test_historical_features_from_parquet_sources( event_timestamp, full_feature_names=full_feature_names, ) + + expected_df.sort_values( + by=[event_timestamp, "order_id", "driver_id", "customer_id"] + ).reset_index(drop=True) + expected_df = expected_df.reindex(sorted(expected_df.columns), axis=1) + + actual_df = actual_df.sort_values( + by=[event_timestamp, "order_id", "driver_id", "customer_id"] + ).reset_index(drop=True) + actual_df = actual_df.reindex(sorted(actual_df.columns), axis=1) + assert_frame_equal( - expected_df.sort_values( - by=[event_timestamp, "order_id", "driver_id", "customer_id"] - ).reset_index(drop=True), - actual_df.sort_values( - by=[event_timestamp, "order_id", "driver_id", "customer_id"] - ).reset_index(drop=True), + expected_df, actual_df, + ) + + feature_service_job = store.get_historical_features( + entity_df=orders_df, + features=customer_fs, + full_feature_names=full_feature_names, + ) + feature_service_df = feature_service_job.to_df() + feature_service_df = feature_service_df.sort_values( + by=[event_timestamp, "order_id", "driver_id", "customer_id"] + ).reset_index(drop=True) + feature_service_df = feature_service_df.reindex( + sorted(feature_service_df.columns), axis=1 ) + assert_frame_equal(expected_df, feature_service_df) store.teardown() @@ -473,7 +509,7 @@ def test_historical_features_from_bigquery_sources( job_from_sql = store.get_historical_features( entity_df=entity_df_query, - feature_refs=[ + features=[ "driver_stats:conv_rate", "driver_stats:avg_daily_trips", "customer_profile:current_balance", @@ -526,7 +562,7 @@ def test_historical_features_from_bigquery_sources( errors.FeastEntityDFMissingColumnsError ).when_called_with( entity_df=entity_df_query_with_invalid_join_key, - feature_refs=[ + features=[ "driver_stats:conv_rate", "driver_stats:avg_daily_trips", "customer_profile:current_balance", @@ -537,7 +573,7 @@ def test_historical_features_from_bigquery_sources( job_from_df = store.get_historical_features( entity_df=orders_df, - feature_refs=[ + features=[ "driver_stats:conv_rate", "driver_stats:avg_daily_trips", "customer_profile:current_balance", @@ -555,7 +591,7 @@ def test_historical_features_from_bigquery_sources( errors.FeastEntityDFMissingColumnsError ).when_called_with( entity_df=orders_df_with_invalid_join_key, - feature_refs=[ + features=[ "driver_stats:conv_rate", "driver_stats:avg_daily_trips", "customer_profile:current_balance", diff --git a/sdk/python/tests/integration/online_store/test_e2e_local.py b/sdk/python/tests/integration/online_store/test_e2e_local.py index 58e32ad26cc..459ba41614a 100644 --- a/sdk/python/tests/integration/online_store/test_e2e_local.py +++ b/sdk/python/tests/integration/online_store/test_e2e_local.py @@ -27,7 +27,7 @@ def _assert_online_features( """Assert that features in online store are up to date with `max_date` date.""" # Read features back result = store.get_online_features( - feature_refs=[ + features=[ "driver_hourly_stats:conv_rate", "driver_hourly_stats:avg_daily_trips", ], diff --git a/sdk/python/tests/integration/online_store/test_feature_service_read.py b/sdk/python/tests/integration/online_store/test_feature_service_read.py new file mode 100644 index 00000000000..33c318b9ed9 --- /dev/null +++ b/sdk/python/tests/integration/online_store/test_feature_service_read.py @@ -0,0 +1,22 @@ +import pytest + +from tests.utils.cli_utils import CliRunner, get_example_repo +from tests.utils.online_read_write_test import basic_rw_test + + +@pytest.mark.integration +def test_feature_service_read() -> None: + """ + Read feature values from the FeatureStore using a FeatureService. + """ + + runner = CliRunner() + with runner.local_repo( + get_example_repo("example_feature_repo_1.py"), "bigquery" + ) as store: + + basic_rw_test( + store, + view_name="driver_locations", + feature_service_name="driver_locations_service", + ) diff --git a/sdk/python/tests/integration/online_store/test_online_retrieval.py b/sdk/python/tests/integration/online_store/test_online_retrieval.py index c981552fdf8..b94f6f1772c 100644 --- a/sdk/python/tests/integration/online_store/test_online_retrieval.py +++ b/sdk/python/tests/integration/online_store/test_online_retrieval.py @@ -94,7 +94,7 @@ def test_online() -> None: # Retrieve two features using two keys, one valid one non-existing result = store.get_online_features( - feature_refs=[ + features=[ "driver_locations:lon", "customer_profile:avg_orders_day", "customer_profile:name", @@ -116,7 +116,7 @@ def test_online() -> None: # Ensure features are still in result when keys not found result = store.get_online_features( - feature_refs=["customer_driver_combined:trips"], + features=["customer_driver_combined:trips"], entity_rows=[{"driver": 0, "customer": 0}], full_feature_names=False, ).to_dict() @@ -126,7 +126,7 @@ def test_online() -> None: # invalid table reference with pytest.raises(FeatureViewNotFoundException): store.get_online_features( - feature_refs=["driver_locations_bad:lon"], + features=["driver_locations_bad:lon"], entity_rows=[{"driver": 1}], full_feature_names=False, ) @@ -146,7 +146,7 @@ def test_online() -> None: # Should download the registry and cache it permanently (or until manually refreshed) result = fs_fast_ttl.get_online_features( - feature_refs=[ + features=[ "driver_locations:lon", "customer_profile:avg_orders_day", "customer_profile:name", @@ -167,7 +167,7 @@ def test_online() -> None: # Will try to reload registry because it has expired (it will fail because we deleted the actual registry file) with pytest.raises(FileNotFoundError): fs_fast_ttl.get_online_features( - feature_refs=[ + features=[ "driver_locations:lon", "customer_profile:avg_orders_day", "customer_profile:name", @@ -182,7 +182,7 @@ def test_online() -> None: # Test if registry is actually reloaded and whether results return result = fs_fast_ttl.get_online_features( - feature_refs=[ + features=[ "driver_locations:lon", "customer_profile:avg_orders_day", "customer_profile:name", @@ -208,7 +208,7 @@ def test_online() -> None: # Should return results (and fill the registry cache) result = fs_infinite_ttl.get_online_features( - feature_refs=[ + features=[ "driver_locations:lon", "customer_profile:avg_orders_day", "customer_profile:name", @@ -228,7 +228,7 @@ def test_online() -> None: # TTL is infinite so this method should use registry cache result = fs_infinite_ttl.get_online_features( - feature_refs=[ + features=[ "driver_locations:lon", "customer_profile:avg_orders_day", "customer_profile:name", @@ -359,7 +359,7 @@ def test_online_to_df(): # Get online features in dataframe result_df = store.get_online_features( - feature_refs=[ + features=[ "driver_locations:lon", "driver_locations:lat", "customer_profile:avg_orders_day", diff --git a/sdk/python/tests/integration/registration/test_cli_local.py b/sdk/python/tests/integration/registration/test_cli_local.py index 5ec7ea2871a..d59d2ebf51f 100644 --- a/sdk/python/tests/integration/registration/test_cli_local.py +++ b/sdk/python/tests/integration/registration/test_cli_local.py @@ -50,6 +50,8 @@ def test_workflow() -> None: assertpy.assert_that(result.returncode).is_equal_to(0) result = runner.run(["feature-views", "list"], cwd=repo_path) assertpy.assert_that(result.returncode).is_equal_to(0) + result = runner.run(["feature-services", "list"], cwd=repo_path) + assertpy.assert_that(result.returncode).is_equal_to(0) # entity & feature view describe commands should succeed when objects exist result = runner.run(["entities", "describe", "driver"], cwd=repo_path) @@ -58,12 +60,18 @@ def test_workflow() -> None: ["feature-views", "describe", "driver_locations"], cwd=repo_path ) assertpy.assert_that(result.returncode).is_equal_to(0) + result = runner.run( + ["feature-services", "describe", "driver_locations_service"], cwd=repo_path + ) + assertpy.assert_that(result.returncode).is_equal_to(0) # entity & feature view describe commands should fail when objects don't exist result = runner.run(["entities", "describe", "foo"], cwd=repo_path) assertpy.assert_that(result.returncode).is_equal_to(1) result = runner.run(["feature-views", "describe", "foo"], cwd=repo_path) assertpy.assert_that(result.returncode).is_equal_to(1) + result = runner.run(["feature-services", "describe", "foo"], cwd=repo_path) + assertpy.assert_that(result.returncode).is_equal_to(1) # Doing another apply should be a no op, and should not cause errors result = runner.run(["apply"], cwd=repo_path) diff --git a/sdk/python/tests/integration/registration/test_feature_service_apply.py b/sdk/python/tests/integration/registration/test_feature_service_apply.py new file mode 100644 index 00000000000..7824f6333ea --- /dev/null +++ b/sdk/python/tests/integration/registration/test_feature_service_apply.py @@ -0,0 +1,29 @@ +import pytest + +from feast import FeatureService +from tests.utils.cli_utils import CliRunner, get_example_repo + + +@pytest.mark.integration +def test_read_pre_applied() -> None: + """ + Read feature values from the FeatureStore using a FeatureService. + """ + runner = CliRunner() + with runner.local_repo( + get_example_repo("example_feature_repo_1.py"), "bigquery" + ) as store: + + assert len(store.list_feature_services()) == 1 + fs = store.get_feature_service("driver_locations_service") + assert len(fs.tags) == 1 + assert fs.tags["release"] == "production" + + fv = store.get_feature_view("driver_locations") + + fs = FeatureService(name="new_feature_service", features=[fv[["lon"]]]) + + store.apply([fs]) + + assert len(store.list_feature_services()) == 2 + store.get_feature_service("new_feature_service") diff --git a/sdk/python/tests/utils/online_read_write_test.py b/sdk/python/tests/utils/online_read_write_test.py index 99339534f4d..ae623d8c7b5 100644 --- a/sdk/python/tests/utils/online_read_write_test.py +++ b/sdk/python/tests/utils/online_read_write_test.py @@ -1,11 +1,14 @@ from datetime import datetime, timedelta +from typing import Optional from feast.feature_store import FeatureStore from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto -def basic_rw_test(store: FeatureStore, view_name: str) -> None: +def basic_rw_test( + store: FeatureStore, view_name: str, feature_service_name: Optional[str] = None +) -> None: """ This is a provider-independent test suite for reading and writing from the online store, to be used by provider-specific tests. @@ -39,13 +42,23 @@ def _driver_rw_test(event_ts, created_ts, write, expect_read): progress=None, ) - read_rows = provider.online_read( - config=store.config, table=table, entity_keys=[entity_key] - ) - assert len(read_rows) == 1 - _, val = read_rows[0] - assert val["lon"].string_val == expect_lon - assert abs(val["lat"].double_val - expect_lat) < 1e-6 + if feature_service_name: + entity_dict = {"driver": 1} + feature_service = store.get_feature_service(feature_service_name) + features = store.get_online_features( + features=feature_service, entity_rows=[entity_dict] + ).to_dict() + assert len(features["driver"]) == 1 + assert features["lon"][0] == expect_lon + assert abs(features["lat"][0] - expect_lat) < 1e-6 + else: + read_rows = provider.online_read( + config=store.config, table=table, entity_keys=[entity_key] + ) + assert len(read_rows) == 1 + _, val = read_rows[0] + assert val["lon"].string_val == expect_lon + assert abs(val["lat"].double_val - expect_lat) < 1e-6 """ 1. Basic test: write value, read it back """