-
Notifications
You must be signed in to change notification settings - Fork 1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Local feature server implementation (HTTP endpoint) (#1780)
* Local feature server implementation (HTTP endpoint) Signed-off-by: Tsotne Tabidze <[email protected]> * Update the request protobufs and hack json/protobuf conversion to avoid extra structure (e.g. "int64_val") in json Signed-off-by: Tsotne Tabidze <[email protected]> * Revert update to the service Signed-off-by: Tsotne Tabidze <[email protected]>
- Loading branch information
Tsotne Tabidze
authored
Aug 21, 2021
1 parent
954565e
commit 745a1b4
Showing
9 changed files
with
427 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| import uvicorn | ||
| from fastapi import FastAPI, HTTPException, Request | ||
| from fastapi.logger import logger | ||
| from google.protobuf.json_format import MessageToDict, Parse | ||
|
|
||
| import feast | ||
| from feast import proto_json | ||
| from feast.protos.feast.serving.ServingService_pb2 import GetOnlineFeaturesRequest | ||
|
|
||
|
|
||
| def get_app(store: "feast.FeatureStore"): | ||
| proto_json.patch() | ||
|
|
||
| app = FastAPI() | ||
|
|
||
| @app.get("/get-online-features/") | ||
| async def get_online_features(request: Request): | ||
| try: | ||
| # Validate and parse the request data into GetOnlineFeaturesRequest Protobuf object | ||
| body = await request.body() | ||
| request_proto = GetOnlineFeaturesRequest() | ||
| Parse(body, request_proto) | ||
|
|
||
| # Initialize parameters for FeatureStore.get_online_features(...) call | ||
| if request_proto.HasField("feature_service"): | ||
| features = store.get_feature_service(request_proto.feature_service) | ||
| else: | ||
| features = list(request_proto.features.val) | ||
|
|
||
| full_feature_names = request_proto.full_feature_names | ||
|
|
||
| batch_sizes = [len(v.val) for v in request_proto.entities.values()] | ||
| num_entities = batch_sizes[0] | ||
| if any(batch_size != num_entities for batch_size in batch_sizes): | ||
| raise HTTPException(status_code=500, detail="Uneven number of columns") | ||
|
|
||
| entity_rows = [ | ||
| {k: v.val[idx] for k, v in request_proto.entities.items()} | ||
| for idx in range(num_entities) | ||
| ] | ||
|
|
||
| response_proto = store.get_online_features( | ||
| features, entity_rows, full_feature_names=full_feature_names | ||
| ).proto | ||
|
|
||
| # Convert the Protobuf object to JSON and return it | ||
| return MessageToDict(response_proto, preserving_proto_field_name=True) | ||
| except Exception as e: | ||
| # Print the original exception on the server side | ||
| logger.exception(e) | ||
| # Raise HTTPException to return the error message to the client | ||
| raise HTTPException(status_code=500, detail=str(e)) | ||
|
|
||
| return app | ||
|
|
||
|
|
||
| def start_server(store: "feast.FeatureStore", port: int): | ||
| app = get_app(store) | ||
| uvicorn.run(app, host="127.0.0.1", port=port) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,196 @@ | ||
| import uuid | ||
| from typing import Any, Callable, Type | ||
|
|
||
| from google.protobuf.json_format import ( # type: ignore | ||
| _WKTJSONMETHODS, | ||
| ParseError, | ||
| _Parser, | ||
| _Printer, | ||
| ) | ||
|
|
||
| from feast.protos.feast.serving.ServingService_pb2 import FeatureList | ||
| from feast.protos.feast.types.Value_pb2 import RepeatedValue, Value | ||
|
|
||
| ProtoMessage = Any | ||
| JsonObject = Any | ||
|
|
||
|
|
||
| def _patch_proto_json_encoding( | ||
| proto_type: Type[ProtoMessage], | ||
| to_json_object: Callable[[_Printer, ProtoMessage], JsonObject], | ||
| from_json_object: Callable[[_Parser, JsonObject, ProtoMessage], None], | ||
| ) -> None: | ||
| """Patch Protobuf JSON Encoder / Decoder for a desired Protobuf type with to_json & from_json methods.""" | ||
| to_json_fn_name = "_" + uuid.uuid4().hex | ||
| from_json_fn_name = "_" + uuid.uuid4().hex | ||
| setattr(_Printer, to_json_fn_name, to_json_object) | ||
| setattr(_Parser, from_json_fn_name, from_json_object) | ||
| _WKTJSONMETHODS[proto_type.DESCRIPTOR.full_name] = [ | ||
| to_json_fn_name, | ||
| from_json_fn_name, | ||
| ] | ||
|
|
||
|
|
||
| def _patch_feast_value_json_encoding(): | ||
| """Patch Protobuf JSON Encoder / Decoder with a Feast Value type. | ||
| This allows encoding the proto object as a native type, without the dummy structural wrapper. | ||
| Here's a before example: | ||
| { | ||
| "value_1": { | ||
| "int64_val": 1 | ||
| }, | ||
| "value_2": { | ||
| "double_list_val": [1.0, 2.0, 3.0] | ||
| }, | ||
| } | ||
| And here's an after example: | ||
| { | ||
| "value_1": 1, | ||
| "value_2": [1.0, 2.0, 3.0] | ||
| } | ||
| """ | ||
|
|
||
| def to_json_object(printer: _Printer, message: ProtoMessage) -> JsonObject: | ||
| which = message.WhichOneof("val") | ||
| # If the Value message is not set treat as null_value when serialize | ||
| # to JSON. The parse back result will be different from original message. | ||
| if which is None or which == "null_val": | ||
| return None | ||
| elif "_list_" in which: | ||
| value = list(getattr(message, which).val) | ||
| else: | ||
| value = getattr(message, which) | ||
| return value | ||
|
|
||
| def from_json_object( | ||
| parser: _Parser, value: JsonObject, message: ProtoMessage | ||
| ) -> None: | ||
| if value is None: | ||
| message.null_val = 0 | ||
| elif isinstance(value, bool): | ||
| message.bool_val = value | ||
| elif isinstance(value, str): | ||
| message.string_val = value | ||
| elif isinstance(value, int): | ||
| message.int64_val = value | ||
| elif isinstance(value, float): | ||
| message.double_val = value | ||
| elif isinstance(value, list): | ||
| if len(value) == 0: | ||
| # Clear will mark the struct as modified so it will be created even if there are no values | ||
| message.int64_list_val.Clear() | ||
| elif isinstance(value[0], bool): | ||
| message.bool_list_val.val.extend(value) | ||
| elif isinstance(value[0], str): | ||
| message.string_list_val.val.extend(value) | ||
| elif isinstance(value[0], (float, int, type(None))): | ||
| # Identify array as ints if all of the elements are ints | ||
| if all(isinstance(item, int) for item in value): | ||
| message.int64_list_val.val.extend(value) | ||
| # If any of the elements are floats or nulls, then parse it as a float array | ||
| else: | ||
| # Convert each null as NaN. | ||
| message.double_list_val.val.extend( | ||
| [item if item is not None else float("nan") for item in value] | ||
| ) | ||
| else: | ||
| raise ParseError( | ||
| "Value {0} has unexpected type {1}.".format( | ||
| value[0], type(value[0]) | ||
| ) | ||
| ) | ||
| else: | ||
| raise ParseError( | ||
| "Value {0} has unexpected type {1}.".format(value, type(value)) | ||
| ) | ||
|
|
||
| _patch_proto_json_encoding(Value, to_json_object, from_json_object) | ||
|
|
||
|
|
||
| def _patch_feast_repeated_value_json_encoding(): | ||
| """Patch Protobuf JSON Encoder / Decoder with a Feast RepeatedValue type. | ||
| This allows list of lists without dummy field name "val". | ||
| Here's a before example: | ||
| { | ||
| "repeated_value": [ | ||
| {"val": [1,2,3]}, | ||
| {"val": [4,5,6]} | ||
| ] | ||
| } | ||
| And here's an after example: | ||
| { | ||
| "repeated_value": [ | ||
| [1,2,3], | ||
| [4,5,6] | ||
| ] | ||
| } | ||
| """ | ||
|
|
||
| def to_json_object(printer: _Printer, message: ProtoMessage) -> JsonObject: | ||
| return [printer._MessageToJsonObject(item) for item in message.val] | ||
|
|
||
| def from_json_object( | ||
| parser: _Parser, value: JsonObject, message: ProtoMessage | ||
| ) -> None: | ||
| array = value if isinstance(value, list) else value["val"] | ||
| for item in array: | ||
| parser.ConvertMessage(item, message.val.add()) | ||
|
|
||
| _patch_proto_json_encoding(RepeatedValue, to_json_object, from_json_object) | ||
|
|
||
|
|
||
| def _patch_feast_feature_list_json_encoding(): | ||
| """Patch Protobuf JSON Encoder / Decoder with a Feast FeatureList type. | ||
| This allows list of lists without dummy field name "features". | ||
| Here's a before example: | ||
| { | ||
| "feature_list": { | ||
| "features": [ | ||
| "feature-1", | ||
| "feature-2", | ||
| "feature-3" | ||
| ] | ||
| } | ||
| } | ||
| And here's an after example: | ||
| { | ||
| "feature_list": [ | ||
| "feature-1", | ||
| "feature-2", | ||
| "feature-3" | ||
| ] | ||
| } | ||
| """ | ||
|
|
||
| def to_json_object(printer: _Printer, message: ProtoMessage) -> JsonObject: | ||
| return list(message.val) | ||
|
|
||
| def from_json_object( | ||
| parser: _Parser, value: JsonObject, message: ProtoMessage | ||
| ) -> None: | ||
| array = value if isinstance(value, list) else value["val"] | ||
| message.val.extend(array) | ||
|
|
||
| _patch_proto_json_encoding(FeatureList, to_json_object, from_json_object) | ||
|
|
||
|
|
||
| def patch(): | ||
| """Patch Protobuf JSON Encoder / Decoder with all desired Feast types.""" | ||
| _patch_feast_value_json_encoding() | ||
| _patch_feast_repeated_value_json_encoding() | ||
| _patch_feast_feature_list_json_encoding() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.