Skip to content

Commit b7416eb

Browse files
felixwang9817felixwang9817
committed
Fix bugs with stream feature view materialization
Signed-off-by: Felix Wang <wangfelix98@gmail.com>
1 parent 2ff317f commit b7416eb

7 files changed

Lines changed: 132 additions & 95 deletions

File tree

‎sdk/python/feast/feature_store.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -569,6 +569,9 @@ def _make_inferences(
569569
update_feature_views_with_inferred_features_and_entities(
570570
views_to_update, entities + entities_to_update, self.config
571571
)
572+
update_feature_views_with_inferred_features_and_entities(
573+
sfvs_to_update, entities + entities_to_update, self.config
574+
)
572575
# TODO(kevjumba): Update schema inferrence
573576
for sfv in sfvs_to_update:
574577
if not sfv.schema:
@@ -931,8 +934,8 @@ def apply(
931934

932935
self._get_provider().update_infra(
933936
project=self.project,
934-
tables_to_delete=views_to_delete if not partial else [],
935-
tables_to_keep=views_to_update,
937+
tables_to_delete=views_to_delete + sfvs_to_delete if not partial else [],
938+
tables_to_keep=views_to_update + sfvs_to_update,
936939
entities_to_delete=entities_to_delete if not partial else [],
937940
entities_to_keep=entities_to_update,
938941
partial=partial,
@@ -1357,6 +1360,7 @@ def push(
13571360
from feast.data_source import PushSource
13581361

13591362
all_fvs = self.list_feature_views(allow_cache=allow_registry_cache)
1363+
all_fvs += self.list_stream_feature_views(allow_cache=allow_registry_cache)
13601364

13611365
fvs_with_push_sources = {
13621366
fv

‎sdk/python/feast/inference.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,10 @@ def update_feature_views_with_inferred_features_and_entities(
9999
other columns except designated timestamp columns are considered to be feature columns. If
100100
the feature view already has features, feature inference is skipped.
101101
102+
Note that this inference logic currently does not take transformations into account. For
103+
example, even if a stream feature view has a transformation, this method assumes that the
104+
batch source contains transformed data with the correct final schema.
105+
102106
Args:
103107
fvs: The feature views to be updated.
104108
entities: A list containing entities associated with the feature views.

‎sdk/python/feast/registry.py‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1267,6 +1267,30 @@ def apply_materialization(
12671267
self.commit()
12681268
return
12691269

1270+
for idx, existing_stream_feature_view_proto in enumerate(
1271+
self.cached_registry_proto.stream_feature_views
1272+
):
1273+
if (
1274+
existing_stream_feature_view_proto.spec.name == feature_view.name
1275+
and existing_stream_feature_view_proto.spec.project == project
1276+
):
1277+
existing_stream_feature_view = StreamFeatureView.from_proto(
1278+
existing_stream_feature_view_proto
1279+
)
1280+
existing_stream_feature_view.materialization_intervals.append(
1281+
(start_date, end_date)
1282+
)
1283+
existing_stream_feature_view.last_updated_timestamp = datetime.utcnow()
1284+
stream_feature_view_proto = existing_stream_feature_view.to_proto()
1285+
stream_feature_view_proto.spec.project = project
1286+
del self.cached_registry_proto.stream_feature_views[idx]
1287+
self.cached_registry_proto.stream_feature_views.append(
1288+
stream_feature_view_proto
1289+
)
1290+
if commit:
1291+
self.commit()
1292+
return
1293+
12701294
raise FeatureViewNotFoundException(feature_view.name, project)
12711295

12721296
def list_feature_views(

‎sdk/python/feast/stream_feature_view.py‎

Lines changed: 44 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
import copy
22
import functools
33
import warnings
4-
from datetime import timedelta
4+
from datetime import datetime, timedelta
55
from types import MethodType
6-
from typing import Dict, List, Optional, Union
6+
from typing import Dict, List, Optional, Tuple, Union
77

88
import dill
99
from google.protobuf.duration_pb2 import Duration
@@ -42,26 +42,42 @@ class StreamFeatureView(FeatureView):
4242
schemas with Feast.
4343
4444
Attributes:
45-
name: str. The unique name of the stream feature view.
46-
entities: Union[List[Entity], List[str]]. List of entities or entity join keys.
47-
ttl: timedelta. The amount of time this group of features lives. A ttl of 0 indicates that
45+
name: The unique name of the stream feature view.
46+
entities: List of entities or entity join keys.
47+
ttl: The amount of time this group of features lives. A ttl of 0 indicates that
4848
this group of features lives forever. Note that large ttl's or a ttl of 0
4949
can result in extremely computationally intensive queries.
50-
tags: Dict[str, str]. A dictionary of key-value pairs to store arbitrary metadata.
51-
online: bool. Defines whether this stream feature view is used in online feature retrieval.
52-
description: str. A human-readable description.
50+
schema: The schema of the feature view, including feature, timestamp, and entity
51+
columns. If not specified, can be inferred from the underlying data source.
52+
source: DataSource. The stream source of data where this group of features is stored.
53+
aggregations: List of aggregations registered with the stream feature view.
54+
mode: The mode of execution.
55+
timestamp_field: Must be specified if aggregations are specified. Defines the timestamp column on which to aggregate windows.
56+
online: Defines whether this stream feature view is used in online feature retrieval.
57+
description: A human-readable description.
58+
tags: A dictionary of key-value pairs to store arbitrary metadata.
5359
owner: The owner of the on demand feature view, typically the email of the primary
5460
maintainer.
55-
schema: List[Field] The schema of the feature view, including feature, timestamp, and entity
56-
columns. If not specified, can be inferred from the underlying data source.
57-
source: DataSource. The stream source of data where this group of features
58-
is stored.
59-
aggregations (optional): List[Aggregation]. List of aggregations registered with the stream feature view.
60-
mode(optional): str. The mode of execution.
61-
timestamp_field (optional): Must be specified if aggregations are specified. Defines the timestamp column on which to aggregate windows.
62-
udf (optional): MethodType The user defined transformation function. This transformation function should have all of the corresponding imports imported within the function.
61+
udf: The user defined transformation function. This transformation function should have all of the corresponding imports imported within the function.
6362
"""
6463

64+
name: str
65+
entities: List[str]
66+
ttl: Optional[timedelta]
67+
source: DataSource
68+
schema: List[Field]
69+
entity_columns: List[Field]
70+
features: List[Field]
71+
online: bool
72+
description: str
73+
tags: Dict[str, str]
74+
owner: str
75+
aggregations: List[Aggregation]
76+
mode: str
77+
timestamp_field: str
78+
materialization_intervals: List[Tuple[datetime, datetime]]
79+
udf: Optional[MethodType]
80+
6581
def __init__(
6682
self,
6783
*,
@@ -222,7 +238,7 @@ def from_proto(cls, sfv_proto):
222238
if sfv_proto.spec.HasField("user_defined_function")
223239
else None
224240
)
225-
sfv_feature_view = cls(
241+
stream_feature_view = cls(
226242
name=sfv_proto.spec.name,
227243
description=sfv_proto.spec.description,
228244
tags=dict(sfv_proto.spec.tags),
@@ -247,23 +263,27 @@ def from_proto(cls, sfv_proto):
247263
)
248264

249265
if batch_source:
250-
sfv_feature_view.batch_source = batch_source
266+
stream_feature_view.batch_source = batch_source
251267

252268
if stream_source:
253-
sfv_feature_view.stream_source = stream_source
269+
stream_feature_view.stream_source = stream_source
254270

255-
sfv_feature_view.entities = list(sfv_proto.spec.entities)
271+
stream_feature_view.entities = list(sfv_proto.spec.entities)
256272

257-
sfv_feature_view.features = [
273+
stream_feature_view.features = [
258274
Field.from_proto(field_proto) for field_proto in sfv_proto.spec.features
259275
]
276+
stream_feature_view.entity_columns = [
277+
Field.from_proto(field_proto)
278+
for field_proto in sfv_proto.spec.entity_columns
279+
]
260280

261281
if sfv_proto.meta.HasField("created_timestamp"):
262-
sfv_feature_view.created_timestamp = (
282+
stream_feature_view.created_timestamp = (
263283
sfv_proto.meta.created_timestamp.ToDatetime()
264284
)
265285
if sfv_proto.meta.HasField("last_updated_timestamp"):
266-
sfv_feature_view.last_updated_timestamp = (
286+
stream_feature_view.last_updated_timestamp = (
267287
sfv_proto.meta.last_updated_timestamp.ToDatetime()
268288
)
269289

@@ -275,7 +295,7 @@ def from_proto(cls, sfv_proto):
275295
)
276296
)
277297

278-
return sfv_feature_view
298+
return stream_feature_view
279299

280300
def __copy__(self):
281301
fv = StreamFeatureView(

‎sdk/python/tests/integration/feature_repos/universal/feature_views.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
Field,
1212
OnDemandFeatureView,
1313
PushSource,
14+
StreamFeatureView,
1415
ValueType,
1516
)
1617
from feast.data_source import DataSource, RequestSource
@@ -297,7 +298,7 @@ def create_pushable_feature_view(batch_source: DataSource):
297298
push_source = PushSource(
298299
name="location_stats_push_source", batch_source=batch_source,
299300
)
300-
return FeatureView(
301+
return StreamFeatureView(
301302
name="pushable_location_stats",
302303
entities=[location()],
303304
schema=[

‎sdk/python/tests/unit/test_feature_view.py‎

Lines changed: 0 additions & 68 deletions
This file was deleted.

‎sdk/python/tests/unit/test_feature_views.py‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
from feast.data_format import AvroFormat
88
from feast.data_source import KafkaSource, PushSource
99
from feast.entity import Entity
10+
from feast.feature_view import FeatureView
1011
from feast.field import Field
1112
from feast.infra.offline_stores.file_source import FileSource
1213
from feast.stream_feature_view import StreamFeatureView, stream_feature_view
@@ -201,3 +202,54 @@ def test_stream_feature_view_initialization_with_optional_fields_omitted():
201202

202203
new_sfv = StreamFeatureView.from_proto(sfv_proto=sfv_proto)
203204
assert new_sfv == sfv
205+
206+
207+
def test_hash():
208+
file_source = FileSource(name="my-file-source", path="test.parquet")
209+
feature_view_1 = FeatureView(
210+
name="my-feature-view",
211+
entities=[],
212+
schema=[
213+
Field(name="feature1", dtype=Float32),
214+
Field(name="feature2", dtype=Float32),
215+
],
216+
source=file_source,
217+
)
218+
feature_view_2 = FeatureView(
219+
name="my-feature-view",
220+
entities=[],
221+
schema=[
222+
Field(name="feature1", dtype=Float32),
223+
Field(name="feature2", dtype=Float32),
224+
],
225+
source=file_source,
226+
)
227+
feature_view_3 = FeatureView(
228+
name="my-feature-view",
229+
entities=[],
230+
schema=[Field(name="feature1", dtype=Float32)],
231+
source=file_source,
232+
)
233+
feature_view_4 = FeatureView(
234+
name="my-feature-view",
235+
entities=[],
236+
schema=[Field(name="feature1", dtype=Float32)],
237+
source=file_source,
238+
description="test",
239+
)
240+
241+
s1 = {feature_view_1, feature_view_2}
242+
assert len(s1) == 1
243+
244+
s2 = {feature_view_1, feature_view_3}
245+
assert len(s2) == 2
246+
247+
s3 = {feature_view_3, feature_view_4}
248+
assert len(s3) == 2
249+
250+
s4 = {feature_view_1, feature_view_2, feature_view_3, feature_view_4}
251+
assert len(s4) == 3
252+
253+
254+
# TODO(felixwang9817): Add tests for proto conversion.
255+
# TODO(felixwang9817): Add tests for field mapping logic.

0 commit comments

Comments
 (0)