Skip to content

Commit fd27ee1

Browse files
Ly Caoachals
Ly Cao
authored and
achals
committed
fixed python connector error + add a script to build plugin binary
Signed-off-by: Felix Wang <wangfelix98@gmail.com> Signed-off-by: Achal Shah <achals@gmail.com>
1 parent 16fb7d3 commit fd27ee1

4 files changed

Lines changed: 87 additions & 109 deletions

File tree

‎go/test_repo/Makefile‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
PYTHON_VERSION ?= 3.7.10
2+
PATH_TO_FILE ?= ~/.pyenv/versions/${PYTHON_VERSION}
3+
4+
all: create-plugin-binary
5+
6+
create-plugin-binary:
7+
# If package is installed, remove and reinstall with --enable-framework
8+
ifeq ("$(wildcard "${PATH_TO_FILE}")","")
9+
echo "${PATH_TO_FILE} exists"
10+
pyenv uninstall ${PYTHON_VERSION}
11+
endif
12+
env PYTHON_CONFIGURE_OPTS="--enable-framework" pyenv install ${PYTHON_VERSION}
13+
pip install pyinstaller
14+
pip install .
15+
sudo pyinstaller plugin.py --onefile

‎go/test_repo/feature_store.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ project: test_repo
33
provider: local
44
online_store:
55
type: connector
6-
KV_PLUGIN: python plugin.py
6+
KV_PLUGIN: ./dist/plugin
77
offline_store:
88
type: file
99
flags:

‎go/test_repo/plugin.py‎

100755100644
Lines changed: 38 additions & 108 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,32 @@
11
from concurrent import futures
2+
import concurrent.futures
23
import sys
34
import time
5+
from datetime import datetime
46

57
import grpc
8+
import os
69

710
from grpc_health.v1.health import HealthServicer
811
from grpc_health.v1 import health_pb2, health_pb2_grpc
912

1013
from connector_python import Connector_pb2_grpc
1114
from connector_python import Connector_pb2
1215
from connector_python import ServingService_pb2
16+
from connector_python.ServingService_pb2 import FeatureReferenceV2 as FeatureReferenceV2Proto
17+
from connector_python.Connector_pb2 import ConnectorFeature as ConnectorFeatureProto
18+
from connector_python.Connector_pb2 import ConnectorFeatureList as ConnectorFeatureListProto
1319
from connector_python.ServingService_pb2 import FeatureList as FeatureListProto
1420
from connector_python.EntityKey_pb2 import EntityKey as EntityKeyProto
1521
from connector_python.Value_pb2 import Value as ValueProto
22+
from connector_python.Value_pb2 import ValueType
1623
from google.protobuf.timestamp_pb2 import Timestamp
1724

25+
import typing
1826
from typing import (
1927
Any,
2028
ByteString,
29+
Callable,
2130
Dict,
2231
List,
2332
Optional,
@@ -31,18 +40,6 @@
3140
import mmh3
3241
import struct
3342

34-
# sdk/python/feast/usage.py
35-
import contextlib
36-
import contextvars
37-
import dataclasses
38-
import os
39-
import typing
40-
import uuid
41-
from datetime import datetime
42-
43-
sys.path.append("/Users/lycao/Documents/feast/go/test_repo/connector_python")
44-
45-
# sdk/python/feast/infra/online_stores/helpers.py
4643
def _redis_key(project: str, entity_key: EntityKeyProto) -> bytes:
4744
key: List[bytes] = [serialize_entity_key(entity_key), project.encode("utf-8")]
4845
return b"".join(key)
@@ -84,76 +81,17 @@ def serialize_entity_key(entity_key: EntityKeyProto) -> bytes:
8481

8582
return b"".join(output)
8683

87-
# sdk/python/feast/usage.py
88-
89-
@dataclasses.dataclass
90-
class FnCall:
91-
fn_name: str
92-
id: str
93-
94-
start: datetime
95-
end: typing.Optional[datetime] = None
96-
97-
parent_id: typing.Optional[str] = None
98-
99-
100-
class Sampler:
101-
def should_record(self, event) -> bool:
102-
raise NotImplementedError
103-
104-
@property
105-
def priority(self):
106-
return 0
107-
108-
109-
class AlwaysSampler(Sampler):
110-
def should_record(self, event) -> bool:
111-
return True
112-
113-
114-
class UsageContext:
115-
attributes: typing.Dict[str, typing.Any]
116-
117-
call_stack: typing.List[FnCall]
118-
completed_calls: typing.List[FnCall]
119-
120-
exception: typing.Optional[Exception] = None
121-
traceback: typing.Optional[typing.Tuple[str, int, str]] = None
122-
123-
sampler: Sampler = AlwaysSampler()
124-
125-
def __init__(self):
126-
self.attributes = {}
127-
self.call_stack = []
128-
self.completed_calls = []
129-
130-
131-
_context = contextvars.ContextVar("usage_context", default=UsageContext())
132-
133-
@contextlib.contextmanager
134-
def tracing_span(name):
135-
"""
136-
Context manager for wrapping heavy parts of code in tracing span
137-
"""
138-
if _is_enabled:
139-
ctx = _context.get()
140-
if not ctx.call_stack:
141-
raise RuntimeError("tracing_span must be called in usage context")
142-
143-
last_call = ctx.call_stack[-1]
144-
fn_call = FnCall(
145-
id=uuid.uuid4().hex,
146-
parent_id=last_call.id,
147-
fn_name=f"{last_call.fn_name}.{name}",
148-
start=datetime.utcnow(),
149-
)
150-
try:
151-
yield
152-
finally:
153-
if _is_enabled:
154-
fn_call.end = datetime.utcnow()
155-
ctx.completed_calls.append(fn_call)
156-
84+
def _serialize_val(value_type, v: ValueProto) -> Tuple[bytes, int]:
85+
if value_type == "string_val":
86+
return v.string_val.encode("utf8"), ValueType.STRING
87+
elif value_type == "bytes_val":
88+
return v.bytes_val, ValueType.BYTES
89+
elif value_type == "int32_val":
90+
return struct.pack("<i", v.int32_val), ValueType.INT32
91+
elif value_type == "int64_val":
92+
return struct.pack("<l", v.int64_val), ValueType.INT64
93+
else:
94+
raise ValueError(f"Value type not supported for Firestore: {v}")
15795

15896
# sdk/python/feast/infra/online_stores/redis.py
15997
class ConnectorOnlineStore(Connector_pb2_grpc.OnlineStoreServicer):
@@ -165,62 +103,54 @@ def __init__(self, project, host: str, port: str):
165103
self.port = port
166104

167105
def OnlineRead(self, request, context):
168-
response = {'results': [[]]}
106+
response = Connector_pb2.OnlineReadResponse()
169107

170-
feature_view = request.View
171-
project = config.project
172-
requested_features = request.Features
173-
entity_keys = request.EntityKeys
108+
feature_view = request.view
109+
project = self.project
110+
requested_features = request.features
111+
entity_keys = request.entityKeys
174112

175113
hset_keys = [_mmh3(f"{feature_view}:{k}") for k in requested_features]
176114

177115
ts_key = f"_ts:{feature_view}"
178116
hset_keys.append(ts_key)
179117
requested_features.append(ts_key)
180-
181-
keys = []
182-
for entity_key in entity_keys:
118+
results : List[ConnectorFeatureListProto] = [None] * len(entity_keys)
119+
keys = [None] * len(entity_keys)
120+
for index, entity_key in enumerate(entity_keys):
183121
redis_key_bin = _redis_key(self.project, entity_key)
184-
keys.append(redis_key_bin)
185-
with self.client.pipeline() as pipe:
186-
for redis_key_bin in keys:
187-
pipe.hmget(redis_key_bin, hset_keys)
188-
# TODO
189-
with tracing_span(name="remote_call"):
190-
redis_values = pipe.execute()
191-
for values in redis_values:
122+
keys[index] = redis_key_bin
123+
values = self._client.hmget(redis_key_bin, hset_keys)
192124
feature_list = self._get_features_for_entity(
193125
values, feature_view, requested_features
194126
)
195-
response['results'].append(feature_list)
196-
return response
127+
results[index] = feature_list
128+
return Connector_pb2.OnlineReadResponse(results=results)
197129

198130
def _get_features_for_entity(
199131
self,
200132
values: List[ByteString],
201133
feature_view: str,
202134
requested_features: List[str],
203-
) -> Optional[List[FeatureListProto]]:
135+
) -> ConnectorFeatureListProto:
204136

205137
res_val = dict(zip(requested_features, values))
206-
207138
res_ts = Timestamp()
208139
ts_val = res_val.pop(f"_ts:{feature_view}")
209140
if ts_val:
210141
res_ts.ParseFromString(bytes(ts_val))
211-
timestamp = datetime.fromtimestamp(res_ts.seconds)
212-
feature_list = [None] * len(requested_features)
142+
feature_list = [None] * len(res_val)
213143

214144
index = 0
215145
for feature_name, val_bin in res_val.items():
216146
val = ValueProto()
217147
if val_bin:
218148
val.ParseFromString(bytes(val_bin))
219-
feature_ref = {'feature_view_name': feature_view, 'feature_name': feature_name} #ServingService_pb2.FeatureReferenceV2()
220-
feature_list[index] = {'timestamp': timestamp, 'reference': feature_ref, 'value': val}
149+
feature_ref = FeatureReferenceV2Proto(feature_view_name=feature_view, feature_name=feature_name)
150+
feature_list[index] = ConnectorFeatureProto(timestamp=res_ts, reference=feature_ref, value=val)
221151
index += 1
222152

223-
return feature_list
153+
return ConnectorFeatureListProto(featureList= feature_list)
224154

225155

226156
def serve():
@@ -230,7 +160,7 @@ def serve():
230160

231161
# Start the server.
232162
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
233-
Connector_pb2_grpc.add_OnlineStoreServicer_to_server(ConnectorOnlineStore(project="test_repo", host=":", port="6379"), server)
163+
Connector_pb2_grpc.add_OnlineStoreServicer_to_server(ConnectorOnlineStore(project="test_repo", host="localhost", port="6379"), server)
234164
health_pb2_grpc.add_HealthServicer_to_server(health, server)
235165
server.add_insecure_port('127.0.0.1:1234')
236166
server.start()

‎go/test_repo/setup.py‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
from setuptools import setup
2+
from setuptools import find_packages
3+
4+
NAME = "Python Redis Connector"
5+
DESCRIPTION = "Python SDK for Feast"
6+
URL = "https://github.com/feast-dev/feast/go/test_repo"
7+
AUTHOR = "Feast"
8+
REQUIRES_PYTHON = ">=3.7.0"
9+
10+
REQUIRED = [
11+
"grpcio",
12+
"grpcio-health-checking",
13+
"mmh3",
14+
"redis",
15+
"protobuf"
16+
]
17+
18+
setup(
19+
name=NAME,
20+
author=AUTHOR,
21+
description=DESCRIPTION,
22+
python_requires=REQUIRES_PYTHON,
23+
url=URL,
24+
packages=["connector_python"],
25+
install_requires=REQUIRED,
26+
classifiers=[
27+
"License :: OSI Approved :: MIT License",
28+
"Programming Language :: Python",
29+
"Programming Language :: Python :: 3",
30+
"Programming Language :: Python :: 3.7",
31+
],
32+
entry_points={"console_scripts": ["plugin=test_repo.plugin:server"]},
33+
)

0 commit comments

Comments
 (0)