on-deamnd tranformations working

python/hsfs/builtin_transformations.py

@@ -0,0 +1,66 @@
+#
+#   Copyright 2024 Hopsworks AB
+#
+#   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
+#
+#       http://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.
+#
+
+import numpy as np
+import pandas as pd
+from hsfs.hopsworks_udf import udf
+from hsfs.transformation_statistics import TransformationStatistics
+
+
+feature_statistics = TransformationStatistics("feature")
+
+
+@udf(float, drop=["feature"])
+def min_max_scaler(feature: pd.Series, statistics=feature_statistics) -> pd.Series:
+    return (feature - statistics.feature.min) / (
+        statistics.feature.max - statistics.feature.min
+    )
+
+
+@udf(float, drop=["feature"])
+def standard_scaler(feature: pd.Series, statistics=feature_statistics) -> pd.Series:
+    return (feature - statistics.feature.mean) / statistics.feature.stddev
+
+
+@udf(float, drop=["feature"])
+def robust_scaler(feature: pd.Series, statistics=feature_statistics) -> pd.Series:
+    return (feature - statistics.feature.percentiles[49]) / (
+        statistics.feature.percentiles[74] - statistics.feature.percentiles[24]
+    )
+
+
+@udf(int, drop=["feature"])
+def label_encoder(feature: pd.Series, statistics=feature_statistics) -> pd.Series:
+    unique_data = sorted(
+        [value for value in statistics.feature.extended_statistics["unique_values"]]
+    )
+    value_to_index = {value: index for index, value in enumerate(unique_data)}
+    return pd.Series(
+        [value_to_index[data] if not pd.isna(data) else np.nan for data in feature]
+    )
+
+
+@udf(bool, drop=["feature"])
+def one_hot_encoder(feature: pd.Series, statistics=feature_statistics) -> pd.Series:
+    unique_data = [
+        value for value in statistics.feature.extended_statistics["unique_values"]
+    ]
+    one_hot = pd.get_dummies(feature, dtype="bool")
+    for data in unique_data:
+        if data not in one_hot:
+            one_hot[data] = False
+    # Sorting by columns so as to maintain consistency in column order.
+    return one_hot.reindex(sorted(one_hot.columns), axis=1)

python/hsfs/constructor/query.py

@@ -59,7 +59,7 @@ def __init__(
             fg_mod.ExternalFeatureGroup,
             fg_mod.SpineGroup,
         ],
-        left_features: List[Union[str, "Feature"]],
+        left_features: List[Union[str, "Feature", Dict]],
         feature_store_name: Optional[str] = None,
         feature_store_id: Optional[int] = None,
         left_feature_group_start_time: Optional[Union[str, int, date, datetime]] = None,

python/hsfs/core/feature_group_api.py

@@ -51,13 +51,17 @@ def save(
             feature_group_instance.feature_store_id,
             "featuregroups",
         ]
+        query_params = {
+            "expand": ["features", "expectationsuite", "transformationfunctions"]
+        }
         headers = {"content-type": "application/json"}
         feature_group_object = feature_group_instance.update_from_response_json(
             _client._send_request(
                 "POST",
                 path_params,
                 headers=headers,
                 data=feature_group_instance.json(),
+                query_params=query_params,
             ),
         )
         return feature_group_object
@@ -93,7 +97,11 @@ def get(
             "featuregroups",
             name,
         ]
-        query_params = None if version is None else {"version": version}
+        query_params = {
+            "expand": ["features", "expectationsuite", "transformationfunctions"]
+        }
+        if version is not None:
+            query_params["version"] = version
 
         fg_objs = []
         # In principle unique names are enforced across fg type and this should therefore
@@ -157,8 +165,10 @@ def get_by_id(
             "featuregroups",
             feature_group_id,
         ]
-
-        fg_json = _client._send_request("GET", path_params)
+        query_params = {
+            "expand": ["features", "expectationsuite", "transformationfunctions"]
+        }
+        fg_json = _client._send_request("GET", path_params, query_params)
         if (
             fg_json["type"] == FeatureGroupApi.BACKEND_FG_STREAM
             or fg_json["type"] == FeatureGroupApi.BACKEND_FG_BATCH

python/hsfs/core/feature_group_engine.py

@@ -88,7 +88,9 @@ def insert(
         validation_options: dict = None,
     ):
         dataframe_features = engine.get_instance().parse_schema_feature_group(
-            feature_dataframe, feature_group.time_travel_format
+            feature_dataframe,
+            feature_group.time_travel_format,
+            feature_group.transformation_functions,
         )
         util.validate_embedding_feature_type(
             feature_group.embedding_index, dataframe_features
@@ -281,7 +283,9 @@ def insert_stream(
             )
 
         dataframe_features = engine.get_instance().parse_schema_feature_group(
-            dataframe, feature_group.time_travel_format
+            dataframe,
+            feature_group.time_travel_format,
+            feature_group.transformation_functions,
         )
         util.validate_embedding_feature_type(
             feature_group.embedding_index, dataframe_features

python/hsfs/core/feature_view_api.py

@@ -17,12 +17,7 @@
 
 from typing import List, Optional, Union
 
-from hsfs import (
-    client,
-    feature_view,
-    training_dataset,
-    transformation_function_attached,
-)
+from hsfs import client, feature_view, training_dataset
 from hsfs.client.exceptions import RestAPIError
 from hsfs.constructor import query, serving_prepared_statement
 from hsfs.core import explicit_provenance, job, training_dataset_job_conf
@@ -78,13 +73,28 @@ def update(self, feature_view_obj: feature_view.FeatureView) -> None:
             data=feature_view_obj.json(),
         )
 
-    def get_by_name(self, name: str) -> feature_view.FeatureView:
+    def get_by_name(self, name: str) -> List[feature_view.FeatureView]:
+        """
+        Get a feature view from the backend using its name.
+
+        # Arguments
+            name `str`: Name of the feature view.
+
+        # Returns
+            `List[FeatureView]`: A list that contains all version of the feature view.
+
+        # Raises
+            `RestAPIError`: If the feature view cannot be found from the backend.
+            `ValueError`: If the feature group associated with the feature view cannot be found.
+        """
         path = self._base_path + [name]
         try:
             return [
                 feature_view.FeatureView.from_response_json(fv)
                 for fv in self._client._send_request(
-                    self._GET, path, {"expand": ["query", "features"]}
+                    self._GET,
+                    path,
+                    {"expand": ["query", "features", "transformationfunctions"]},
                 )["items"]
             ]
         except RestAPIError as e:
@@ -98,11 +108,27 @@ def get_by_name(self, name: str) -> feature_view.FeatureView:
                 raise e
 
     def get_by_name_version(self, name: str, version: int) -> feature_view.FeatureView:
+        """
+        Get a feature view form the backend using both name and version
+
+        # Arguments
+            name `str`: Name of feature view.
+            version `version`: Version of the feature view.
+
+        # Returns
+            `FeatureView`
+
+        # Raises
+            `RestAPIError`: If the feature view cannot be found from the backend.
+            `ValueError`: If the feature group associated with the feature view cannot be found.
+        """
         path = self._base_path + [name, self._VERSION, version]
         try:
             return feature_view.FeatureView.from_response_json(
                 self._client._send_request(
-                    self._GET, path, {"expand": ["query", "features"]}
+                    self._GET,
+                    path,
+                    {"expand": ["query", "features", "transformationfunctions"]},
                 )
             )
         except RestAPIError as e:
@@ -180,17 +206,6 @@ def get_serving_prepared_statement(
             self._client._send_request("GET", path, query_params, headers=headers)
         )
 
-    def get_attached_transformation_fn(
-        self, name: str, version: int
-    ) -> Union[
-        "transformation_function_attached.TransformationFunctionAttached",
-        List["transformation_function_attached.TransformationFunctionAttached"],
-    ]:
-        path = self._base_path + [name, self._VERSION, version, self._TRANSFORMATION]
-        return transformation_function_attached.TransformationFunctionAttached.from_response_json(
-            self._client._send_request("GET", path)
-        )
-
     def create_training_dataset(
         self,
         name: str,