Skip to content

Commit 9bbc1c6

Browse files
franciscojavierarceofranciscojavierarceo
authored
feat: Adding support for native Python transformations on a single dictionary (#4724)
* feat: Adding support for native Python transformations on a dictionary Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * Updated type checking and added exception handling to try basic dict...not an ideal solution Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * updated tests Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * adding protos Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> * fixed unit test Signed-off-by: Francisco Javier Arceo <farceo@redhat.com> --------- Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
1 parent 84b24b5 commit 9bbc1c6

9 files changed

Lines changed: 134 additions & 56 deletions

File tree

‎protos/feast/core/OnDemandFeatureView.proto‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ message OnDemandFeatureViewSpec {
6969
repeated string entities = 13;
7070
// List of specifications for each entity defined as part of this feature view.
7171
repeated FeatureSpecV2 entity_columns = 14;
72+
bool singleton = 15;
7273
}
7374

7475
message OnDemandFeatureViewMeta {

‎sdk/python/feast/on_demand_feature_view.py‎

Lines changed: 38 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,7 @@ class OnDemandFeatureView(BaseFeatureView):
7474
tags: dict[str, str]
7575
owner: str
7676
write_to_online_store: bool
77+
singleton: bool
7778

7879
def __init__( # noqa: C901
7980
self,
@@ -98,6 +99,7 @@ def __init__( # noqa: C901
9899
tags: Optional[dict[str, str]] = None,
99100
owner: str = "",
100101
write_to_online_store: bool = False,
102+
singleton: bool = False,
101103
):
102104
"""
103105
Creates an OnDemandFeatureView object.
@@ -121,6 +123,8 @@ def __init__( # noqa: C901
121123
of the primary maintainer.
122124
write_to_online_store (optional): A boolean that indicates whether to write the on demand feature view to
123125
the online store for faster retrieval.
126+
singleton (optional): A boolean that indicates whether the transformation is executed on a singleton
127+
(only applicable when mode="python").
124128
"""
125129
super().__init__(
126130
name=name,
@@ -204,6 +208,9 @@ def __init__( # noqa: C901
204208
self.features = features
205209
self.feature_transformation = feature_transformation
206210
self.write_to_online_store = write_to_online_store
211+
self.singleton = singleton
212+
if self.singleton and self.mode != "python":
213+
raise ValueError("Singleton is only supported for Python mode.")
207214

208215
@property
209216
def proto_class(self) -> type[OnDemandFeatureViewProto]:
@@ -221,6 +228,7 @@ def __copy__(self):
221228
tags=self.tags,
222229
owner=self.owner,
223230
write_to_online_store=self.write_to_online_store,
231+
singleton=self.singleton,
224232
)
225233
fv.entities = self.entities
226234
fv.features = self.features
@@ -247,6 +255,7 @@ def __eq__(self, other):
247255
or self.feature_transformation != other.feature_transformation
248256
or self.write_to_online_store != other.write_to_online_store
249257
or sorted(self.entity_columns) != sorted(other.entity_columns)
258+
or self.singleton != other.singleton
250259
):
251260
return False
252261

@@ -328,6 +337,7 @@ def to_proto(self) -> OnDemandFeatureViewProto:
328337
tags=self.tags,
329338
owner=self.owner,
330339
write_to_online_store=self.write_to_online_store,
340+
singleton=self.singleton if self.singleton else False,
331341
)
332342

333343
return OnDemandFeatureViewProto(spec=spec, meta=meta)
@@ -434,6 +444,9 @@ def from_proto(
434444
]
435445
else:
436446
entity_columns = []
447+
singleton = False
448+
if hasattr(on_demand_feature_view_proto.spec, "singleton"):
449+
singleton = on_demand_feature_view_proto.spec.singleton
437450

438451
on_demand_feature_view_obj = cls(
439452
name=on_demand_feature_view_proto.spec.name,
@@ -451,6 +464,7 @@ def from_proto(
451464
tags=dict(on_demand_feature_view_proto.spec.tags),
452465
owner=on_demand_feature_view_proto.spec.owner,
453466
write_to_online_store=write_to_online_store,
467+
singleton=singleton,
454468
)
455469

456470
on_demand_feature_view_obj.entities = entities
@@ -614,17 +628,19 @@ def transform_dict(
614628
feature_dict[full_feature_ref] = feature_dict[feature.name]
615629
columns_to_cleanup.append(str(full_feature_ref))
616630

617-
output_dict: dict[str, Any] = self.feature_transformation.transform(
618-
feature_dict
619-
)
631+
if self.singleton and self.mode == "python":
632+
output_dict: dict[str, Any] = (
633+
self.feature_transformation.transform_singleton(feature_dict)
634+
)
635+
else:
636+
output_dict = self.feature_transformation.transform(feature_dict)
620637
for feature_name in columns_to_cleanup:
621638
del output_dict[feature_name]
622639
return output_dict
623640

624641
def infer_features(self) -> None:
625-
inferred_features = self.feature_transformation.infer_features(
626-
self._construct_random_input()
627-
)
642+
random_input = self._construct_random_input(singleton=self.singleton)
643+
inferred_features = self.feature_transformation.infer_features(random_input)
628644

629645
if self.features:
630646
missing_features = []
@@ -644,8 +660,10 @@ def infer_features(self) -> None:
644660
f"Could not infer Features for the feature view '{self.name}'.",
645661
)
646662

647-
def _construct_random_input(self) -> dict[str, list[Any]]:
648-
rand_dict_value: dict[ValueType, list[Any]] = {
663+
def _construct_random_input(
664+
self, singleton: bool = False
665+
) -> dict[str, Union[list[Any], Any]]:
666+
rand_dict_value: dict[ValueType, Union[list[Any], Any]] = {
649667
ValueType.BYTES: [str.encode("hello world")],
650668
ValueType.STRING: ["hello world"],
651669
ValueType.INT32: [1],
@@ -663,20 +681,25 @@ def _construct_random_input(self) -> dict[str, list[Any]]:
663681
ValueType.BOOL_LIST: [[True]],
664682
ValueType.UNIX_TIMESTAMP_LIST: [[_utc_now()]],
665683
}
684+
if singleton:
685+
rand_dict_value = {k: rand_dict_value[k][0] for k in rand_dict_value}
666686

687+
rand_missing_value = [None] if singleton else None
667688
feature_dict = {}
668689
for feature_view_projection in self.source_feature_view_projections.values():
669690
for feature in feature_view_projection.features:
670691
feature_dict[f"{feature_view_projection.name}__{feature.name}"] = (
671-
rand_dict_value.get(feature.dtype.to_value_type(), [None])
692+
rand_dict_value.get(
693+
feature.dtype.to_value_type(), rand_missing_value
694+
)
672695
)
673696
feature_dict[f"{feature.name}"] = rand_dict_value.get(
674-
feature.dtype.to_value_type(), [None]
697+
feature.dtype.to_value_type(), rand_missing_value
675698
)
676699
for request_data in self.source_request_sources.values():
677700
for field in request_data.schema:
678701
feature_dict[f"{field.name}"] = rand_dict_value.get(
679-
field.dtype.to_value_type(), [None]
702+
field.dtype.to_value_type(), rand_missing_value
680703
)
681704

682705
return feature_dict
@@ -713,6 +736,7 @@ def on_demand_feature_view(
713736
tags: Optional[dict[str, str]] = None,
714737
owner: str = "",
715738
write_to_online_store: bool = False,
739+
singleton: bool = False,
716740
):
717741
"""
718742
Creates an OnDemandFeatureView object with the given user function as udf.
@@ -731,6 +755,8 @@ def on_demand_feature_view(
731755
of the primary maintainer.
732756
write_to_online_store (optional): A boolean that indicates whether to write the on demand feature view to
733757
the online store for faster retrieval.
758+
singleton (optional): A boolean that indicates whether the transformation is executed on a singleton
759+
(only applicable when mode="python").
734760
"""
735761

736762
def mainify(obj) -> None:
@@ -775,6 +801,7 @@ def decorator(user_function):
775801
owner=owner,
776802
write_to_online_store=write_to_online_store,
777803
entities=entities,
804+
singleton=singleton,
778805
)
779806
functools.update_wrapper(
780807
wrapper=on_demand_feature_view_obj, wrapped=user_function

‎sdk/python/feast/protos/feast/core/OnDemandFeatureView_pb2.py‎

Lines changed: 12 additions & 12 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎sdk/python/feast/protos/feast/core/OnDemandFeatureView_pb2.pyi‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,7 @@ class OnDemandFeatureViewSpec(google.protobuf.message.Message):
107107
WRITE_TO_ONLINE_STORE_FIELD_NUMBER: builtins.int
108108
ENTITIES_FIELD_NUMBER: builtins.int
109109
ENTITY_COLUMNS_FIELD_NUMBER: builtins.int
110+
SINGLETON_FIELD_NUMBER: builtins.int
110111
name: builtins.str
111112
"""Name of the feature view. Must be unique. Not updated."""
112113
project: builtins.str
@@ -137,6 +138,7 @@ class OnDemandFeatureViewSpec(google.protobuf.message.Message):
137138
@property
138139
def entity_columns(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[feast.core.Feature_pb2.FeatureSpecV2]:
139140
"""List of specifications for each entity defined as part of this feature view."""
141+
singleton: builtins.bool
140142
def __init__(
141143
self,
142144
*,
@@ -153,9 +155,10 @@ class OnDemandFeatureViewSpec(google.protobuf.message.Message):
153155
write_to_online_store: builtins.bool = ...,
154156
entities: collections.abc.Iterable[builtins.str] | None = ...,
155157
entity_columns: collections.abc.Iterable[feast.core.Feature_pb2.FeatureSpecV2] | None = ...,
158+
singleton: builtins.bool = ...,
156159
) -> None: ...
157160
def HasField(self, field_name: typing_extensions.Literal["feature_transformation", b"feature_transformation", "user_defined_function", b"user_defined_function"]) -> builtins.bool: ...
158-
def ClearField(self, field_name: typing_extensions.Literal["description", b"description", "entities", b"entities", "entity_columns", b"entity_columns", "feature_transformation", b"feature_transformation", "features", b"features", "mode", b"mode", "name", b"name", "owner", b"owner", "project", b"project", "sources", b"sources", "tags", b"tags", "user_defined_function", b"user_defined_function", "write_to_online_store", b"write_to_online_store"]) -> None: ...
161+
def ClearField(self, field_name: typing_extensions.Literal["description", b"description", "entities", b"entities", "entity_columns", b"entity_columns", "feature_transformation", b"feature_transformation", "features", b"features", "mode", b"mode", "name", b"name", "owner", b"owner", "project", b"project", "singleton", b"singleton", "sources", b"sources", "tags", b"tags", "user_defined_function", b"user_defined_function", "write_to_online_store", b"write_to_online_store"]) -> None: ...
159162

160163
global___OnDemandFeatureViewSpec = OnDemandFeatureViewSpec
161164

‎sdk/python/feast/transformation/pandas_transformation.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,11 @@ def transform_arrow(
3535
def transform(self, input_df: pd.DataFrame) -> pd.DataFrame:
3636
return self.udf(input_df)
3737

38+
def transform_singleton(self, input_df: pd.DataFrame) -> pd.DataFrame:
39+
raise ValueError(
40+
"PandasTransformation does not support singleton transformations."
41+
)
42+
3843
def infer_features(self, random_input: dict[str, list[Any]]) -> list[Field]:
3944
df = pd.DataFrame.from_dict(random_input)
4045
output_df: pd.DataFrame = self.transform(df)

‎sdk/python/feast/transformation/python_transformation.py‎

Lines changed: 24 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -37,24 +37,39 @@ def transform(self, input_dict: dict) -> dict:
3737
output_dict = self.udf.__call__(input_dict)
3838
return {**input_dict, **output_dict}
3939

40-
def infer_features(self, random_input: dict[str, list[Any]]) -> list[Field]:
41-
output_dict: dict[str, list[Any]] = self.transform(random_input)
40+
def transform_singleton(self, input_dict: dict) -> dict:
41+
# This flattens the list of elements to extract the first one
42+
# in the case of a singleton element, it takes the value directly
43+
# in the case of a list of lists, it takes the first list
44+
input_dict = {k: v[0] for k, v in input_dict.items()}
45+
output_dict = self.udf.__call__(input_dict)
46+
return {**input_dict, **output_dict}
47+
48+
def infer_features(self, random_input: dict[str, Any]) -> list[Field]:
49+
output_dict: dict[str, Any] = self.transform(random_input)
4250

4351
fields = []
4452
for feature_name, feature_value in output_dict.items():
45-
if len(feature_value) <= 0:
46-
raise TypeError(
47-
f"Failed to infer type for feature '{feature_name}' with value "
48-
+ f"'{feature_value}' since no items were returned by the UDF."
49-
)
53+
if isinstance(feature_value, list):
54+
if len(feature_value) <= 0:
55+
raise TypeError(
56+
f"Failed to infer type for feature '{feature_name}' with value "
57+
+ f"'{feature_value}' since no items were returned by the UDF."
58+
)
59+
inferred_type = type(feature_value[0])
60+
inferred_value = feature_value[0]
61+
else:
62+
inferred_type = type(feature_value)
63+
inferred_value = feature_value
64+
5065
fields.append(
5166
Field(
5267
name=feature_name,
5368
dtype=from_value_type(
5469
python_type_to_feast_value_type(
5570
feature_name,
56-
value=feature_value[0],
57-
type_name=type(feature_value[0]).__name__,
71+
value=inferred_value,
72+
type_name=inferred_type.__name__,
5873
)
5974
),
6075
)

‎sdk/python/feast/transformation/substrait_transformation.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,11 @@ def table_provider(names, schema: pyarrow.Schema):
3838
).read_all()
3939
return table.to_pandas()
4040

41+
def transform_singleton(self, input_df: pd.DataFrame) -> pd.DataFrame:
42+
raise ValueError(
43+
"SubstraitTransform does not support singleton transformations."
44+
)
45+
4146
def transform_ibis(self, table):
4247
return self.ibis_function(table)
4348

0 commit comments

Comments
 (0)