Source code for feast.infra.online_stores.helpers

import struct
from typing import Any, List

import mmh3

from feast.importer import import_class
from feast.infra.key_encoding_utils import (
from feast.infra.online_stores.online_store import OnlineStore
from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto

[docs]def get_online_store_from_config(online_store_config: Any) -> OnlineStore: """Creates an online store corresponding to the given online store config.""" module_name = online_store_config.__module__ qualified_name = type(online_store_config).__name__ class_name = qualified_name.replace("Config", "") online_store_class = import_class(module_name, class_name, "OnlineStore") return online_store_class()
def _redis_key( project: str, entity_key: EntityKeyProto, entity_key_serialization_version=1 ) -> bytes: key: List[bytes] = [ serialize_entity_key( entity_key, entity_key_serialization_version=entity_key_serialization_version, ), project.encode("utf-8"), ] return b"".join(key) def _redis_key_prefix(entity_keys: List[str]) -> bytes: return serialize_entity_key_prefix(entity_keys) def _mmh3(key: str): """ Calculate murmur3_32 hash which is equal to scala version which is using little endian: """ key_hash = mmh3.hash(key, signed=False) return bytes.fromhex(struct.pack("<Q", key_hash).hex()[:8])
[docs]def compute_entity_id( entity_key: EntityKeyProto, entity_key_serialization_version=1 ) -> str: """ Compute Entity id given Feast Entity Key for online stores. Remember that Entity here refers to `EntityKeyProto` which is used in some online stores to encode the keys. It has nothing to do with the Entity concept we have in Feast. """ return mmh3.hash_bytes( serialize_entity_key( entity_key, entity_key_serialization_version=entity_key_serialization_version, ) ).hex()