diff --git a/protos/feast/core/Aggregation.proto b/protos/feast/core/Aggregation.proto new file mode 100644 index 00000000000..d848ce69721 --- /dev/null +++ b/protos/feast/core/Aggregation.proto @@ -0,0 +1,14 @@ +syntax = "proto3"; +package feast.core; + +option go_package = "github.com/feast-dev/feast/go/protos/feast/core"; +option java_outer_classname = "AggregationProto"; +option java_package = "feast.proto.core"; + +import "google/protobuf/duration.proto"; + +message Aggregation { + string column = 1; + string function = 2; + google.protobuf.Duration time_window = 3; +} \ No newline at end of file diff --git a/protos/feast/core/DataFormat.proto b/protos/feast/core/DataFormat.proto index 9fd01e865c2..c453e5e4c83 100644 --- a/protos/feast/core/DataFormat.proto +++ b/protos/feast/core/DataFormat.proto @@ -26,7 +26,7 @@ option java_package = "feast.proto.core"; message FileFormat { // Defines options for the Parquet data format message ParquetFormat {} - + oneof format { ParquetFormat parquet_format = 1; } @@ -40,7 +40,7 @@ message StreamFormat { // Feature data from the obtained stream message string class_path = 1; } - + // Defines options for the avro data format message AvroFormat { // Optional if used in a File DataSource as schema is embedded in avro file. @@ -48,9 +48,14 @@ message StreamFormat { string schema_json = 1; } + message JsonFormat { + string schema_json = 1; + } + // Specifies the data format and format specific options oneof format { AvroFormat avro_format = 1; ProtoFormat proto_format = 2; + JsonFormat json_format = 3; } } diff --git a/protos/feast/core/FeatureService.proto b/protos/feast/core/FeatureService.proto index 2654703cc59..51b9c6c02a2 100644 --- a/protos/feast/core/FeatureService.proto +++ b/protos/feast/core/FeatureService.proto @@ -5,7 +5,6 @@ option go_package = "github.com/feast-dev/feast/go/protos/feast/core"; option java_outer_classname = "FeatureServiceProto"; option java_package = "feast.proto.core"; -import "google/protobuf/duration.proto"; import "google/protobuf/timestamp.proto"; import "feast/core/FeatureViewProjection.proto"; diff --git a/protos/feast/core/Registry.proto b/protos/feast/core/Registry.proto index 2c31101510b..19f17a81589 100644 --- a/protos/feast/core/Registry.proto +++ b/protos/feast/core/Registry.proto @@ -28,12 +28,13 @@ import "feast/core/FeatureView.proto"; import "feast/core/InfraObject.proto"; import "feast/core/OnDemandFeatureView.proto"; import "feast/core/RequestFeatureView.proto"; +import "feast/core/StreamFeatureView.proto"; import "feast/core/DataSource.proto"; import "feast/core/SavedDataset.proto"; import "feast/core/ValidationProfile.proto"; import "google/protobuf/timestamp.proto"; -// Next id: 14 +// Next id: 15 message Registry { repeated Entity entities = 1; repeated FeatureTable feature_tables = 2; @@ -41,6 +42,7 @@ message Registry { repeated DataSource data_sources = 12; repeated OnDemandFeatureView on_demand_feature_views = 8; repeated RequestFeatureView request_feature_views = 9; + repeated StreamFeatureView stream_feature_views = 14; repeated FeatureService feature_services = 7; repeated SavedDataset saved_datasets = 11; repeated ValidationReference validation_references = 13; diff --git a/protos/feast/core/StreamFeatureView.proto b/protos/feast/core/StreamFeatureView.proto new file mode 100644 index 00000000000..3be9dc866af --- /dev/null +++ b/protos/feast/core/StreamFeatureView.proto @@ -0,0 +1,98 @@ +// +// Copyright 2020 The Feast Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + + +syntax = "proto3"; +package feast.core; + +option go_package = "github.com/feast-dev/feast/go/protos/feast/core"; +option java_outer_classname = "StreamFeatureViewProto"; +option java_package = "feast.proto.core"; + + +import "google/protobuf/duration.proto"; +import "google/protobuf/timestamp.proto"; +import "feast/core/OnDemandFeatureView.proto"; +import "feast/core/Feature.proto"; +import "feast/core/DataSource.proto"; +import "feast/core/Aggregation.proto"; + +message StreamFeatureView { + // User-specified specifications of this feature view. + StreamFeatureViewSpec spec = 1; + StreamFeatureViewMeta meta = 2; +} + +// Next available id: 17 +message StreamFeatureViewSpec { + // Name of the feature view. Must be unique. Not updated. + string name = 1; + + // Name of Feast project that this feature view belongs to. + string project = 2; + + // List of names of entities associated with this feature view. + repeated string entities = 3; + + // List of specifications for each feature defined as part of this feature view. + repeated FeatureSpecV2 features = 4; + + // List of specifications for each entity defined as part of this feature view. + repeated FeatureSpecV2 entity_columns = 5; + + // Description of the feature view. + string description = 6; + + // User defined metadata + map tags = 7; + + // Owner of the feature view. + string owner = 8; + + // Features in this feature view can only be retrieved from online serving + // younger than ttl. Ttl is measured as the duration of time between + // the feature's event timestamp and when the feature is retrieved + // Feature values outside ttl will be returned as unset values and indicated to end user + google.protobuf.Duration ttl = 9; + + // Batch/Offline DataSource where this view can retrieve offline feature data. + DataSource batch_source = 10; + // Streaming DataSource from where this view can consume "online" feature data. + DataSource stream_source = 11; + + // Whether these features should be served online or not + bool online = 12; + + // Serialized function that is encoded in the streamfeatureview + UserDefinedFunction user_defined_function = 13; + + // Mode of execution + string mode = 14; + + // Aggregation definitions + repeated Aggregation aggregations = 15; + + // Timestamp field for aggregation + string timestamp_field = 16; +} + +message StreamFeatureViewMeta { + // Time where this Feature View is created + google.protobuf.Timestamp created_timestamp = 1; + + // Time where this Feature View is last updated + google.protobuf.Timestamp last_updated_timestamp = 2; +} diff --git a/protos/feast/core/ValidationProfile.proto b/protos/feast/core/ValidationProfile.proto index b660e449bd2..10027995859 100644 --- a/protos/feast/core/ValidationProfile.proto +++ b/protos/feast/core/ValidationProfile.proto @@ -22,8 +22,6 @@ option java_package = "feast.proto.core"; option java_outer_classname = "ValidationProfile"; option go_package = "github.com/feast-dev/feast/go/protos/feast/core"; -import "feast/core/SavedDataset.proto"; - message GEValidationProfiler { message UserDefinedProfiler { // The python-syntax function body (serialized by dill) diff --git a/sdk/python/feast/aggregation.py b/sdk/python/feast/aggregation.py new file mode 100644 index 00000000000..0a5fe845659 --- /dev/null +++ b/sdk/python/feast/aggregation.py @@ -0,0 +1,69 @@ +from datetime import timedelta +from typing import Optional + +from google.protobuf.duration_pb2 import Duration + +from feast.protos.feast.core.Aggregation_pb2 import Aggregation as AggregationProto + + +class Aggregation: + """ + NOTE: Feast-handled aggregations are not yet supported. This class provides a way to register user-defined aggregations. + + Attributes: + column: str # Column name of the feature we are aggregating. + function: str # Provided built in aggregations sum, max, min, count mean + time_window: timedelta # The time window for this aggregation. + """ + + column: str + function: str + time_window: Optional[timedelta] + + def __init__( + self, + column: Optional[str] = "", + function: Optional[str] = "", + time_window: Optional[timedelta] = None, + ): + self.column = column or "" + self.function = function or "" + self.time_window = time_window + + def to_proto(self) -> AggregationProto: + window_duration = None + if self.time_window is not None: + window_duration = Duration() + window_duration.FromTimedelta(self.time_window) + + return AggregationProto( + column=self.column, function=self.function, time_window=window_duration + ) + + @classmethod + def from_proto(cls, agg_proto: AggregationProto): + time_window = ( + timedelta(days=0) + if agg_proto.time_window.ToNanoseconds() == 0 + else agg_proto.time_window.ToTimedelta() + ) + + aggregation = cls( + column=agg_proto.column, + function=agg_proto.function, + time_window=time_window, + ) + return aggregation + + def __eq__(self, other): + if not isinstance(other, Aggregation): + raise TypeError("Comparisons should only involve Aggregations.") + + if ( + self.column != other.column + or self.function != other.function + or self.time_window != other.time_window + ): + return False + + return True diff --git a/sdk/python/feast/data_format.py b/sdk/python/feast/data_format.py index b6c7bf94e92..8f3b195e3e6 100644 --- a/sdk/python/feast/data_format.py +++ b/sdk/python/feast/data_format.py @@ -89,6 +89,8 @@ def from_proto(cls, proto): fmt = proto.WhichOneof("format") if fmt == "avro_format": return AvroFormat(schema_json=proto.avro_format.schema_json) + if fmt == "json_format": + return JsonFormat(schema_json=proto.json_format.schema_json) if fmt == "proto_format": return ProtoFormat(class_path=proto.proto_format.class_path) raise NotImplementedError(f"StreamFormat is unsupported: {fmt}") @@ -113,6 +115,28 @@ def to_proto(self): return StreamFormatProto(avro_format=proto) +class JsonFormat(StreamFormat): + """ + Defines the Json streaming data format that encodes data in Json format + """ + + def __init__(self, schema_json: str): + """ + Construct a new Json data format. + + For spark, uses pyspark ddl string format. Example shown here: + https://vincent.doba.fr/posts/20211004_spark_data_description_language_for_defining_spark_schema/ + + Args: + schema_json: Json schema definition + """ + self.schema_json = schema_json + + def to_proto(self): + proto = StreamFormatProto.JsonFormat(schema_json=self.schema_json) + return StreamFormatProto(json_format=proto) + + class ProtoFormat(StreamFormat): """ Defines the Protobuf data format diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 4392314bb8d..eecf1882369 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -410,6 +410,9 @@ def __init__( if _message_format is None: raise ValueError("Message format must be specified for Kafka source") + if not timestamp_field and not _event_timestamp_column: + raise ValueError("Timestamp field must be specified for Kafka source") + super().__init__( event_timestamp_column=_event_timestamp_column, created_timestamp_column=created_timestamp_column, diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 2e312240fe1..1279c470ee1 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -42,6 +42,7 @@ from feast import feature_server, flags, flags_helper, ui_server, utils from feast.base_feature_view import BaseFeatureView +from feast.batch_feature_view import BatchFeatureView from feast.data_source import DataSource from feast.diff.infra_diff import InfraDiff, diff_infra_protos from feast.diff.registry_diff import RegistryDiff, apply_diff_to_registry, diff_between @@ -84,6 +85,7 @@ from feast.repo_contents import RepoContents from feast.request_feature_view import RequestFeatureView from feast.saved_dataset import SavedDataset, SavedDatasetStorage, ValidationReference +from feast.stream_feature_view import StreamFeatureView from feast.type_map import ( feast_value_type_to_python_type, python_values_to_proto_values, @@ -273,6 +275,20 @@ def list_on_demand_feature_views( self.project, allow_cache=allow_cache ) + @log_exceptions_and_usage + def list_stream_feature_views( + self, allow_cache: bool = False + ) -> List[StreamFeatureView]: + """ + Retrieves the list of stream feature views from the registry. + + Returns: + A list of stream feature views. + """ + return self._registry.list_stream_feature_views( + self.project, allow_cache=allow_cache + ) + @log_exceptions_and_usage def list_data_sources(self, allow_cache: bool = False) -> List[DataSource]: """ @@ -457,6 +473,7 @@ def _validate_all_feature_views( views_to_update: List[FeatureView], odfvs_to_update: List[OnDemandFeatureView], request_views_to_update: List[RequestFeatureView], + sfvs_to_update: List[StreamFeatureView], ): """Validates all feature views.""" if ( @@ -468,7 +485,12 @@ def _validate_all_feature_views( set_usage_attribute("odfv", bool(odfvs_to_update)) _validate_feature_views( - [*views_to_update, *odfvs_to_update, *request_views_to_update] + [ + *views_to_update, + *odfvs_to_update, + *request_views_to_update, + *sfvs_to_update, + ] ) def _make_inferences( @@ -477,6 +499,7 @@ def _make_inferences( entities_to_update: List[Entity], views_to_update: List[FeatureView], odfvs_to_update: List[OnDemandFeatureView], + sfvs_to_update: List[StreamFeatureView], feature_services_to_update: List[FeatureService], ): """Makes inferences for entities, feature views, odfvs, and feature services.""" @@ -488,16 +511,28 @@ def _make_inferences( [view.batch_source for view in views_to_update], self.config ) + update_data_sources_with_inferred_event_timestamp_col( + [view.batch_source for view in sfvs_to_update], self.config + ) + # New feature views may reference previously applied entities. entities = self._list_entities() update_feature_views_with_inferred_features_and_entities( views_to_update, entities + entities_to_update, self.config ) + # TODO(kevjumba): Update schema inferrence + for sfv in sfvs_to_update: + if not sfv.schema: + raise ValueError( + f"schema inference not yet supported for stream feature views. please define schema for stream feature view: {sfv.name}" + ) for odfv in odfvs_to_update: odfv.infer_features() - fvs_to_update_map = {view.name: view for view in views_to_update} + fvs_to_update_map = { + view.name: view for view in [*views_to_update, *sfvs_to_update] + } for feature_service in feature_services_to_update: feature_service.infer_features(fvs_to_update=fvs_to_update_map) @@ -540,6 +575,7 @@ def _plan( ... data_sources=[driver_hourly_stats], ... feature_views=[driver_hourly_stats_view], ... on_demand_feature_views=list(), + ... stream_feature_views=list(), ... request_feature_views=list(), ... entities=[driver], ... feature_services=list())) # register entity and feature view @@ -549,6 +585,7 @@ def _plan( desired_repo_contents.feature_views, desired_repo_contents.on_demand_feature_views, desired_repo_contents.request_feature_views, + desired_repo_contents.stream_feature_views, ) _validate_data_sources(desired_repo_contents.data_sources) self._make_inferences( @@ -556,6 +593,7 @@ def _plan( desired_repo_contents.entities, desired_repo_contents.feature_views, desired_repo_contents.on_demand_feature_views, + desired_repo_contents.stream_feature_views, desired_repo_contents.feature_services, ) @@ -607,6 +645,7 @@ def apply( FeatureView, OnDemandFeatureView, RequestFeatureView, + StreamFeatureView, FeatureService, ValidationReference, List[FeastObject], @@ -661,7 +700,16 @@ def apply( # Separate all objects into entities, feature services, and different feature view types. entities_to_update = [ob for ob in objects if isinstance(ob, Entity)] - views_to_update = [ob for ob in objects if isinstance(ob, FeatureView)] + views_to_update = [ + ob + for ob in objects + if ( + isinstance(ob, FeatureView) + and not isinstance(ob, StreamFeatureView) + and not isinstance(ob, BatchFeatureView) + ) + ] + sfvs_to_update = [ob for ob in objects if isinstance(ob, StreamFeatureView)] request_views_to_update = [ ob for ob in objects if isinstance(ob, RequestFeatureView) ] @@ -674,7 +722,7 @@ def apply( ob for ob in objects if isinstance(ob, ValidationReference) ] - for fv in views_to_update: + for fv in itertools.chain(views_to_update, sfvs_to_update): data_sources_set_to_update.add(fv.batch_source) if fv.stream_source: data_sources_set_to_update.add(fv.stream_source) @@ -700,13 +748,14 @@ def apply( # Validate all feature views and make inferences. self._validate_all_feature_views( - views_to_update, odfvs_to_update, request_views_to_update + views_to_update, odfvs_to_update, request_views_to_update, sfvs_to_update ) self._make_inferences( data_sources_to_update, entities_to_update, views_to_update, odfvs_to_update, + sfvs_to_update, services_to_update, ) @@ -714,7 +763,7 @@ def apply( for ds in data_sources_to_update: self._registry.apply_data_source(ds, project=self.project, commit=False) for view in itertools.chain( - views_to_update, odfvs_to_update, request_views_to_update + views_to_update, odfvs_to_update, request_views_to_update, sfvs_to_update ): self._registry.apply_feature_view(view, project=self.project, commit=False) for ent in entities_to_update: @@ -742,6 +791,9 @@ def apply( odfvs_to_delete = [ ob for ob in objects_to_delete if isinstance(ob, OnDemandFeatureView) ] + sfvs_to_delete = [ + ob for ob in objects_to_delete if isinstance(ob, StreamFeatureView) + ] services_to_delete = [ ob for ob in objects_to_delete if isinstance(ob, FeatureService) ] @@ -772,6 +824,10 @@ def apply( self._registry.delete_feature_view( odfv.name, project=self.project, commit=False ) + for sfv in sfvs_to_delete: + self._registry.delete_feature_view( + sfv.name, project=self.project, commit=False + ) for service in services_to_delete: self._registry.delete_feature_service( service.name, project=self.project, commit=False @@ -879,6 +935,7 @@ def get_historical_features( all_feature_views, all_request_feature_views, all_on_demand_feature_views, + all_stream_feature_views, ) = self._get_feature_views_to_use(features) if all_request_feature_views: @@ -1153,7 +1210,6 @@ def materialize( Examples: Materialize all features into the online store over the interval from 3 hours ago to 10 minutes ago. - >>> from feast import FeatureStore, RepoConfig >>> from datetime import datetime, timedelta >>> fs = FeatureStore(repo_path="feature_repo") @@ -1223,6 +1279,7 @@ def push( ): """ Push features to a push source. This updates all the feature views that have the push source as stream source. + Args: push_source_name: The name of the push source we want to push data to. df: the data being pushed. @@ -1399,6 +1456,7 @@ def _get_online_features( requested_feature_views, requested_request_feature_views, requested_on_demand_feature_views, + request_stream_feature_views, ) = self._get_feature_views_to_use( features=features, allow_cache=True, hide_dummy_entity=False ) @@ -1935,7 +1993,12 @@ def _get_feature_views_to_use( features: Optional[Union[List[str], FeatureService]], allow_cache=False, hide_dummy_entity: bool = True, - ) -> Tuple[List[FeatureView], List[RequestFeatureView], List[OnDemandFeatureView]]: + ) -> Tuple[ + List[FeatureView], + List[RequestFeatureView], + List[OnDemandFeatureView], + List[StreamFeatureView], + ]: fvs = { fv.name: fv @@ -1956,8 +2019,15 @@ def _get_feature_views_to_use( ) } + sfvs = { + fv.name: fv + for fv in self._registry.list_stream_feature_views( + project=self.project, allow_cache=allow_cache + ) + } + if isinstance(features, FeatureService): - fvs_to_use, request_fvs_to_use, od_fvs_to_use = [], [], [] + fvs_to_use, request_fvs_to_use, od_fvs_to_use, sfvs_to_use = [], [], [], [] for fv_name, projection in [ (projection.name, projection) for projection in features.feature_view_projections @@ -1978,18 +2048,23 @@ def _get_feature_views_to_use( fv = fvs[projection.name].with_projection(copy.copy(projection)) if fv not in fvs_to_use: fvs_to_use.append(fv) + elif fv_name in sfvs: + sfvs_to_use.append( + sfvs[fv_name].with_projection(copy.copy(projection)) + ) else: raise ValueError( f"The provided feature service {features.name} contains a reference to a feature view" f"{fv_name} which doesn't exist. Please make sure that you have created the feature view" f'{fv_name} and that you have registered it by running "apply".' ) - views_to_use = (fvs_to_use, request_fvs_to_use, od_fvs_to_use) + views_to_use = (fvs_to_use, request_fvs_to_use, od_fvs_to_use, sfvs_to_use) else: views_to_use = ( [*fvs.values()], [*request_fvs.values()], [*od_fvs.values()], + [*sfvs.values()], ) return views_to_use diff --git a/sdk/python/feast/inference.py b/sdk/python/feast/inference.py index aed90c4ac83..37f0cb8b05e 100644 --- a/sdk/python/feast/inference.py +++ b/sdk/python/feast/inference.py @@ -1,5 +1,5 @@ import re -from typing import List, Set +from typing import List, Set, Union from feast.data_source import DataSource, PushSource, RequestSource from feast.entity import Entity @@ -11,6 +11,7 @@ from feast.infra.offline_stores.redshift_source import RedshiftSource from feast.infra.offline_stores.snowflake_source import SnowflakeSource from feast.repo_config import RepoConfig +from feast.stream_feature_view import StreamFeatureView from feast.types import String from feast.value_type import ValueType @@ -19,7 +20,6 @@ def update_data_sources_with_inferred_event_timestamp_col( data_sources: List[DataSource], config: RepoConfig ) -> None: ERROR_MSG_PREFIX = "Unable to infer DataSource timestamp_field" - for data_source in data_sources: if isinstance(data_source, RequestSource): continue @@ -88,7 +88,9 @@ def update_data_sources_with_inferred_event_timestamp_col( def update_feature_views_with_inferred_features_and_entities( - fvs: List[FeatureView], entities: List[Entity], config: RepoConfig + fvs: Union[List[FeatureView], List[StreamFeatureView]], + entities: List[Entity], + config: RepoConfig, ) -> None: """ Infers the features and entities associated with each feature view and updates it in place. diff --git a/sdk/python/feast/registry.py b/sdk/python/feast/registry.py index c46eba8a5d6..7f298b19b82 100644 --- a/sdk/python/feast/registry.py +++ b/sdk/python/feast/registry.py @@ -51,6 +51,7 @@ from feast.repo_contents import RepoContents from feast.request_feature_view import RequestFeatureView from feast.saved_dataset import SavedDataset, ValidationReference +from feast.stream_feature_view import StreamFeatureView REGISTRY_SCHEMA_VERSION = "1" @@ -476,7 +477,11 @@ def apply_feature_view( self._check_conflicting_feature_view_names(feature_view) existing_feature_views_of_same_type: RepeatedCompositeFieldContainer - if isinstance(feature_view, FeatureView): + if isinstance(feature_view, StreamFeatureView): + existing_feature_views_of_same_type = ( + self.cached_registry_proto.stream_feature_views + ) + elif isinstance(feature_view, FeatureView): existing_feature_views_of_same_type = ( self.cached_registry_proto.feature_views ) @@ -506,11 +511,32 @@ def apply_feature_view( else: del existing_feature_views_of_same_type[idx] break - existing_feature_views_of_same_type.append(feature_view_proto) if commit: self.commit() + def list_stream_feature_views( + self, project: str, allow_cache: bool = False + ) -> List[StreamFeatureView]: + """ + Retrieve a list of stream feature views from the registry + + Args: + project: Filter stream feature views based on project name + allow_cache: Whether to allow returning stream feature views from a cached registry + + Returns: + List of stream feature views + """ + registry = self._get_registry_proto(allow_cache=allow_cache) + stream_feature_views = [] + for stream_feature_view in registry.stream_feature_views: + if stream_feature_view.spec.project == project: + stream_feature_views.append( + StreamFeatureView.from_proto(stream_feature_view) + ) + return stream_feature_views + def list_on_demand_feature_views( self, project: str, allow_cache: bool = False ) -> List[OnDemandFeatureView]: @@ -764,6 +790,18 @@ def delete_feature_view(self, name: str, project: str, commit: bool = True): self.commit() return + for idx, existing_stream_feature_view_proto in enumerate( + self.cached_registry_proto.stream_feature_views + ): + if ( + existing_stream_feature_view_proto.spec.name == name + and existing_stream_feature_view_proto.spec.project == project + ): + del self.cached_registry_proto.stream_feature_views[idx] + if commit: + self.commit() + return + raise FeatureViewNotFoundException(name, project) def delete_entity(self, name: str, project: str, commit: bool = True): diff --git a/sdk/python/feast/repo_contents.py b/sdk/python/feast/repo_contents.py index 4d7c92f2a6d..fe5cbd284bc 100644 --- a/sdk/python/feast/repo_contents.py +++ b/sdk/python/feast/repo_contents.py @@ -20,6 +20,7 @@ from feast.on_demand_feature_view import OnDemandFeatureView from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto from feast.request_feature_view import RequestFeatureView +from feast.stream_feature_view import StreamFeatureView class RepoContents(NamedTuple): @@ -31,6 +32,7 @@ class RepoContents(NamedTuple): feature_views: List[FeatureView] on_demand_feature_views: List[OnDemandFeatureView] request_feature_views: List[RequestFeatureView] + stream_feature_views: List[StreamFeatureView] entities: List[Entity] feature_services: List[FeatureService] @@ -50,4 +52,7 @@ def to_registry_proto(self) -> RegistryProto: registry_proto.feature_services.extend( [fs.to_proto() for fs in self.feature_services] ) + registry_proto.stream_feature_views.extend( + [fv.to_proto() for fv in self.stream_feature_views] + ) return registry_proto diff --git a/sdk/python/feast/repo_operations.py b/sdk/python/feast/repo_operations.py index 0e82fdf47ad..8b81a71bae4 100644 --- a/sdk/python/feast/repo_operations.py +++ b/sdk/python/feast/repo_operations.py @@ -107,6 +107,7 @@ def parse_repo(repo_root: Path) -> RepoContents: feature_views=[], feature_services=[], on_demand_feature_views=[], + stream_feature_views=[], request_feature_views=[], ) diff --git a/sdk/python/feast/stream_feature_view.py b/sdk/python/feast/stream_feature_view.py index cfb3f63d7df..bba16e2627f 100644 --- a/sdk/python/feast/stream_feature_view.py +++ b/sdk/python/feast/stream_feature_view.py @@ -1,16 +1,42 @@ +import functools +import warnings from datetime import timedelta +from types import MethodType from typing import Dict, List, Optional, Union -from feast.data_source import DataSource +import dill +from google.protobuf.duration_pb2 import Duration + +from feast.aggregation import Aggregation +from feast.data_source import DataSource, KafkaSource from feast.entity import Entity from feast.feature_view import FeatureView from feast.field import Field from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto +from feast.protos.feast.core.OnDemandFeatureView_pb2 import ( + UserDefinedFunction as UserDefinedFunctionProto, +) +from feast.protos.feast.core.StreamFeatureView_pb2 import ( + StreamFeatureView as StreamFeatureViewProto, +) +from feast.protos.feast.core.StreamFeatureView_pb2 import ( + StreamFeatureViewMeta as StreamFeatureViewMetaProto, +) +from feast.protos.feast.core.StreamFeatureView_pb2 import ( + StreamFeatureViewSpec as StreamFeatureViewSpecProto, +) + +warnings.simplefilter("once", RuntimeWarning) -SUPPORTED_STREAM_SOURCES = {"KafkaSource", "KinesisSource", "PushSource"} +SUPPORTED_STREAM_SOURCES = {"KafkaSource", "PushSource"} class StreamFeatureView(FeatureView): + """ + NOTE: Stream Feature Views are not yet fully implemented and exist to allow users to register their stream sources and + schemas with Feast. + """ + def __init__( self, *, @@ -18,15 +44,24 @@ def __init__( entities: Optional[Union[List[Entity], List[str]]] = None, ttl: Optional[timedelta] = None, tags: Optional[Dict[str, str]] = None, - online: bool = True, - description: str = "", - owner: str = "", + online: Optional[bool] = True, + description: Optional[str] = "", + owner: Optional[str] = "", schema: Optional[List[Field]] = None, source: Optional[DataSource] = None, + aggregations: Optional[List[Aggregation]] = None, + mode: Optional[str] = "spark", # Mode of ingestion/transformation + timestamp_field: Optional[str] = "", # Timestamp for aggregation + udf: Optional[MethodType] = None, ): - + warnings.warn( + "Stream Feature Views are experimental features in alpha development. " + "Some functionality may still be unstable so functionality can change in the future.", + RuntimeWarning, + ) if source is None: - raise ValueError("Feature views need a source specified") + raise ValueError("Stream Feature views need a source specified") + # source uses the batch_source of the kafkasource in feature_view if ( type(source).__name__ not in SUPPORTED_STREAM_SOURCES and source.to_proto().type != DataSourceProto.SourceType.CUSTOM_SOURCE @@ -35,13 +70,20 @@ def __init__( f"Stream feature views need a stream source, expected one of {SUPPORTED_STREAM_SOURCES} " f"or CUSTOM_SOURCE, got {type(source).__name__}: {source.name} instead " ) + self.aggregations = aggregations + self.mode = mode + self.timestamp_field = timestamp_field + self.udf = udf + _batch_source = None + if isinstance(source, KafkaSource): + _batch_source = source.batch_source if source.batch_source else None super().__init__( name=name, entities=entities, ttl=ttl, - batch_source=None, - stream_source=None, + batch_source=_batch_source, + stream_source=source, tags=tags, online=online, description=description, @@ -49,3 +91,174 @@ def __init__( schema=schema, source=source, ) + + def __eq__(self, other): + if not isinstance(other, StreamFeatureView): + raise TypeError("Comparisons should only involve StreamFeatureViews") + + if not super().__eq__(other): + return False + + if ( + self.mode != other.mode + or self.timestamp_field != other.timestamp_field + or self.udf.__code__.co_code != other.udf.__code__.co_code + or self.aggregations != other.aggregations + ): + return False + + return True + + def __hash__(self): + return super().__hash__() + + def to_proto(self): + meta = StreamFeatureViewMetaProto() + if self.created_timestamp: + meta.created_timestamp.FromDatetime(self.created_timestamp) + if self.last_updated_timestamp: + meta.last_updated_timestamp.FromDatetime(self.last_updated_timestamp) + + ttl_duration = None + if self.ttl is not None: + ttl_duration = Duration() + ttl_duration.FromTimedelta(self.ttl) + + if self.batch_source: + batch_source_proto = self.batch_source.to_proto() + batch_source_proto.data_source_class_type = f"{self.batch_source.__class__.__module__}.{self.batch_source.__class__.__name__}" + + stream_source_proto = None + if self.stream_source: + stream_source_proto = self.stream_source.to_proto() + stream_source_proto.data_source_class_type = f"{self.stream_source.__class__.__module__}.{self.stream_source.__class__.__name__}" + + spec = StreamFeatureViewSpecProto( + name=self.name, + entities=self.entities, + entity_columns=[field.to_proto() for field in self.entity_columns], + features=[field.to_proto() for field in self.schema], + user_defined_function=UserDefinedFunctionProto( + name=self.udf.__name__, body=dill.dumps(self.udf, recurse=True), + ) + if self.udf + else None, + description=self.description, + tags=self.tags, + owner=self.owner, + ttl=(ttl_duration if ttl_duration is not None else None), + online=self.online, + batch_source=batch_source_proto or None, + stream_source=stream_source_proto, + timestamp_field=self.timestamp_field, + aggregations=[agg.to_proto() for agg in self.aggregations], + mode=self.mode, + ) + + return StreamFeatureViewProto(spec=spec, meta=meta) + + @classmethod + def from_proto(cls, sfv_proto): + batch_source = ( + DataSource.from_proto(sfv_proto.spec.batch_source) + if sfv_proto.spec.HasField("batch_source") + else None + ) + stream_source = ( + DataSource.from_proto(sfv_proto.spec.stream_source) + if sfv_proto.spec.HasField("stream_source") + else None + ) + sfv_feature_view = cls( + name=sfv_proto.spec.name, + description=sfv_proto.spec.description, + tags=dict(sfv_proto.spec.tags), + owner=sfv_proto.spec.owner, + online=sfv_proto.spec.online, + schema=[ + Field.from_proto(field_proto) for field_proto in sfv_proto.spec.features + ], + ttl=( + timedelta(days=0) + if sfv_proto.spec.ttl.ToNanoseconds() == 0 + else sfv_proto.spec.ttl.ToTimedelta() + ), + source=stream_source, + mode=sfv_proto.spec.mode, + udf=dill.loads(sfv_proto.spec.user_defined_function.body), + aggregations=[ + Aggregation.from_proto(agg_proto) + for agg_proto in sfv_proto.spec.aggregations + ], + timestamp_field=sfv_proto.spec.timestamp_field, + ) + + if batch_source: + sfv_feature_view.batch_source = batch_source + + if stream_source: + sfv_feature_view.stream_source = stream_source + + sfv_feature_view.entities = list(sfv_proto.spec.entities) + + sfv_feature_view.features = [ + Field.from_proto(field_proto) for field_proto in sfv_proto.spec.features + ] + + if sfv_proto.meta.HasField("created_timestamp"): + sfv_feature_view.created_timestamp = ( + sfv_proto.meta.created_timestamp.ToDatetime() + ) + if sfv_proto.meta.HasField("last_updated_timestamp"): + sfv_feature_view.last_updated_timestamp = ( + sfv_proto.meta.last_updated_timestamp.ToDatetime() + ) + + return sfv_feature_view + + +def stream_feature_view( + *, + entities: Optional[Union[List[Entity], List[str]]] = None, + ttl: Optional[timedelta] = None, + tags: Optional[Dict[str, str]] = None, + online: Optional[bool] = True, + description: Optional[str] = "", + owner: Optional[str] = "", + schema: Optional[List[Field]] = None, + source: Optional[DataSource] = None, + aggregations: Optional[List[Aggregation]] = None, + mode: Optional[str] = "spark", # Mode of ingestion/transformation + timestamp_field: Optional[str] = "", # Timestamp for aggregation +): + """ + Creates an StreamFeatureView object with the given user function as udf. + """ + + def mainify(obj): + # Needed to allow dill to properly serialize the udf. Otherwise, clients will need to have a file with the same + # name as the original file defining the sfv. + if obj.__module__ != "__main__": + obj.__module__ = "__main__" + + def decorator(user_function): + mainify(user_function) + stream_feature_view_obj = StreamFeatureView( + name=user_function.__name__, + entities=entities, + ttl=ttl, + source=source, + schema=schema, + udf=user_function, + description=description, + tags=tags, + online=online, + owner=owner, + aggregations=aggregations, + mode=mode, + timestamp_field=timestamp_field, + ) + functools.update_wrapper(wrapper=stream_feature_view_obj, wrapped=user_function) + return stream_feature_view_obj + + return decorator diff --git a/sdk/python/tests/integration/registration/test_registry.py b/sdk/python/tests/integration/registration/test_registry.py index bb02f9a9e32..222eb116d26 100644 --- a/sdk/python/tests/integration/registration/test_registry.py +++ b/sdk/python/tests/integration/registration/test_registry.py @@ -20,7 +20,9 @@ from pytest_lazyfixture import lazy_fixture from feast import FileSource -from feast.data_format import ParquetFormat +from feast.aggregation import Aggregation +from feast.data_format import AvroFormat, ParquetFormat +from feast.data_source import KafkaSource from feast.entity import Entity from feast.feature import Feature from feast.feature_view import FeatureView @@ -28,6 +30,7 @@ from feast.on_demand_feature_view import RequestSource, on_demand_feature_view from feast.registry import Registry from feast.repo_config import RegistryConfig +from feast.stream_feature_view import StreamFeatureView from feast.types import Array, Bytes, Float32, Int32, Int64, String from feast.value_type import ValueType @@ -299,6 +302,70 @@ def location_features_from_push(inputs: pd.DataFrame) -> pd.DataFrame: test_registry._get_registry_proto() +@pytest.mark.parametrize( + "test_registry", [lazy_fixture("local_registry")], +) +def test_apply_stream_feature_view_success(test_registry): + # Create Feature Views + def simple_udf(x: int): + return x + 3 + + entity = Entity(name="driver_entity", join_keys=["test_key"]) + + stream_source = KafkaSource( + name="kafka", + timestamp_field="event_timestamp", + bootstrap_servers="", + message_format=AvroFormat(""), + topic="topic", + batch_source=FileSource(path="some path"), + ) + + sfv = StreamFeatureView( + name="test kafka stream feature view", + entities=[entity], + ttl=timedelta(days=30), + owner="test@example.com", + online=True, + schema=[Field(name="dummy_field", dtype=Float32)], + description="desc", + aggregations=[ + Aggregation( + column="dummy_field", function="max", time_window=timedelta(days=1), + ), + Aggregation( + column="dummy_field2", function="count", time_window=timedelta(days=24), + ), + ], + timestamp_field="event_timestamp", + mode="spark", + source=stream_source, + udf=simple_udf, + tags={}, + ) + + project = "project" + + # Register Feature View + test_registry.apply_feature_view(sfv, project) + + stream_feature_views = test_registry.list_stream_feature_views(project) + + # List Feature Views + assert len(stream_feature_views) == 1 + assert stream_feature_views[0] == sfv + + test_registry.delete_feature_view("test kafka stream feature view", project) + stream_feature_views = test_registry.list_stream_feature_views(project) + assert len(stream_feature_views) == 0 + + test_registry.teardown() + + # Will try to reload registry, which will fail because the file has been deleted + with pytest.raises(FileNotFoundError): + test_registry._get_registry_proto() + + @pytest.mark.parametrize( "test_registry", [lazy_fixture("local_registry")], ) diff --git a/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py b/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py new file mode 100644 index 00000000000..b01ca434fa8 --- /dev/null +++ b/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py @@ -0,0 +1,62 @@ +from datetime import timedelta + +import pytest + +from feast import Entity, Field, FileSource +from feast.aggregation import Aggregation +from feast.data_format import AvroFormat +from feast.data_source import KafkaSource +from feast.stream_feature_view import stream_feature_view +from feast.types import Float32 + + +@pytest.mark.integration +def test_read_pre_applied(environment) -> None: + """ + Test apply of StreamFeatureView. + """ + fs = environment.feature_store + + # Create Feature Views + entity = Entity(name="driver_entity", join_keys=["test_key"]) + + stream_source = KafkaSource( + name="kafka", + timestamp_field="event_timestamp", + bootstrap_servers="", + message_format=AvroFormat(""), + topic="topic", + batch_source=FileSource(path="test_path", timestamp_field="event_timestamp"), + ) + + @stream_feature_view( + entities=[entity], + ttl=timedelta(days=30), + owner="test@example.com", + online=True, + schema=[Field(name="dummy_field", dtype=Float32)], + description="desc", + aggregations=[ + Aggregation( + column="dummy_field", function="max", time_window=timedelta(days=1), + ), + Aggregation( + column="dummy_field2", function="count", time_window=timedelta(days=24), + ), + ], + timestamp_field="event_timestamp", + mode="spark", + source=stream_source, + tags={}, + ) + def simple_sfv(df): + return df + + fs.apply([entity, simple_sfv]) + stream_feature_views = fs.list_stream_feature_views() + assert len(stream_feature_views) == 1 + assert stream_feature_views[0] == simple_sfv + + entities = fs.list_entities() + assert len(entities) == 1 + assert entities[0] == entity diff --git a/sdk/python/tests/unit/test_feature_views.py b/sdk/python/tests/unit/test_feature_views.py index f72ae4fe9cb..904260dfe61 100644 --- a/sdk/python/tests/unit/test_feature_views.py +++ b/sdk/python/tests/unit/test_feature_views.py @@ -2,12 +2,15 @@ import pytest -from feast import PushSource +from feast.aggregation import Aggregation from feast.batch_feature_view import BatchFeatureView from feast.data_format import AvroFormat -from feast.data_source import KafkaSource +from feast.data_source import KafkaSource, PushSource +from feast.entity import Entity +from feast.field import Field from feast.infra.offline_stores.file_source import FileSource from feast.stream_feature_view import StreamFeatureView +from feast.types import Float32 def test_create_batch_feature_view(): @@ -26,7 +29,7 @@ def test_create_batch_feature_view(): stream_source = KafkaSource( name="kafka", - timestamp_field="", + timestamp_field="event_timestamp", bootstrap_servers="", message_format=AvroFormat(""), topic="topic", @@ -44,7 +47,7 @@ def test_create_batch_feature_view(): def test_create_stream_feature_view(): stream_source = KafkaSource( name="kafka", - timestamp_field="", + timestamp_field="event_timestamp", bootstrap_servers="", message_format=AvroFormat(""), topic="topic", @@ -55,6 +58,7 @@ def test_create_stream_feature_view(): entities=[], ttl=timedelta(days=30), source=stream_source, + aggregations=[], ) push_source = PushSource( @@ -65,11 +69,15 @@ def test_create_stream_feature_view(): entities=[], ttl=timedelta(days=30), source=push_source, + aggregations=[], ) with pytest.raises(ValueError): StreamFeatureView( - name="test batch feature view", entities=[], ttl=timedelta(days=30) + name="test batch feature view", + entities=[], + ttl=timedelta(days=30), + aggregations=[], ) with pytest.raises(ValueError): @@ -78,4 +86,46 @@ def test_create_stream_feature_view(): entities=[], ttl=timedelta(days=30), source=FileSource(path="some path"), + aggregations=[], ) + + +def simple_udf(x: int): + return x + 3 + + +def test_stream_feature_view_serialization(): + entity = Entity(name="driver_entity", join_keys=["test_key"]) + stream_source = KafkaSource( + name="kafka", + timestamp_field="event_timestamp", + bootstrap_servers="", + message_format=AvroFormat(""), + topic="topic", + batch_source=FileSource(path="some path"), + ) + + sfv = StreamFeatureView( + name="test kafka stream feature view", + entities=[entity], + ttl=timedelta(days=30), + owner="test@example.com", + online=True, + schema=[Field(name="dummy_field", dtype=Float32)], + description="desc", + aggregations=[ + Aggregation( + column="dummy_field", function="max", time_window=timedelta(days=1), + ) + ], + timestamp_field="event_timestamp", + mode="spark", + source=stream_source, + udf=simple_udf, + tags={}, + ) + + sfv_proto = sfv.to_proto() + + new_sfv = StreamFeatureView.from_proto(sfv_proto=sfv_proto) + assert new_sfv == sfv