Infers the features for the projections of this feature service, and updates this feature service in place. This method is necessary since feature services may rely on feature views which require feature inference. Args: fvs_to_update: A mapping
(
self, fvs_to_update: Dict[str, Union[FeatureView, BaseFeatureView]]
)
| 92 | self.feature_view_projections.append(feature_grouping.projection) |
| 93 | |
| 94 | def infer_features( |
| 95 | self, fvs_to_update: Dict[str, Union[FeatureView, BaseFeatureView]] |
| 96 | ): |
| 97 | """ |
| 98 | Infers the features for the projections of this feature service, and updates this feature |
| 99 | service in place. |
| 100 | |
| 101 | This method is necessary since feature services may rely on feature views which require |
| 102 | feature inference. |
| 103 | |
| 104 | Args: |
| 105 | fvs_to_update: A mapping of feature view names to corresponding feature views that |
| 106 | contains all the feature views necessary to run inference. |
| 107 | """ |
| 108 | for feature_grouping in self._features: |
| 109 | if isinstance(feature_grouping, BaseFeatureView): |
| 110 | projection = feature_grouping.projection |
| 111 | |
| 112 | if projection.desired_features: |
| 113 | # The projection wants to select a specific set of inferred features. |
| 114 | # Example: FeatureService(features=[fv[["inferred_feature"]]]), where |
| 115 | # 'fv' is a feature view that was defined without a schema. |
| 116 | if feature_grouping.name in fvs_to_update: |
| 117 | # First we validate that the selected features have actually been inferred. |
| 118 | desired_features = set(projection.desired_features) |
| 119 | actual_features = set( |
| 120 | [ |
| 121 | f.name |
| 122 | for f in fvs_to_update[feature_grouping.name].features |
| 123 | ] |
| 124 | ) |
| 125 | assert desired_features.issubset(actual_features) |
| 126 | |
| 127 | # Then we extract the selected features and add them to the projection. |
| 128 | projection.features = [] |
| 129 | for f in fvs_to_update[feature_grouping.name].features: |
| 130 | if f.name in desired_features: |
| 131 | projection.features.append(f) |
| 132 | else: |
| 133 | raise FeatureViewMissingDuringFeatureServiceInference( |
| 134 | feature_view_name=feature_grouping.name, |
| 135 | feature_service_name=self.name, |
| 136 | ) |
| 137 | |
| 138 | continue |
| 139 | |
| 140 | if projection.features: |
| 141 | # The projection has already selected features from a feature view with a |
| 142 | # known schema, so no action needs to be taken. |
| 143 | # Example: FeatureService(features=[fv[["existing_feature"]]]), where |
| 144 | # 'existing_feature' was defined as part of the schema of 'fv'. |
| 145 | # Example: FeatureService(features=[fv]), where 'fv' was defined with a schema. |
| 146 | continue |
| 147 | |
| 148 | # The projection wants to select all possible inferred features. |
| 149 | # Example: FeatureService(features=[fv]), where 'fv' is a feature view that |
| 150 | # was defined without a schema. |
| 151 | if feature_grouping.name in fvs_to_update: |