Skip to content

Add support for Redis and Redis Cluster #1511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
feast-ci-bot merged 22 commits into from
Jun 9, 2021
Merged

Add support for Redis and Redis Cluster #1511

Changes from 1 commit
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
d97aaea
Add support for Redis as online store
qooba Apr 25, 2021
ae9ed8e
Add support for Redis as online store
qooba Apr 25, 2021
e165f00
Add redis provider
qooba Apr 26, 2021
df8f3cb
Add redis online provider
qooba Apr 27, 2021
6bea80d
Add redis online provider
qooba Apr 27, 2021
5745ff7
Add redis online provider
qooba Apr 27, 2021
a3c4f3b
Add redis online provider
qooba Apr 28, 2021
e434e59
Add redis online provider - integration tests
qooba May 6, 2021
8ccfc6b
Add redis online provider - integration tests
qooba May 6, 2021
be9a17f
Add redis provider
qooba May 12, 2021
7606fe7
Add redis provider
qooba May 12, 2021
2a95e3e
correct redis provider
qooba May 24, 2021
435c856
correct redis provider
qooba May 26, 2021
d329439
correct redis provider test
qooba May 26, 2021
b5269e7
correct redis provider
qooba May 30, 2021
f602b9a
correct redis provider
qooba May 30, 2021
bbde0b0
change redis connection string
qooba Jun 2, 2021
d86f23c
change redis connection string
qooba Jun 2, 2021
b0c8101
fix redis materialize
qooba Jun 7, 2021
52232cc
Rename connection string and set defaults
woop Jun 8, 2021
53f8f72
Fix argument in redis test
woop Jun 9, 2021
e12ecff
Fix broken connection string
woop Jun 9, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Add redis online provider 
Signed-off-by: qooba <dev@qooba.net>
  • Loading branch information
@qooba @woop
qooba authored and woop committed Jun 9, 2021
commit a3c4f3bf464c416371b7f0617b52562357daccda
64 changes: 32 additions & 32 deletions sdk/python/feast/infra/redis_provider.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
from feast import FeatureTable, utils
from feast.entity import Entity
from feast.feature_view import FeatureView
from feast.infra.key_encoding_utils import serialize_entity_key
from feast.infra.offline_stores.helpers import get_offline_store_from_sources
from feast.infra.provider import (
Provider,
Expand Down Expand Up @@ -47,7 +46,7 @@ def update_infra(
entities_to_keep: Sequence[Entity],
partial: bool,
):
client = self._get_client()
pass

def teardown_infra(
self,
Expand Down Expand Up @@ -80,9 +79,8 @@ def online_write_batch(

for entity_key, values, timestamp, created_ts in data:
redis_key_bin = _redis_key(project, entity_key)
timestamp = int(utils.make_tzaware(timestamp).timestamp())
ts = Timestamp()
ts.seconds = timestamp
ts.seconds = int(utils.make_tzaware(timestamp).timestamp())
entity_hset[f"_ts:{feature_view}"] = ts.SerializeToString()
entity_hset[f"_ex:{feature_view}"] = ex_str

Expand All @@ -105,31 +103,33 @@ def online_read(

result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []

for entity_key in entity_keys:
redis_key_bin = _redis_key(project, entity_key)
hset_keys = [_mmh3(f"{feature_view}:{k}") for k in requested_features]
ts_key = f"_ts:{feature_view}"
hset_keys.append(ts_key)
values = client.hmget(redis_key_bin, hset_keys)
requested_features.append(ts_key)
res_val = dict(zip(requested_features, values))

res_ts = Timestamp()
ts_val = res_val.pop(ts_key)
if ts_val:
res_ts.ParseFromString(ts_val)

res = {}
for feature_name, val_bin in res_val.items():
val = ValueProto()
if val_bin:
val.ParseFromString(val_bin)
res[feature_name] = val

if not res:
result.append((None, None))
else:
result.append((res_ts, res))
if requested_features:
for entity_key in entity_keys:
redis_key_bin = _redis_key(project, entity_key)
hset_keys = [_mmh3(f"{feature_view}:{k}") for k in requested_features]
ts_key = f"_ts:{feature_view}"
hset_keys.append(ts_key)
values = client.hmget(redis_key_bin, hset_keys)
requested_features.append(ts_key)
res_val = dict(zip(requested_features, values))

res_ts = Timestamp()
ts_val = res_val.pop(ts_key)
if ts_val:
res_ts.ParseFromString(ts_val)

res = {}
for feature_name, val_bin in res_val.items():
val = ValueProto()
if val_bin:
val.ParseFromString(val_bin)
res[feature_name] = val

if not res:
result.append((None, None))
else:
timestamp = datetime.fromtimestamp(res_ts.seconds)
result.append((timestamp, res))
return result

def materialize_single_feature_view(
Expand Down Expand Up @@ -188,7 +188,7 @@ def _get_cs(self):
startup_nodes = [
dict(zip(["host", "port"], c.split(":")))
for c in connection_string.split(",")
if not "=" in c
if "=" not in c
]
params = {}
for c in connection_string.split(","):
Expand Down Expand Up @@ -240,7 +240,7 @@ def get_historical_features(
)


def _redis_key(project: str, entity_key: EntityKeyProto) -> str:
def _redis_key(project: str, entity_key: EntityKeyProto):
redis_key = RedisKeyProto(
project=project,
entity_names=entity_key.join_keys,
Expand All @@ -250,7 +250,7 @@ def _redis_key(project: str, entity_key: EntityKeyProto) -> str:
return redis_key.SerializeToString()


def _mmh3(key: str) -> str:
def _mmh3(key: str):
"""
Calculate murmur3_32 hash which is equal to scala version which is using little endian:
https://stackoverflow.com/questions/29932956/murmur3-hash-different-result-between-python-and-java-implementation
Expand Down