FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
feast/sdk/python/feast/feature_service.py at featureview-versioning · feast-dev/feast · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
feast-dev
/
feast
Public
Notifications
You must be signed in to change notification settings
Fork
1.4k
Star
7.2k
Code
Issues
218
Pull requests
191
Discussions
Actions
Security and quality
1
Insights
Additional navigation options
Code
Issues
Pull requests
Discussions
Actions
Security and quality
Insights
Expand file tree
Breadcrumbs
feast
/
sdk
/
python
/
feast
/
feature_service.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
253 lines (221 loc) · 9.96 KB
Breadcrumbs
feast
/
sdk
/
python
/
feast
/
feature_service.py
Copy path
File metadata and controls
253 lines (221 loc) · 9.96 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
from
datetime
import
datetime
from
typing
import
Dict
,
List
,
Optional
,
Union
from
google
.
protobuf
.
json_format
import
MessageToJson
from
typeguard
import
typechecked
from
feast
.
base_feature_view
import
BaseFeatureView
from
feast
.
errors
import
FeatureViewMissingDuringFeatureServiceInference
from
feast
.
feature_logging
import
LoggingConfig
from
feast
.
feature_view
import
FeatureView
from
feast
.
feature_view_projection
import
FeatureViewProjection
from
feast
.
on_demand_feature_view
import
OnDemandFeatureView
from
feast
.
protos
.
feast
.
core
.
FeatureService_pb2
import
(
FeatureService
as
FeatureServiceProto
,
)
from
feast
.
protos
.
feast
.
core
.
FeatureService_pb2
import
(
FeatureServiceMeta
as
FeatureServiceMetaProto
,
)
from
feast
.
protos
.
feast
.
core
.
FeatureService_pb2
import
(
FeatureServiceSpec
as
FeatureServiceSpecProto
,
)
@
typechecked
class
FeatureService
:
"""
A feature service defines a logical group of features from one or more feature views.
This group of features can be retrieved together during training or serving.
Attributes:
name: The unique name of the feature service.
feature_view_projections: A list containing feature views and feature view
projections, representing the features in the feature service.
description: A human-readable description.
tags: A dictionary of key-value pairs to store arbitrary metadata.
owner: The owner of the feature service, typically the email of the primary
maintainer.
created_timestamp: The time when the feature service was created.
last_updated_timestamp: The time when the feature service was last updated.
"""
name
:
str
_features
:
List
[
Union
[
FeatureView
,
OnDemandFeatureView
]]
feature_view_projections
:
List
[
FeatureViewProjection
]
description
:
str
tags
:
Dict
[
str
,
str
]
owner
:
str
created_timestamp
:
Optional
[
datetime
]
=
None
last_updated_timestamp
:
Optional
[
datetime
]
=
None
logging_config
:
Optional
[
LoggingConfig
]
=
None
def
__init__
(
self
,
*
,
name
:
str
,
features
:
List
[
Union
[
FeatureView
,
OnDemandFeatureView
]],
tags
:
Optional
[
Dict
[
str
,
str
]]
=
None
,
description
:
str
=
""
,
owner
:
str
=
""
,
logging_config
:
Optional
[
LoggingConfig
]
=
None
,
):
"""
Creates a FeatureService object.
Args:
name: The unique name of the feature service.
feature_view_projections: A list containing feature views and feature view
projections, representing the features in the feature service.
description (optional): A human-readable description.
tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
owner (optional): The owner of the feature view, typically the email of the
primary maintainer.
"""
self
.
name
=
name
self
.
_features
=
features
self
.
feature_view_projections
=
[]
self
.
description
=
description
self
.
tags
=
tags
or
{}
self
.
owner
=
owner
self
.
created_timestamp
=
None
self
.
last_updated_timestamp
=
None
self
.
logging_config
=
logging_config
for
feature_grouping
in
self
.
_features
:
if
isinstance
(
feature_grouping
,
BaseFeatureView
):
self
.
feature_view_projections
.
append
(
feature_grouping
.
projection
)
def
infer_features
(
self
,
fvs_to_update
:
Dict
[
str
,
Union
[
FeatureView
,
BaseFeatureView
]]
):
"""
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 of feature view names to corresponding feature views that
contains all the feature views necessary to run inference.
"""
for
feature_grouping
in
self
.
_features
:
if
isinstance
(
feature_grouping
,
BaseFeatureView
):
projection
=
feature_grouping
.
projection
if
projection
.
desired_features
:
# The projection wants to select a specific set of inferred features.
# Example: FeatureService(features=[fv[["inferred_feature"]]]), where
# 'fv' is a feature view that was defined without a schema.
if
feature_grouping
.
name
in
fvs_to_update
:
# First we validate that the selected features have actually been inferred.
desired_features
=
set
(
projection
.
desired_features
)
actual_features
=
set
(
[
f
.
name
for
f
in
fvs_to_update
[
feature_grouping
.
name
].
features
]
)
assert
desired_features
.
issubset
(
actual_features
)
# Then we extract the selected features and add them to the projection.
projection
.
features
=
[]
for
f
in
fvs_to_update
[
feature_grouping
.
name
].
features
:
if
f
.
name
in
desired_features
:
projection
.
features
.
append
(
f
)
else
:
raise
FeatureViewMissingDuringFeatureServiceInference
(
feature_view_name
=
feature_grouping
.
name
,
feature_service_name
=
self
.
name
,
)
continue
if
projection
.
features
:
# The projection has already selected features from a feature view with a
# known schema, so no action needs to be taken.
# Example: FeatureService(features=[fv[["existing_feature"]]]), where
# 'existing_feature' was defined as part of the schema of 'fv'.
# Example: FeatureService(features=[fv]), where 'fv' was defined with a schema.
continue
# The projection wants to select all possible inferred features.
# Example: FeatureService(features=[fv]), where 'fv' is a feature view that
# was defined without a schema.
if
feature_grouping
.
name
in
fvs_to_update
:
projection
.
features
=
fvs_to_update
[
feature_grouping
.
name
].
features
else
:
raise
FeatureViewMissingDuringFeatureServiceInference
(
feature_view_name
=
feature_grouping
.
name
,
feature_service_name
=
self
.
name
,
)
else
:
raise
ValueError
(
f"The feature service
{
self
.
name
}
has been provided with an invalid type "
f'
{
type
(
feature_grouping
)
}
as part of the "features" argument.)'
)
def
__repr__
(
self
):
items
=
(
f"
{
k
}
=
{
v
}
"
for
k
,
v
in
self
.
__dict__
.
items
())
return
f"<
{
self
.
__class__
.
__name__
}
(
{
', '
.
join
(
items
)
}
)>"
def
__str__
(
self
):
return
str
(
MessageToJson
(
self
.
to_proto
()))
def
__hash__
(
self
):
return
hash
(
self
.
name
)
def
__eq__
(
self
,
other
):
if
not
isinstance
(
other
,
FeatureService
):
raise
TypeError
(
"Comparisons should only involve FeatureService class objects."
)
if
(
self
.
name
!=
other
.
name
or
self
.
description
!=
other
.
description
or
self
.
tags
!=
other
.
tags
or
self
.
owner
!=
other
.
owner
):
return
False
if
sorted
(
self
.
feature_view_projections
)
!=
sorted
(
other
.
feature_view_projections
):
return
False
return
True
@
classmethod
def
from_proto
(
cls
,
feature_service_proto
:
FeatureServiceProto
):
"""
Converts a FeatureServiceProto to a FeatureService object.
Args:
feature_service_proto: A protobuf representation of a FeatureService.
"""
fs
=
cls
(
name
=
feature_service_proto
.
spec
.
name
,
features
=
[],
tags
=
dict
(
feature_service_proto
.
spec
.
tags
),
description
=
feature_service_proto
.
spec
.
description
,
owner
=
feature_service_proto
.
spec
.
owner
,
logging_config
=
LoggingConfig
.
from_proto
(
feature_service_proto
.
spec
.
logging_config
),
)
fs
.
feature_view_projections
.
extend
(
[
FeatureViewProjection
.
from_proto
(
projection
)
for
projection
in
feature_service_proto
.
spec
.
features
]
)
if
feature_service_proto
.
meta
.
HasField
(
"created_timestamp"
):
fs
.
created_timestamp
=
(
feature_service_proto
.
meta
.
created_timestamp
.
ToDatetime
()
)
if
feature_service_proto
.
meta
.
HasField
(
"last_updated_timestamp"
):
fs
.
last_updated_timestamp
=
(
feature_service_proto
.
meta
.
last_updated_timestamp
.
ToDatetime
()
)
return
fs
def
to_proto
(
self
)
->
FeatureServiceProto
:
"""
Converts a feature service to its protobuf representation.
Returns:
A FeatureServiceProto protobuf.
"""
meta
=
FeatureServiceMetaProto
()
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
)
spec
=
FeatureServiceSpecProto
(
name
=
self
.
name
,
features
=
[
projection
.
to_proto
()
for
projection
in
self
.
feature_view_projections
],
tags
=
self
.
tags
,
description
=
self
.
description
,
owner
=
self
.
owner
,
logging_config
=
self
.
logging_config
.
to_proto
()
if
self
.
logging_config
else
None
,
)
return
FeatureServiceProto
(
spec
=
spec
,
meta
=
meta
)
def
validate
(
self
):
pass
Back
|
FazBrowse Home
|
New Git URL