diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..9414382 --- /dev/null +++ b/.dockerignore @@ -0,0 +1 @@ +Dockerfile diff --git a/.env b/.env new file mode 100644 index 0000000..28e1749 --- /dev/null +++ b/.env @@ -0,0 +1,2 @@ +DKUBE_IP="" +DKUBE_TOKEN="" diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d9005f2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,152 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintainted in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..ef77ae7 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "feast"] + path = feast + url = https://github.com/feast-dev/feast.git diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..1bc79e3 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,33 @@ +FROM python:3.9.10-slim + +WORKDIR /usr/src/feast + +ENV PYTHONDONTWRITEBYTECODE 1 +ENV PYTHONUNBUFFERED 1 +ENV PYTHONPATH=$PYTHONPATH:/usr/src/feast:/usr/src/feast/provider/sdk + +RUN apt-get update \ + && apt-get -y install gcc git build-essential procps net-tools \ + && apt-get clean \ + && pip install --no-cache-dir -U pip \ + && rm -rf /var/lib/apt/lists/* + +COPY . . + +RUN pip install --no-cache-dir -r ./online_server/requirements.txt + +RUN git submodule init && git submodule update + +COPY setup.py ./feast/sdk/python/setup.py + +RUN cd ./feast/ && patch -p1 < ../feast_ol_server.patch && make install-python + +RUN pip3 install git+https://github.com/oneconvergence/dkube.git@feast_changes --upgrade + +WORKDIR /usr/src/feast/online_server + +RUN addgroup --system olserver && adduser --system --group olserver + +USER olserver + +CMD ["python", "server.py"] diff --git a/README.md b/README.md index 3ff1382..430ef83 100644 --- a/README.md +++ b/README.md @@ -1 +1,42 @@ # feast + +## Run online service as Knative service + + 1. Build image + git clone https://github.com/oneconvergence/feast.git + git checkout -b online_server_changes origin/online_server_changes + cd feast/ + docker build -i . + + 2. Run Knative service + Edit file: + online_server/online_svc.yaml + kubectl apply -f online_server/online_svc.yaml + kubectl get ksvc + + +## Installation steps to run local dev environment + +- Pre-requisite - Setup a virtual environment with python 3.9.7 or above. +- Steps: + + git clone https://github.com/oneconvergence/feast.git + git checkout -b online_server_changes origin/online_server_changes + cd feast/feast + git submodule init + git submodule update + patch -p1 < ../feast_ol_server.patch + make install-python + +- Set PYTHONPATH: + + export BASE= + export PYTHONPATH=$BASE:$PYTHONPATH + export PYTHONPATH=$BASE/provider/sdk/:$PYTHONPATH + +- That's it. + +## Generate repo configuraton + + cd online_server + python -m common.utils.utils $(pwd)/config/cfg.json diff --git a/feast b/feast new file mode 160000 index 0000000..99e5d25 --- /dev/null +++ b/feast @@ -0,0 +1 @@ +Subproject commit 99e5d25e650ea9fc2d288bda9d794fdab60516b7 diff --git a/feast_ol_server.patch b/feast_ol_server.patch new file mode 100644 index 0000000..520b31d --- /dev/null +++ b/feast_ol_server.patch @@ -0,0 +1,414 @@ +diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py +index f5c40d24..4df5e899 100644 +--- a/sdk/python/feast/data_source.py ++++ b/sdk/python/feast/data_source.py +@@ -303,7 +303,7 @@ class DataSource(ABC): + + @staticmethod + @abstractmethod +- def from_proto(data_source: DataSourceProto) -> Any: ++ def from_proto(data_source: DataSourceProto, **kwargs) -> Any: + """ + Converts data source config in protobuf spec to a DataSource class object. + +@@ -326,7 +326,7 @@ class DataSource(ABC): + + if data_source_type == DataSourceProto.SourceType.CUSTOM_SOURCE: + cls = get_data_source_class_from_type(data_source.data_source_class_type) +- return cls.from_proto(data_source) ++ return cls.from_proto(data_source, **kwargs) + + cls = get_data_source_class_from_type(_DATA_SOURCE_OPTIONS[data_source_type]) + return cls.from_proto(data_source) +diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py +index 91215501..ab6082f8 100644 +--- a/sdk/python/feast/feature_store.py ++++ b/sdk/python/feast/feature_store.py +@@ -118,6 +118,7 @@ class FeatureStore: + @log_exceptions + def __init__( + self, repo_path: Optional[str] = None, config: Optional[RepoConfig] = None, ++ project: Optional[str] = None + ): + """ + Creates a FeatureStore object. +@@ -140,6 +141,8 @@ class FeatureStore: + if registry_config.registry_type == "sql": + self._registry = SqlRegistry(registry_config, None) + else: ++ if project: ++ self.config.project = project + r = Registry(registry_config, repo_path=self.repo_path) + r._initialize_registry(self.config.project) + self._registry = r +@@ -161,6 +164,10 @@ class FeatureStore: + """Gets the project of this feature store.""" + return self.config.project + ++ @project.setter ++ def project(self, project: str) -> None: ++ self.config.project = project ++ + def _get_provider(self) -> Provider: + # TODO: Bake self.repo_path into self.config so that we dont only have one interface to paths + return self._provider +@@ -1183,6 +1190,7 @@ class FeatureStore: + @log_exceptions_and_usage + def materialize_incremental( + self, end_date: datetime, feature_views: Optional[List[str]] = None, ++ user: Optional[str] = None, offline_dataset: Optional[str] = None + ) -> None: + """ + Materialize incremental new data from the offline store into the online store. +@@ -1253,15 +1261,28 @@ class FeatureStore: + start_date = utils.make_tzaware(start_date) + end_date = utils.make_tzaware(end_date) + +- provider.materialize_single_feature_view( +- config=self.config, +- feature_view=feature_view, +- start_date=start_date, +- end_date=end_date, +- registry=self._registry, +- project=self.project, +- tqdm_builder=tqdm_builder, +- ) ++ if "dkuberegistrystore" == self._registry._registry_store.__class__.__name__.lower(): ++ provider.materialize_single_feature_view( ++ config=self.config, ++ feature_view=feature_view, ++ start_date=start_date, ++ end_date=end_date, ++ registry=self._registry, ++ project=self.project, ++ tqdm_builder=tqdm_builder, ++ user=user, ++ offline_dataset=offline_dataset ++ ) ++ else: ++ provider.materialize_single_feature_view( ++ config=self.config, ++ feature_view=feature_view, ++ start_date=start_date, ++ end_date=end_date, ++ registry=self._registry, ++ project=self.project, ++ tqdm_builder=tqdm_builder, ++ ) + + self._registry.apply_materialization( + feature_view, self.project, start_date, end_date, +@@ -1273,6 +1294,8 @@ class FeatureStore: + start_date: datetime, + end_date: datetime, + feature_views: Optional[List[str]] = None, ++ user: Optional[str] = None, ++ offline_dataset: Optional[str] = None + ) -> None: + """ + Materialize data from the offline store into the online store. +@@ -1325,15 +1348,28 @@ class FeatureStore: + start_date = utils.make_tzaware(start_date) + end_date = utils.make_tzaware(end_date) + +- provider.materialize_single_feature_view( +- config=self.config, +- feature_view=feature_view, +- start_date=start_date, +- end_date=end_date, +- registry=self._registry, +- project=self.project, +- tqdm_builder=tqdm_builder, +- ) ++ if "dkuberegistrystore" == self._registry._registry_store.__class__.__name__.lower(): ++ provider.materialize_single_feature_view( ++ config=self.config, ++ feature_view=feature_view, ++ start_date=start_date, ++ end_date=end_date, ++ registry=self._registry, ++ project=self.project, ++ tqdm_builder=tqdm_builder, ++ user=user, ++ offline_dataset=offline_dataset ++ ) ++ else: ++ provider.materialize_single_feature_view( ++ config=self.config, ++ feature_view=feature_view, ++ start_date=start_date, ++ end_date=end_date, ++ registry=self._registry, ++ project=self.project, ++ tqdm_builder=tqdm_builder, ++ ) + + self._registry.apply_materialization( + feature_view, self.project, start_date, end_date, +diff --git a/sdk/python/feast/feature_view.py b/sdk/python/feast/feature_view.py +index dd8cb4f0..d92484e2 100644 +--- a/sdk/python/feast/feature_view.py ++++ b/sdk/python/feast/feature_view.py +@@ -455,7 +455,7 @@ class FeatureView(BaseFeatureView): + return ttl_duration + + @classmethod +- def from_proto(cls, feature_view_proto: FeatureViewProto): ++ def from_proto(cls, feature_view_proto: FeatureViewProto, **kwargs): + """ + Creates a feature view from a protobuf representation of a feature view. + +@@ -465,9 +465,9 @@ class FeatureView(BaseFeatureView): + Returns: + A FeatureViewProto object based on the feature view protobuf. + """ +- batch_source = DataSource.from_proto(feature_view_proto.spec.batch_source) ++ batch_source = DataSource.from_proto(feature_view_proto.spec.batch_source, **kwargs) + stream_source = ( +- DataSource.from_proto(feature_view_proto.spec.stream_source) ++ DataSource.from_proto(feature_view_proto.spec.stream_source, **kwargs) + if feature_view_proto.spec.HasField("stream_source") + else None + ) +diff --git a/sdk/python/feast/infra/passthrough_provider.py b/sdk/python/feast/infra/passthrough_provider.py +index 8c6dd831..53954795 100644 +--- a/sdk/python/feast/infra/passthrough_provider.py ++++ b/sdk/python/feast/infra/passthrough_provider.py +@@ -95,10 +95,18 @@ class PassthroughProvider(Provider): + Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] + ], + progress: Optional[Callable[[int], Any]], ++ user: Optional[str] = None, + ) -> None: + set_usage_attribute("provider", self.__class__.__name__) + if self.online_store: +- self.online_store.online_write_batch(config, table, data, progress) ++ # self.online_store.online_write_batch(config, table, data, progress) ++ registry_store = self.get_registry_type() ++ if registry_store and "dkuberegistrystore" in registry_store.lower(): ++ self.online_store.online_write_batch( ++ config, table, data, progress, user ++ ) ++ else: ++ self.online_store.online_write_batch(config, table, data, progress) + + def offline_write_batch( + self, +@@ -121,13 +129,34 @@ class PassthroughProvider(Provider): + table: FeatureView, + entity_keys: List[EntityKeyProto], + requested_features: List[str] = None, ++ user: Optional[str] = None, + ) -> List: ++ """ Read from online datastore. ++ ++ Args: ++ config (RepoConfig): feature store config ++ table (FeatureView): FeatureView table ++ entity_keys (List[EntityKeyProto]): List of entities ++ requested_features (List[str], optional): List of requested features. Defaults to None. ++ user (Optional[str], optional): List of user. Defaults to None. ++ ++ Returns: ++ List: List of result ++ ++ Note:- As of now, this function doesn't gets used by online service. ++ """ + set_usage_attribute("provider", self.__class__.__name__) + result = [] + if self.online_store: +- result = self.online_store.online_read( +- config, table, entity_keys, requested_features +- ) ++ registry_store = self.get_registry_type() ++ if registry_store and "dkuberegistrystore" in registry_store.lower(): ++ result = self.online_store.online_read( ++ config, table, entity_keys, requested_features, user ++ ) ++ else: ++ result = self.online_store.online_read( ++ config, table, entity_keys, requested_features ++ ) + return result + + def ingest_df( +@@ -163,6 +192,8 @@ class PassthroughProvider(Provider): + registry: BaseRegistry, + project: str, + tqdm_builder: Callable[[int], tqdm], ++ user: Optional[str] = None, ++ offline_dataset: Optional[str] = None + ) -> None: + set_usage_attribute("provider", self.__class__.__name__) + +@@ -177,16 +208,43 @@ class PassthroughProvider(Provider): + created_timestamp_column, + ) = _get_column_names(feature_view, entities) + +- offline_job = self.offline_store.pull_latest_from_table_or_query( +- config=config, +- data_source=feature_view.batch_source, +- join_key_columns=join_key_columns, +- feature_name_columns=feature_name_columns, +- timestamp_field=timestamp_field, +- created_timestamp_column=created_timestamp_column, +- start_date=start_date, +- end_date=end_date, +- ) ++ # offline_job = self.offline_store.pull_latest_from_table_or_query( ++ # config=config, ++ # data_source=feature_view.batch_source, ++ # join_key_columns=join_key_columns, ++ # feature_name_columns=feature_name_columns, ++ # timestamp_field=timestamp_field, ++ # created_timestamp_column=created_timestamp_column, ++ # start_date=start_date, ++ # end_date=end_date, ++ # ) ++ registry_store = self.get_registry_type() ++ if registry_store and "dkuberegistrystore" in registry_store.lower(): ++ offline_job = self.offline_store.pull_latest_from_table_or_query( ++ config=config, ++ data_source=feature_view.batch_source, ++ join_key_columns=join_key_columns, ++ feature_name_columns=feature_name_columns, ++ # event_timestamp_column=event_timestamp_column, ++ timestamp_field=timestamp_field, ++ created_timestamp_column=created_timestamp_column, ++ start_date=start_date, ++ end_date=end_date, ++ user=user, ++ offline_dataset=offline_dataset ++ ) ++ else: ++ offline_job = self.offline_store.pull_latest_from_table_or_query( ++ config=config, ++ data_source=feature_view.batch_source, ++ join_key_columns=join_key_columns, ++ feature_name_columns=feature_name_columns, ++ # event_timestamp_column=event_timestamp_column, ++ timestamp_field=timestamp_field, ++ created_timestamp_column=created_timestamp_column, ++ start_date=start_date, ++ end_date=end_date, ++ ) + + table = offline_job.to_arrow() + +@@ -198,17 +256,40 @@ class PassthroughProvider(Provider): + for entity in feature_view.entity_columns + } + +- with tqdm_builder(table.num_rows) as pbar: +- for batch in table.to_batches(DEFAULT_BATCH_SIZE): +- rows_to_write = _convert_arrow_to_proto( +- batch, feature_view, join_key_to_value_type +- ) +- self.online_write_batch( +- self.repo_config, +- feature_view, +- rows_to_write, +- lambda x: pbar.update(x), +- ) ++ # with tqdm_builder(table.num_rows) as pbar: ++ # for batch in table.to_batches(DEFAULT_BATCH_SIZE): ++ # rows_to_write = _convert_arrow_to_proto( ++ # batch, feature_view, join_key_to_value_type ++ # ) ++ # self.online_write_batch( ++ # self.repo_config, ++ # feature_view, ++ # rows_to_write, ++ # lambda x: pbar.update(x), ++ # ) ++ ++ if registry_store and "dkuberegistrystore" in registry_store.lower(): ++ with tqdm_builder(table.num_rows) as pbar: ++ for batch in table.to_batches(DEFAULT_BATCH_SIZE): ++ # rows_to_write = _convert_arrow_to_proto(batch, feature_view, join_keys) ++ rows_to_write = _convert_arrow_to_proto(batch, feature_view, join_key_to_value_type) ++ self.online_write_batch( ++ self.repo_config, ++ feature_view, ++ rows_to_write, ++ lambda x: pbar.update(x), ++ user ++ ) ++ else: ++ with tqdm_builder(table.num_rows) as pbar: ++ for batch in table.to_batches(DEFAULT_BATCH_SIZE): ++ rows_to_write = _convert_arrow_to_proto(batch, feature_view, join_key_to_value_type) ++ self.online_write_batch( ++ self.repo_config, ++ feature_view, ++ rows_to_write, ++ lambda x: pbar.update(x), ++ ) + + def get_historical_features( + self, +@@ -303,3 +384,8 @@ class PassthroughProvider(Provider): + start_date=make_tzaware(start_date), + end_date=make_tzaware(end_date), + ) ++ ++ def get_registry_type(self): ++ registry_cfg = self.repo_config.get_registry_config() ++ registry_type = registry_cfg.registry_store_type.lower() ++ return registry_type +diff --git a/sdk/python/feast/registry.py b/sdk/python/feast/registry.py +index 248d156a..e58e01c5 100644 +--- a/sdk/python/feast/registry.py ++++ b/sdk/python/feast/registry.py +@@ -1301,7 +1301,8 @@ class Registry(BaseRegistry): + and existing_feature_view_proto.spec.project == project + ): + existing_feature_view = FeatureView.from_proto( +- existing_feature_view_proto ++ existing_feature_view_proto, ++ project=project + ) + existing_feature_view.materialization_intervals.append( + (start_date, end_date) +@@ -1312,7 +1313,7 @@ class Registry(BaseRegistry): + del self.cached_registry_proto.feature_views[idx] + self.cached_registry_proto.feature_views.append(feature_view_proto) + if commit: +- self.commit() ++ self.commit(project=project) + return + + for idx, existing_stream_feature_view_proto in enumerate( +@@ -1360,7 +1361,8 @@ class Registry(BaseRegistry): + feature_views: List[FeatureView] = [] + for feature_view_proto in registry_proto.feature_views: + if feature_view_proto.spec.project == project: +- feature_views.append(FeatureView.from_proto(feature_view_proto)) ++ feature_views.append(FeatureView.from_proto(feature_view_proto, ++ project=project)) + return feature_views + + def get_request_feature_view(self, name: str, project: str): +@@ -1749,8 +1751,11 @@ class Registry(BaseRegistry): + if project_metadata.project == project + ] + +- def commit(self): ++ def commit(self, **kwargs): + """Commits the state of the registry cache to the remote registry store.""" ++ if "project" in kwargs: ++ # Note(VK): Do not take any action here. ++ return + if self.cached_registry_proto: + self._registry_store.update_registry_proto(self.cached_registry_proto) + +@@ -1825,7 +1830,8 @@ class Registry(BaseRegistry): + assert isinstance(self.cached_registry_proto, RegistryProto) + return self.cached_registry_proto + +- registry_proto = self._registry_store.get_registry_proto() ++ registry_proto = self._registry_store.get_registry_proto( ++ project=project) + self.cached_registry_proto = registry_proto + self.cached_registry_proto_created = datetime.utcnow() + diff --git a/ksvc-cm.yaml b/ksvc-cm.yaml new file mode 100644 index 0000000..dee3e91 --- /dev/null +++ b/ksvc-cm.yaml @@ -0,0 +1,9 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: ol-cm +data: + DKUBE_URL: https://dkube-proxy.dkube:443 + DKUBE_USER_ACCESS_TOKEN: token + DKUBE_USER_LOGIN_NAME: ocdkube + ONLINE_DATASET: online-dataset diff --git a/online_server/apis/__init__.py b/online_server/apis/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/online_server/apis/feast_api.py b/online_server/apis/feast_api.py new file mode 100644 index 0000000..431970e --- /dev/null +++ b/online_server/apis/feast_api.py @@ -0,0 +1,146 @@ +import traceback +from pathlib import Path + +from drivers import mysql_driver +from fastapi import APIRouter, Depends, Response +from feast.repo_config import load_repo_config +from feast.repo_operations import cli_check_repo, registry_dump +from models.exceptions import ValidationError +from models.online_db import ( + InfraDelete, + InfraUpdate, + Materialize, + MaterializeIncremental, +) +from online_server.common.utils.utils import ( + extract_info, + list_user_info, +) + +router = APIRouter() + + +@router.get("/ping", name="ping", status_code=200) +def ping(): + return {"message": "pong"} + + +@router.post( + "/api/v1/materialize", + name="materialize", + status_code=201, + dependencies=[Depends(extract_info)], +) +async def materialize(materialize_input: Materialize) -> None: + try: + mysql_driver.materialize( + materialize_input.project, + materialize_input.start_date, + materialize_input.end_date, + materialize_input.feature_views, + materialize_input.user, + materialize_input.offline_dataset, + ) + except ValidationError as ve: + return Response(content=ve.error_msg, status_code=ve.status_code) + except Exception as ex: + return Response(content=str(ex), status_code=500) + + +@router.post( + "/api/v1/materialize_incr", + name="materialize_incremental", + status_code=201, + dependencies=[Depends(extract_info)], +) +async def materialize_incremental( + materialize_input: MaterializeIncremental, +) -> None: + try: + mysql_driver.materialize_incremental( + materialize_input.project, + materialize_input.end_date, + materialize_input.feature_views, + materialize_input.user, + materialize_input.offline_dataset, + ) + except ValidationError as ve: + return Response(content=ve.error_msg, status_code=ve.status_code) + except Exception as ex: + print(traceback.format_exc()) + return Response(content=str(ex), status_code=500) + + +@router.post( + "/api/v1/infra_update", + name="infra_update", + status_code=201, + dependencies=[Depends(extract_info)], +) +async def update_infra(infra_input: InfraUpdate) -> None: + try: + mysql_driver.infra_update( + infra_input.project, + infra_input.tables_to_keep, + infra_input.tables_to_delete, + infra_input.entities_to_keep, + infra_input.entities_to_delete, + infra_input.user, + infra_input.offline_dataset, + ) + except ValidationError as ve: + return Response(content=ve.error_msg, status_code=ve.status_code) + except Exception as ex: + print(traceback.format_exc()) + return Response(content=str(ex), status_code=500) + + +@router.delete( + "/api/v1/teardown", + name="teardown", + status_code=200, + dependencies=[Depends(extract_info)], +) +async def teardown_infra(infra_delete: InfraDelete) -> None: + try: + mysql_driver.teardown( + infra_delete.tables, infra_delete.entities, infra_delete.user + ) + except ValidationError as ve: + return Response(content=ve.error_msg, status_code=ve.status_code) + except Exception as ex: + print(traceback.format_exc()) + return Response(content=str(ex), status_code=500) + + +@router.get( + "/api/v1/registry/{project}", + name="registry", + status_code=200, + dependencies=[Depends(extract_info)], +) +async def registry_details(project: str): + repo = str(Path().absolute().parent) + "/online_repo" + cli_check_repo(repo) + repo_config = load_repo_config(repo) + repo_config.project = project + dump = registry_dump(repo_config, repo_path=repo) + return dump + + +@router.get( + "/api/v1/user_info", status_code=200, dependencies=[Depends(extract_info)] +) +async def dump_user_info(): + return list_user_info() + + +# if __name__ == "__main__": +# json_data = { +# "start_date": datetime.now().isoformat(), +# "end_date": datetime.now().isoformat(), +# } +# serialized_data = json.dumps(json_data) +# print(serialized_data) +# m = Materialize.parse_raw(serialized_data) +# print("m: ", m) diff --git a/online_server/apis/feature_server.py b/online_server/apis/feature_server.py new file mode 100644 index 0000000..8d31e81 --- /dev/null +++ b/online_server/apis/feature_server.py @@ -0,0 +1,62 @@ +import fastapi +from fastapi import 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 +from feast.type_map import feast_value_type_to_python_type + +from common.utils.utils import get_repo_path + + +router = fastapi.APIRouter() + + +@router.get("/get-online-features") +async def get_online_features(request: Request): + proto_json.patch() + try: + # Validate and parse the request data into GetOnlineFeaturesRequest Protobuf object + body = await request.body() + request_proto = GetOnlineFeaturesRequest() + Parse(body, request_proto) + + repo_path = get_repo_path() + store = feast.FeatureStore(repo_path=repo_path) + + # 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: feast_value_type_to_python_type(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( # type: ignore + response_proto, preserving_proto_field_name=True, float_precision=18 + ) + 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)) diff --git a/online_server/common/__init__.py b/online_server/common/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/online_server/common/base.py b/online_server/common/base.py new file mode 100644 index 0000000..b263728 --- /dev/null +++ b/online_server/common/base.py @@ -0,0 +1,10 @@ +from typing import Any + + +class Base(type): + _instances = {} + + def __call__(cls, *args: Any, **kwds: Any) -> None: + if cls not in cls._instances: + cls._instances[cls] = super(Base, cls).__call__(*args, **kwds) + return cls._instances[cls] diff --git a/online_server/common/templates/feast_repo.j2 b/online_server/common/templates/feast_repo.j2 new file mode 100644 index 0000000..f6c40bf --- /dev/null +++ b/online_server/common/templates/feast_repo.j2 @@ -0,0 +1,31 @@ +project: {{ input.project | default("dummy_project", true) }} +registry: + registry_store_type: custom_provider.custom_registry.ProtoRegistryStore + {%- if input.registry %} + path: {{ input.registry.ip | default("fake_ip", true)+":"+ + input.registry.port | default("fake_port", true)+":"+ + input.registry.user | default("fake_user", true)+"@"+ + input.registry.password | default("fake_password", true)+":"+ + input.registry.db | default("fake_db", true)+":"+ + input.registry.table | default("fake_table", true) + }} + {%- else %} + path: "fake_ip:fake_port:fake_user@fake_passwd:fake_db:fake_table" + {%- endif %} +provider: custom_provider.proto_custom_provider.ProtoCustomProvider +online_store: + type: custom_provider.custom_online_store.CustomOnlineStore + host: {{ input.online_store.ip | default("ol_fake_host", true) }} + port: {{ input.online_store.port | default("ol_fake_port", true) }} + db: {{ input.online_store.db | default("ol_fake_db", true) }} + user: {{ input.online_store.user | default("ol_fake_user", true) }} + password: {{ input.online_store.password | default("ol_fake_password", true) }} +offline_store: + type: custom_provider.mysqlserver.MySQLOfflineStore + connection_str: {{ + input.offline_store.ip | default("of_fake_ip", true)+":"+ + input.offline_store.port | default("of_fake_port", true)+":"+ + input.offline_store.user | default("of_fake_user", true)+"@"+ + input.offline_store.password | default("of_fake_password", true)+":"+ + input.offline_store.db | default("of_fake_db", true) + }} diff --git a/online_server/common/utils/__init__.py b/online_server/common/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/online_server/common/utils/utils.py b/online_server/common/utils/utils.py new file mode 100644 index 0000000..064ea2e --- /dev/null +++ b/online_server/common/utils/utils.py @@ -0,0 +1,126 @@ +import json +import sys +from pathlib import Path +from typing import Any, Dict, Optional +from fastapi import Depends, Request +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +import jinja2 +import yaml + + +USERS = {} +token_auth_scheme = HTTPBearer() + + +def get_repo_path() -> str: + repo_path = Path("online_repo/").absolute() + repo_path_str = str(repo_path) + return repo_path_str + + +def gen_template( + input_file: str, output_file: str, template_file: Optional[str] = None +): + if input_file.endswith((".yml", ".yaml")): + with open(input_file, "r") as f: + data = yaml.load(f, Loader=yaml.SafeLoader) + elif input_file.endswith(".json"): + with open(input_file, "r") as f: + data = json.loads(f.read()) + else: + sys.exit(f"input_file {input_file} type not supported.") + + if template_file and template_file.suffix != ".j2": + sys.exit(f"Template file {template_file} format not supported") + + if not template_file: + template_file = Path("common/templates/feast_repo.j2").absolute() + template_file = str(template_file) + + with open(template_file, "r") as tfile: + template_data = tfile.read() + + template = jinja2.Template(template_data, keep_trailing_newline=True) + config_data = template.render({"input": data}) + + online_repo = Path("online_repo/").absolute() + output_file = f"{str(online_repo)}/{output_file}" + + with open(output_file, "w") as of: + of.write(config_data) + + print(f"Generated o/p file: {output_file}") + + +def add_user_info(user: str, val: dict): + global USERS + USERS[user] = val + + +def get_user_info(user: str) -> dict: + global USERS + if user in USERS: + user_info = USERS[user] + else: + user_info = {user: {}} + return user_info + + +def get_user_info_by_project(project: str) -> dict: + global USERS + for user, user_val in USERS.items(): + if user_val["project"] == project: + return { + "user": user, + **USERS[user] + } + return {} + + +def del_user_info(user: str): + global USERS + if user in USERS: + del USERS[user] + + +def list_user_info() -> Dict[str, Dict[Any, Any]]: + global USERS + return USERS + + +def set_env(project, user, offline_dataset): + pass + + +def unset_env(project): + pass + + +async def extract_info( + request: Request, + token: HTTPAuthorizationCredentials = Depends(token_auth_scheme), +): + try: + req_json = await request.json() + except json.decoder.JSONDecodeError: + req_json = {} + user = req_json.get("user") + offline_dataset = req_json.get("offline_dataset") + project = req_json.get("project") + user_token = token.credentials + if user: + add_user_info( + user, { + "offline_dataset": offline_dataset, + "token": user_token, + "project": project + } + ) + + +if __name__ == "__main__": + if len(sys.argv) < 3: + sys.exit("Usage: python ") + input_file = sys.argv[1] + output_file = sys.argv[2] + gen_template(input_file, output_file) diff --git a/online_server/config/cfg.json b/online_server/config/cfg.json new file mode 100644 index 0000000..5749cff --- /dev/null +++ b/online_server/config/cfg.json @@ -0,0 +1,17 @@ +{ + "project": "", + "online_store": { + "ip": "", + "port": "", + "user": "", + "password": "", + "db": "" + }, + "offline_store": { + "ip": "", + "port": "", + "user": "", + "password": "", + "db": "" + } +} diff --git a/online_server/drivers/__init__.py b/online_server/drivers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/online_server/drivers/mysql_config.py b/online_server/drivers/mysql_config.py new file mode 100644 index 0000000..65c6edb --- /dev/null +++ b/online_server/drivers/mysql_config.py @@ -0,0 +1,24 @@ +from typing import Any, Dict, Optional + +from common.base import Base +from provider.sdk.dkubefs.utils import get_dkube_db_config + + +class MysqlConfig(metaclass=Base): + _CONFIG: Optional[Dict[str, Any]] = None + + def __init__(self, user) -> None: + if not self._CONFIG: + online_store = get_dkube_db_config(user) + self._CONFIG = { + "host": online_store["host"], + "port": online_store["port"], + "user": online_store["user"], + "password": online_store["password"], + "database": online_store["db"], + "autocommit": True, + } + + @staticmethod + def get_config(): + return MysqlConfig._CONFIG diff --git a/online_server/drivers/mysql_driver.py b/online_server/drivers/mysql_driver.py new file mode 100644 index 0000000..432b797 --- /dev/null +++ b/online_server/drivers/mysql_driver.py @@ -0,0 +1,90 @@ +from datetime import datetime +from typing import List, Optional, Sequence + +from feast.feature_store import FeatureStore +from feast.repo_config import RepoConfig +from mysql.connector import connect + +from drivers.mysql_config import MysqlConfig +from common.utils.utils import get_repo_path + + +def materialize( + project: str, + start_date: datetime, + end_date: datetime, + feature_views: Optional[List[str]] = None, + user: Optional[str] = None, + offline_dataset: Optional[str] = None +) -> None: + fs = FeatureStore(repo_path=get_repo_path()) + fs.project = project + fs.user = user + fs.offline_dataset = offline_dataset + fs.materialize(start_date, end_date, feature_views, user, offline_dataset) + + +def materialize_incremental( + project: str, + end_date: datetime, + feature_views: Optional[List[str]] = None, + user: Optional[str] = None, + offline_dataset: Optional[str] = None +) -> None: + fs = FeatureStore(repo_path=get_repo_path(), project=project) + fs.project = project + fs.user = user + fs.offline_dataset = offline_dataset + fs.materialize_incremental(end_date, feature_views, user, offline_dataset) + + +def infra_update( + project: str, + tables_to_keep: Sequence[str], + tables_to_delete: Sequence[str], + entities_to_keep: Optional[Sequence[str]], + entities_to_delete: Optional[Sequence[str]], + user: Optional[str], + offline_dataset: Optional[str] = None +): + with connect(**MysqlConfig(user)._CONFIG) as conn: + with conn.cursor(buffered=True) as cursor: + for table in tables_to_keep: + _create_query = f""" + create table IF NOT EXISTS {table} (entity_key VARCHAR(512), + feature_name VARCHAR(256), + value BLOB, event_ts timestamp, created_ts timestamp, + PRIMARY KEY(entity_key, feature_name)) + """ + cursor.execute(_create_query) + _index_query = f""" + alter table {table} ADD INDEX + {table}_ek (entity_key) + """ + # cursor.execute(_index_query) + + with connect(**MysqlConfig(user)._CONFIG) as conn: + with conn.cursor(buffered=True) as cursor: + for table in tables_to_delete: + _drop_index = f""" + drop index if exists {table}_ek on {table} + """ + cursor.execute(_drop_index) + _drop_table = f""" + drop table if exists {table} + """ + cursor.execute(_drop_table) + + +def teardown( + tables: Sequence[str], + entities: Optional[Sequence[str]] = None, + user: Optional[str] = None + ) -> None: + with connect(**MysqlConfig(user)._CONFIG) as conn: + with conn.cursor(buffered=True) as cursor: + for table in tables: + _drop_table = f""" + drop table if exists {table} + """ + cursor.execute(_drop_table) diff --git a/online_server/feature_store.yaml b/online_server/feature_store.yaml new file mode 100644 index 0000000..62557a1 --- /dev/null +++ b/online_server/feature_store.yaml @@ -0,0 +1,15 @@ +project: demo_project +registry: + registry_store_type: custom_provider.custom_registry.ProtoRegistryStore + path: "10.43.198.52:3306:root@password:feastdb:driver_ranking_reg" +provider: custom_provider.proto_custom_provider.ProtoCustomProvider +online_store: + type: custom_provider.custom_online_store.CustomOnlineStore + host: "10.43.103.119" + port: "3306" + db: feastdb + user: root + password: password +offline_store: + type: custom_provider.mysqlserver.MySQLOfflineStore + connection_str: "10.43.198.52:3306:root@password:feastdb" diff --git a/online_server/models/__init__.py b/online_server/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/online_server/models/exceptions.py b/online_server/models/exceptions.py new file mode 100644 index 0000000..bf11ff1 --- /dev/null +++ b/online_server/models/exceptions.py @@ -0,0 +1,6 @@ +class ValidationError(Exception): + def __init__(self, error_msg: str, status_code: int): + super().__init__(error_msg) + + self.status_code = status_code + self.error_msg = error_msg diff --git a/online_server/models/online_db.py b/online_server/models/online_db.py new file mode 100644 index 0000000..ccba288 --- /dev/null +++ b/online_server/models/online_db.py @@ -0,0 +1,39 @@ +from datetime import datetime +from typing import List, Optional, Sequence + +from pydantic import BaseModel + + +class Materialize(BaseModel): + project: str + start_date: datetime + end_date: datetime + feature_views: Optional[List[str]] + user: Optional[str] + offline_dataset: Optional[str] + + +class MaterializeIncremental(BaseModel): + project: str + end_date: datetime + feature_views: Optional[List[str]] + user: Optional[str] + offline_dataset: Optional[str] + + +class InfraUpdate(BaseModel): + project: str + tables_to_delete: Sequence[str] + tables_to_keep: Sequence[str] + entities_to_keep: Optional[Sequence[str]] = [] + entities_to_delete: Optional[Sequence[str]] = [] + user: Optional[str] + offline_dataset: Optional[str] + + +class InfraDelete(BaseModel): + project: str + tables: Sequence[str] + entities: Optional[Sequence[str]] = None + user: Optional[str] + offline_dataset: Optional[str] diff --git a/online_server/online_repo/.placeholder b/online_server/online_repo/.placeholder new file mode 100644 index 0000000..e69de29 diff --git a/online_server/online_repo/feature_store.yaml b/online_server/online_repo/feature_store.yaml new file mode 100644 index 0000000..0345df0 --- /dev/null +++ b/online_server/online_repo/feature_store.yaml @@ -0,0 +1,9 @@ +project: '' +registry: + registry_store_type: dkubefs.dkube_registry.DkubeRegistryStore + path: "" +provider: dkubefs.dkube_provider.DkubeProvider +online_store: + type: dkubefs.dkube_store.DkubeOnlineStore +offline_store: + type: dkubefs.mysqlserver.MySQLOfflineStore diff --git a/online_server/online_svc.yaml b/online_server/online_svc.yaml new file mode 100644 index 0000000..b251108 --- /dev/null +++ b/online_server/online_svc.yaml @@ -0,0 +1,40 @@ +apiVersion: serving.knative.dev/v1 +kind: Service +metadata: + name: feast-online-server + namespace: default +spec: + template: + metadata: + annotations: + autoscaling.knative.dev/target: "10" + autoscaling.knative.dev/scaleToZeroPodRetentionPeriod: "30m0s" + spec: + containers: + - image: ocdr/feast-olserver + imagePullPolicy: IfNotPresent + ports: + - containerPort: 8081 + env: + - name: DKUBE_URL + valueFrom: + configMapKeyRef: + name: ol-cm + key: DKUBE_URL + - name: DKUBE_USER_LOGIN_NAME + valueFrom: + configMapKeyRef: + name: ol-cm + key: DKUBE_USER_LOGIN_NAME + - name: OFFLINE_DATASET + valueFrom: + configMapKeyRef: + name: ol-cm + key: OFFLINE_DATASET + - name: DKUBE_USER_ACCESS_TOKEN + valueFrom: + configMapKeyRef: + name: ol-cm + key: DKUBE_USER_ACCESS_TOKEN + - name: ONLINE_DATASET + value: online-dataset diff --git a/online_server/requirements.txt b/online_server/requirements.txt new file mode 100644 index 0000000..e2c2cc0 --- /dev/null +++ b/online_server/requirements.txt @@ -0,0 +1,9 @@ +fastapi +uvicorn +mysql-connector-python +sqlalchemy +# aiomysql +python-decouple +packaging +pymysql +git+https://github.com/oneconvergence/dkube.git@feast_changes diff --git a/online_server/server.py b/online_server/server.py new file mode 100644 index 0000000..0f8b55b --- /dev/null +++ b/online_server/server.py @@ -0,0 +1,11 @@ +import fastapi +import uvicorn + +from apis import feast_api, feature_server + +api = fastapi.FastAPI() + +if __name__ == "__main__": + api.include_router(feature_server.router) + api.include_router(feast_api.router) + uvicorn.run(api, port=8081, host="0.0.0.0") diff --git a/provider/sdk/dkubefs/__init__.py b/provider/sdk/dkubefs/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/provider/sdk/dkubefs/dkube_client.py b/provider/sdk/dkubefs/dkube_client.py new file mode 100644 index 0000000..58ac03d --- /dev/null +++ b/provider/sdk/dkubefs/dkube_client.py @@ -0,0 +1,112 @@ +from functools import wraps +import requests +import json +import urllib3 + +urllib3.disable_warnings() + + +class DkubeClient(object): + def __init__(self, **kwargs) -> None: + self.dkube_ip = kwargs.get("dkube_ip", "192.168.x.y") + self.dkube_port = kwargs.get("dkube_port", 32222) + self.dkube_endpoint = kwargs.get("dkube_endpoint", True) + self.token = kwargs.get("token", "") + + def api_endpoint(self, endpoint): + if not self.dkube_endpoint: + return f"http://{self.dkube_ip}:{self.dkube_port}/{endpoint}" + return f"https://{self.dkube_ip}:{self.dkube_port}/dkube/v2/controller/{endpoint}" + + def headers(self, headers=None): + dkube_headers = { + "Content-Type": "application/json", + } + if self.dkube_endpoint: + dkube_headers.update(Authorization=f"Bearer {self.token}") + if headers and isinstance(dict, headers): + dkube_headers.update(headers) + return dkube_headers + + def handle_error(f): + @wraps(f) + def wrapper(*args, **kwargs): + try: + return f(*args, **kwargs) + except requests.exceptions.HTTPError as errh: + raise Exception(f"Call failed. {errh}") + except requests.exceptions.Timeout as errt: + raise Exception(f"Call timed out. {errt}") + except requests.exceptions.ConnectionError as errc: + raise Exception(f"Connection failed. {errc}") + except requests.exceptions.RequestException as err: + raise Exception(f"Exception. {err}") + return wrapper + + def process_response(self, resp): + print(resp.request.url) + if resp.status_code >= 300: + raise Exception("status code: {} err: {}".format(resp.status_code, resp.text if resp.text else "N/A")) + parsed_resp = None + try: + if resp.headers.get('Content-Type').startswith('application/json'): + parsed_resp = resp.json() + else: + parsed_resp = resp.text + print(f"status code: {resp.status_code} - {parsed_resp}") + return parsed_resp + except Exception as err: + print(err) + raise + + @handle_error + def put(self, endpoint, headers=None, data=None, verify=False, params=None, timeout=45): + updated_headers = self.headers(headers) + url = self.api_endpoint(endpoint) + resp = requests.put( + url, + headers=updated_headers, + data=json.dumps(data), + verify=verify, + params=params, + timeout=timeout) + return self.process_response(resp) + + @handle_error + def post(self, endpoint, headers=None, data=None, verify=False, params=None, timeout=45): + updated_headers = self.headers(headers) + url = self.api_endpoint(endpoint) + resp = requests.post( + url, + headers=updated_headers, + data=json.dumps(data), + verify=verify, + params=params, + timeout=timeout) + return self.process_response(resp) + + @handle_error + def get(self, endpoint, headers=None, data=None, verify=False, params=None, timeout=45): + updated_headers = self.headers(headers) + url = self.api_endpoint(endpoint) + resp = requests.get( + url, + headers=updated_headers, + data=json.dumps(data), + verify=verify, + params=params, + timeout=timeout) + return self.process_response(resp) + + @handle_error + def delete(self, endpoint, headers=None, data=None, verify=False, params=None, timeout=45): + updated_headers = self.headers(headers) + url = self.api_endpoint(endpoint) + resp = requests.delete( + url, + headers=updated_headers, + data=json.dumps(data), + verify=verify, + params=params, + timeout=timeout) + return self.process_response(resp) diff --git a/provider/sdk/dkubefs/dkube_provider.py b/provider/sdk/dkubefs/dkube_provider.py new file mode 100644 index 0000000..992d5a2 --- /dev/null +++ b/provider/sdk/dkubefs/dkube_provider.py @@ -0,0 +1,7 @@ +from feast.infra.passthrough_provider import PassthroughProvider + + +class DkubeProvider(PassthroughProvider): + """ + (VK): We need to see if we can work with the PassthroughProvider + """ diff --git a/provider/sdk/dkubefs/dkube_registry.py b/provider/sdk/dkubefs/dkube_registry.py new file mode 100644 index 0000000..f2291b9 --- /dev/null +++ b/provider/sdk/dkubefs/dkube_registry.py @@ -0,0 +1,60 @@ +import base64 +import uuid +from datetime import datetime +from pathlib import Path + +from dkube.sdk import DkubeApi +from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto +from feast.registry_store import RegistryStore +from feast.repo_config import RegistryConfig +from online_server.common.utils.utils import get_user_info_by_project +from provider.sdk.dkubefs.utils import (get_dkube_client, get_registry_config, + get_user_token) + + +class DkubeRegistryStore(RegistryStore): + def __init__(self, registry_config: RegistryConfig, repo_path: Path): + reg_conf = get_registry_config() + self.dkube_url = reg_conf["url"] + + def get_dkube_client(self, project): + user_info = get_user_info_by_project(project) + if "token" not in user_info: + raise Exception("Token not found.") + return DkubeApi(URL=self.dkube_url, token=user_info["token"]) + + def get_registry_proto(self, **kwargs) -> RegistryProto: + registry_proto = RegistryProto() + project = kwargs['project'] + dkube_client = self.get_dkube_client(project) + json_res = dkube_client.get_registry(project) + if 'regblob' in json_res and json_res['regblob']: + reg_proto = base64.b64decode(json_res['regblob'].encode('ascii')) + registry_proto.ParseFromString(reg_proto) + return registry_proto + else: + print("No details found in project registry.") + return RegistryProto() + + def update_registry_proto(self, registry_proto: RegistryProto, **kwargs): + registry_proto.version_id = str(uuid.uuid4()) + registry_proto.last_updated.FromDatetime(datetime.utcnow()) + project = kwargs['project'] + reg_data = { + "project": project, + "regblob": base64.b64encode( + registry_proto.SerializeToString()).decode('ascii'), + "to_add": kwargs['to_add'], + "to_delete": kwargs['to_delete'] + } + dkube_client = get_dkube_client(project) + dkube_client.update_registry(reg_data) + + def teardown(self, **kwargs): + dkube_client = self.get_dkube_client() + dkube_client.delete_registry(kwargs['project']) + + def validate_infra_update_with_registry(self, update_infra): + json_res = self.dkube.validate_project_changes(update_infra) + if json_res.get('valid') is False: + raise Exception("Invalid registry input.") diff --git a/provider/sdk/dkubefs/dkube_store.py b/provider/sdk/dkubefs/dkube_store.py new file mode 100644 index 0000000..fd44038 --- /dev/null +++ b/provider/sdk/dkubefs/dkube_store.py @@ -0,0 +1,219 @@ +from datetime import datetime +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union + +import pytz +from feast import RepoConfig +from feast.entity import Entity +from feast.feature_view import FeatureView +from feast.infra.key_encoding_utils import serialize_entity_key +from feast.infra.online_stores.online_store import OnlineStore +from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.protos.feast.types.Value_pb2 import Value as ValueProto +from feast.repo_config import FeastConfigBaseModel +from mysql.connector import connect +from provider.sdk.dkubefs.utils import get_dkube_db_config +from pydantic.typing import Literal + + +class DkubeOnlineStoreConfig(FeastConfigBaseModel): + type: Literal[ + "dkubefs.dkube_store.DkubeOnlineStore" + ] = "dkubefs.dkube_store.DkubeOnlineStore" + + +class DkubeOnlineStore(OnlineStore): + online_store_config = None + connect_args = None + + def initialize(self, config): + if self.online_store_config: + return + self.online_store_config = config.online_store + # self.connect_args = get_dkube_db_config() + + def online_write_batch( + self, + config: RepoConfig, + table: FeatureView, + data: List[ + Tuple[ + EntityKeyProto, + Dict[str, ValueProto], + datetime, + Optional[datetime], + ] + ], + progress: Optional[Callable[[int], Any]], + user + ) -> None: + self.initialize(config) + project = config.project + for entity_key, values, timestamp, created_ts in data: + entity_key_bin = serialize_entity_key(entity_key).hex() + timestamp = _to_naive_utc(timestamp) + if created_ts: + created_ts = _to_naive_utc(created_ts) + + for feature_name, val in values.items(): + self.insert_into_table( + user, + project, + table, + entity_key_bin, + feature_name, + timestamp, + created_ts, + val, + ) + if progress: + progress(1) + + def insert_into_table( + self, + user, + project, + table, + entity_key_bin, + feature_name, + timestamp, + created_ts, + val, + ): + self.connect_args = get_dkube_db_config(user) + with connect(**self.connect_args) as conn: + with conn.cursor(buffered=True) as cursor: + _update_query = f""" + update {_table_name(project, table)} set value = %s, + event_ts = %s, created_ts = %s + where (entity_key = %s and feature_name = %s) + """ + cursor.execute( + _update_query, + ( + val.SerializeToString(), + timestamp, + created_ts, + entity_key_bin, + feature_name, + ), + ) + with connect(**self.connect_args) as conn: + with conn.cursor(buffered=True) as cursor: + _insert_query = f""" + insert ignore into {_table_name(project, table)} (entity_key, + feature_name, value, event_ts, created_ts) values ( + %s, %s, %s, %s, %s) + """ + cursor.execute( + _insert_query, + ( + entity_key_bin, + feature_name, + val.SerializeToString(), + timestamp, + created_ts, + ), + ) + + def online_read( + self, + config: RepoConfig, + table: FeatureView, + entity_keys: List[EntityKeyProto], + requested_features: List[str] = None, + user: Optional[str] = None, + ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]: + self.initialize(config) + project = config.project + result = list() + for entity_key in entity_keys: + entity_key_bin = serialize_entity_key(entity_key).hex() + _query = f""" + select entity_key, feature_name, value, event_ts from + {_table_name(project, table)} where + entity_key = %s + """ + with connect(**self.connect_args) as conn: + with conn.cursor(buffered=True) as cursor: + cursor.execute(_query, (entity_key_bin,)) + res, res_ts = dict(), None + for _, _feature_name, _value, _ts in cursor.fetchall(): + val = ValueProto() + val.ParseFromString(_value) + res[_feature_name] = val + res_ts = _ts + + if not res: + result.append((None, None)) + else: + result.append((res_ts, res)) + return result + + def update( + self, + config: RepoConfig, + tables_to_delete: Sequence[FeatureView], + tables_to_keep: Sequence[FeatureView], + entities_to_delete: Sequence[Entity], + entities_to_keep: Sequence[Entity], + partial: bool, + ): + self.initialize(config) + project = config.project + with connect(**self.connect_args) as conn: + with conn.cursor(buffered=True) as cursor: + for table in tables_to_keep: + _create_query = f""" + create table IF NOT EXISTS {_table_name(project, table) + } (entity_key VARCHAR(512), feature_name VARCHAR(256), + value BLOB, event_ts timestamp, created_ts timestamp, + PRIMARY KEY(entity_key, feature_name)) + """ + cursor.execute(_create_query) + _index_query = f""" + alter table {_table_name(project, table)} ADD INDEX + {_table_name(project, table)}_ek (entity_key) + """ + # cursor.execute(_index_query) + + with connect(**self.connect_args) as conn: + with conn.cursor(buffered=True) as cursor: + for table in tables_to_delete: + _drop_index = f""" + drop index if exists {_table_name(project, table)}_ek on {_table_name(project, table)} + """ + cursor.execute(_drop_index) + _drop_table = f""" + drop table if exists {_table_name(project, table)} + """ + cursor.execute(_drop_table) + + def teardown( + self, + config: RepoConfig, + tables: Sequence[FeatureView], + entities: Sequence[Entity], + ): + # teardown_infra should remove all deployed infrastructure + # Replace the code below in order to define your own custom teardown + # operations + self.initialize(config) + project = config.project + with connect(**self.connect_args) as conn: + with conn.cursor(buffered=True) as cursor: + for table in tables: + _drop_table = f""" + drop table if exists {_table_name(project, table)} + """ + cursor.execute(_drop_table) + + +def _to_naive_utc(ts: datetime): + if ts.tzinfo is None: + return ts + else: + return ts.astimezone(pytz.utc).replace(tzinfo=None) + + +def _table_name(project, table): + return "%s_%s" % (project, table.name) diff --git a/provider/sdk/dkubefs/mysqlserver.py b/provider/sdk/dkubefs/mysqlserver.py new file mode 100644 index 0000000..8db868d --- /dev/null +++ b/provider/sdk/dkubefs/mysqlserver.py @@ -0,0 +1,566 @@ +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Set, Tuple, Union +from feast.infra.offline_stores.offline_store import RetrievalMetadata +from feast.saved_dataset import SavedDatasetStorage + +import numpy as np +import pandas as pd +import pyarrow +from feast import OnDemandFeatureView, errors +from feast.data_source import DataSource +from feast.feature_view import DUMMY_ENTITY_ID, DUMMY_ENTITY_VAL, FeatureView +from feast.infra.offline_stores import offline_utils +from feast.infra.offline_stores.offline_store import OfflineStore, RetrievalJob +from feast.infra.provider import _get_requested_feature_views_to_features_dict +from feast.registry import Registry +from feast.repo_config import FeastConfigBaseModel, RepoConfig +from feast.usage import log_exceptions_and_usage +from provider.sdk.dkubefs.mysqlserver_source import MySQLServerSource +from provider.sdk.dkubefs.utils import get_mysql_connect_args, get_mysql_url +from pydantic.types import StrictStr +from pydantic.typing import Literal +from sqlalchemy import create_engine + +EntitySchema = Dict[str, np.dtype] + + +class MySQLOfflineStoreConfig(FeastConfigBaseModel): + type: Literal["dkubefs.mysqlserver.MySQLOfflineStore"] + + +class MySQLOfflineStore(OfflineStore): + def __init__(self): + self.mysql_connect_args = None + + def _get_mysql_connect_config(self, config: RepoConfig) -> Dict: + if not self.mysql_connect_args: + self.mysql_connect_args = get_mysql_connect_args() + return self.mysql_connect_args + + @staticmethod + @log_exceptions_and_usage(offline_store="mysql") + def pull_latest_from_table_or_query( + config: RepoConfig, + data_source: DataSource, + join_key_columns: List[str], + feature_name_columns: List[str], + timestamp_field: str, + created_timestamp_column: Optional[str], + start_date: datetime, + end_date: datetime, + user: Optional[str] = None, + offline_dataset: Optional[str] = None + ) -> RetrievalJob: + assert type(data_source).__name__ == "MySQLServerSource" + assert config.offline_store.type == ( + "dkubefs.mysqlserver.MySQLOfflineStore" + ) + from_expression = data_source.get_table_query_string() + + columns_join_string = ", ".join(join_key_columns) + if columns_join_string != "": + columns_join_string = "PARTITION BY " + columns_join_string + timestamps = [timestamp_field] + if created_timestamp_column: + timestamps.append(created_timestamp_column) + + timestamp_desc_string = " DESC, ".join(timestamps) + " DESC" + field_string = ", ".join( + join_key_columns + feature_name_columns + timestamps + ) + + splitted_fields = field_string.split(',') + splitted_fields = [x.strip() for x in splitted_fields] + if 'index' in splitted_fields: + splitted_fields.remove('index') + splitted_fields.insert(0, '`index`') + field_string = ', '.join(splitted_fields) + + query = f""" + SELECT + {field_string} + {f", {repr(DUMMY_ENTITY_VAL)} AS {DUMMY_ENTITY_ID}" + if not join_key_columns else ""} + FROM ( + SELECT {field_string}, + ROW_NUMBER() OVER( + {columns_join_string} ORDER BY {timestamp_desc_string} + ) AS _feast_row + FROM {from_expression} + WHERE {timestamp_field} + BETWEEN TIMESTAMP '{start_date}' AND TIMESTAMP '{end_date}' + ) as tt + WHERE _feast_row = 1 + """ + # call retrieval function + return MySQLRetrievalJob( + query=query, + config=config, + full_feature_names=False, + on_demand_feature_views=None, + user=user, + offline_dataset=offline_dataset + ) + + # REVISIT(VK): We may need to revisit this. + @staticmethod + @log_exceptions_and_usage(offline_store="mysql") + def get_historical_features( + config: RepoConfig, + feature_views: List[FeatureView], + feature_refs: List[str], + entity_df: Union[pd.DataFrame, str], + registry: Registry, + project: str, + full_feature_names: bool = False, + ) -> RetrievalJob: + assert isinstance(config.offline_store, MySQLOfflineStoreConfig) + + # REVISIT(VK): Do we need to bring our own definition ? + expected_join_keys = offline_utils.get_expected_join_keys( + project, feature_views, registry + ) + + ( + table_schema, + table_name, + ) = _upload_entity_df_into_mysql_and_get_entity_schema( + config=config, entity_df=entity_df + ) + + entity_df_event_timestamp_col = ( + offline_utils.infer_event_timestamp_from_entity_df(table_schema) + ) + + offline_utils.assert_expected_columns_in_entity_df( + table_schema, expected_join_keys, entity_df_event_timestamp_col + ) + + query_context = get_feature_view_query_context( + feature_refs, feature_views, registry, project + ) + + query = offline_utils.build_point_in_time_query( + query_context, + left_table_query_string=table_name, + entity_df_event_timestamp_col=entity_df_event_timestamp_col, + query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN, + full_feature_names=full_feature_names, + entity_df_columns=table_schema.keys() + ) + + job = MySQLRetrievalJob( + query=query, + config=config, + full_feature_names=full_feature_names, + on_demand_feature_views=registry.list_on_demand_feature_views( + project + ), + ) + return job + + @staticmethod + @log_exceptions_and_usage(offline_store="mysql") + def pull_all_from_table_or_query( + config: RepoConfig, + data_source: DataSource, + join_key_columns: List[str], + feature_name_columns: List[str], + timestamp_field: str, + start_date: datetime, + end_date: datetime, + ) -> RetrievalJob: + """only placeholder in online server + + Args: + config (RepoConfig): _description_ + data_source (DataSource): _description_ + join_key_columns (List[str]): _description_ + feature_name_columns (List[str]): _description_ + timestamp_field (str): _description_ + start_date (datetime): _description_ + end_date (datetime): _description_ + + Returns: + RetrievalJob: _description_ + """ + assert type(data_source).__name__ == "MySQLServerSource" + assert config.offline_store.type == ( + "dkubefs.mysqlserver.MySQLOfflineStore" + ) + pass + + +@dataclass(frozen=True) +class FeatureViewQueryContext: + """Context object used to template a point-in-time MySQL query""" + + name: str + ttl: int + entities: List[str] + features: List[str] # feature reference format + # event_timestamp_column: str + timestamp_field: str + created_timestamp_column: Optional[str] + table_subquery: str + entity_selections: List[str] + table_ref: str + + +def _upload_entity_df_into_mysql_and_get_entity_schema( + config: RepoConfig, + entity_df: Union[pd.DataFrame, str], +) -> Tuple[EntitySchema, str]: + """ + Uploads a Pandas entity dataframe into MySQL table and constructs the + schema from the original entity_df dataframe. + """ + table_id = offline_utils.get_temp_entity_table_name() + _connect_args = get_mysql_connect_args() + _mysql_url = get_mysql_url(_connect_args) + engine = create_engine(_mysql_url) + if type(entity_df) is str: + _query = f"create table {table_id} as {entity_df}" + with engine.connect() as conn: + conn.execute(_query) + + retrieved_entity_df = MySQLRetrievalJob( + f"SELECT * FROM {table_id} LIMIT 1", + config, + full_feature_names=False, + on_demand_feature_views=None, + ).to_df() # to_df - defined in RetrievalJob + return ( + dict(zip(retrieved_entity_df.columns, retrieved_entity_df.dtypes)), + table_id, + ) + + elif type(entity_df) is pd.DataFrame: + # _mysql_url = get_mysql_url(_connect_args) + # engine = create_engine(_mysql_url) + with engine.connect() as conn: + entity_df.to_sql(name=table_id, con=conn, if_exists="replace") + return dict(zip(entity_df.columns, entity_df.dtypes)), table_id + raise Exception("Unsupported entitydf type") + + +def _assert_expected_columns_in_mysql( + join_keys: Set[str], + entity_df_event_timestamp_col: str, + table_schema: EntitySchema, +): + entity_columns = set(table_schema.keys()) + + expected_columns = join_keys.copy() + expected_columns.add(entity_df_event_timestamp_col) + + missing_keys = expected_columns - entity_columns + + if len(missing_keys) != 0: + raise errors.FeastEntityDFMissingColumnsError( + expected_columns, missing_keys + ) + + +def get_feature_view_query_context( + feature_refs: List[str], + feature_views: List[FeatureView], + registry: Registry, + project: str, +) -> List[FeatureViewQueryContext]: + """Build a query context containing all information required to template + a point-in-time SQL query + """ + + ( + feature_views_to_feature_map, + on_demand_feature_views_to_features, + ) = _get_requested_feature_views_to_features_dict( + feature_refs, + feature_views, + registry.list_on_demand_feature_views(project), + ) + + query_context = [] + for feature_view, features in feature_views_to_feature_map.items(): + join_keys = [] + entity_selections = [] + reverse_field_mapping = { + v: k for k, v in feature_view.input.field_mapping.items() + } + for entity_name in feature_view.entities: + entity = registry.get_entity(entity_name, project) + join_key = feature_view.projection.join_key_map.get( + entity.join_key, entity.join_key + ) + join_keys.append(join_key) + entity_selections.append(f"{entity.join_key} AS {join_key}") + + if isinstance(feature_view.ttl, timedelta): + ttl_seconds = int(feature_view.ttl.total_seconds()) + else: + ttl_seconds = 0 + + assert isinstance(feature_view.input, MySQLServerSource) + # event_timestamp_column = feature_view.input.event_timestamp_column + timestamp_field = feature_view.input.timestamp_field + created_timestamp_column = feature_view.input.created_timestamp_column + + context = FeatureViewQueryContext( + name=feature_view.projection.name_to_use(), + ttl=ttl_seconds, + entities=join_keys, + features=features, + # event_timestamp_column=reverse_field_mapping.get( + # event_timestamp_column, event_timestamp_column + # ), + timestamp_field=reverse_field_mapping.get( + timestamp_field, timestamp_field + ), + created_timestamp_column=reverse_field_mapping.get( + created_timestamp_column, created_timestamp_column + ), + # TODO: Make created column optional and not hardcoded + table_subquery=feature_view.input.get_table_query_string(), + entity_selections=entity_selections, + table_ref=feature_view.input.table_ref, + ) + query_context.append(context) + return query_context + + +class MySQLRetrievalJob(RetrievalJob): + def __init__( + self, + query: StrictStr, + config: RepoConfig, + full_feature_names: bool, + on_demand_feature_views: Optional[List[OnDemandFeatureView]], + drop_columns: Optional[List[str]] = None, + metadata: Optional[RetrievalMetadata] = None, + user: Optional[str] = None, + offline_dataset: Optional[str] = None + ): + self.query = query + self._config = config + self._full_feature_names = full_feature_names + self._on_demand_feature_views = on_demand_feature_views + self._drop_columns = drop_columns + self._metadata = metadata + _mysql_url = get_mysql_url(user=user, offline_dataset=offline_dataset) + self.engine = create_engine(_mysql_url) + + @property + def full_feature_names(self) -> bool: + return self._full_feature_names + + @property + def on_demand_feature_views(self) -> Optional[List[OnDemandFeatureView]]: + return self._on_demand_feature_views + + def _to_df_internal(self) -> pd.DataFrame: + with self.engine.connect() as conn: + df = pd.read_sql(sql=self.query, con=conn).fillna(value=np.nan) + return df + + def _to_arrow_internal(self) -> pyarrow.Table: + with self.engine.connect() as conn: + df = pd.read_sql(sql=self.query, con=conn).fillna(value=np.nan) + return pyarrow.Table.from_pandas(df=df) + + def persist(self, storage: SavedDatasetStorage): + pass + + @property + def metadata(self) -> Optional[RetrievalMetadata]: + return self._metadata + + +# REVISIT(VK): This template is in sync with redshift.py. Changes have been done to make it compatible with mysql. For time being we need to keep sync this with bigquery.py and redshift.py +MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN = """ +/* + Compute a deterministic hash for the `left_table_query_string` that will be used throughout + all the logic as the field to GROUP BY the data +*/ +WITH entity_dataframe AS ( + SELECT *, + {{entity_df_event_timestamp_col}} AS entity_timestamp + {% for featureview in featureviews %} + {% if featureview.entities %} + ,( + {% for entity in featureview.entities %} + CAST({{entity}} as CHAR(100)) || + {% endfor %} + CAST({{entity_df_event_timestamp_col}} AS CHAR(50)) + ) AS {{featureview.name}}__entity_row_unique_id + {% else %} + ,CAST({{entity_df_event_timestamp_col}} AS CHAR(100)) AS {{featureview.name}}__entity_row_unique_id + {% endif %} + {% endfor %} + FROM {{ left_table_query_string }} +), + +{% for featureview in featureviews %} + +{{ featureview.name }}__entity_dataframe AS ( + SELECT + {{ featureview.entities | join(', ')}}{% if featureview.entities %},{% else %}{% endif %} + entity_timestamp, + {{featureview.name}}__entity_row_unique_id + FROM entity_dataframe + GROUP BY + {{ featureview.entities | join(', ')}}{% if featureview.entities %},{% else %}{% endif %} + entity_timestamp, + {{featureview.name}}__entity_row_unique_id +), + +/* + This query template performs the point-in-time correctness join for a single feature set table + to the provided entity table. + + 1. We first join the current feature_view to the entity dataframe that has been passed. + This JOIN has the following logic: + - For each row of the entity dataframe, only keep the rows where the `event_timestamp_column` + is less than the one provided in the entity dataframe + - If there a TTL for the current feature_view, also keep the rows where the `event_timestamp_column` + is higher the the one provided minus the TTL + - For each row, Join on the entity key and retrieve the `entity_row_unique_id` that has been + computed previously + + The output of this CTE will contain all the necessary information and already filtered out most + of the data that is not relevant. +*/ + +{{ featureview.name }}__subquery AS ( + SELECT + {{ featureview.timestamp_field }} as event_timestamp, + {{ featureview.created_timestamp_column ~ ' as created_timestamp,' if featureview.created_timestamp_column else '' }} + {{ featureview.entity_selections | join(', ')}}{% if featureview.entity_selections %},{% else %}{% endif %} + {% for feature in featureview.features %} + {{ feature }} as {% if full_feature_names %}{{ featureview.name }}__{{feature}}{% else %}{{ feature }}{% endif %}{% if loop.last %}{% else %}, {% endif %} + {% endfor %} + FROM {{ featureview.table_subquery }} AS sub + WHERE {{ featureview.timestamp_field }} <= (SELECT MAX(entity_timestamp) FROM entity_dataframe) + {% if featureview.ttl == 0 %}{% else %} + AND {{ featureview.timestamp_field }} >= (SELECT MIN(entity_timestamp) FROM entity_dataframe) - interval {{ featureview.ttl }} second + {% endif %} +), + +{{ featureview.name }}__base AS ( + SELECT + subquery.*, + entity_dataframe.entity_timestamp, + entity_dataframe.{{featureview.name}}__entity_row_unique_id + FROM {{ featureview.name }}__subquery AS subquery + INNER JOIN {{ featureview.name }}__entity_dataframe AS entity_dataframe + ON TRUE + AND subquery.event_timestamp <= entity_dataframe.entity_timestamp + + {% if featureview.ttl == 0 %}{% else %} + AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - interval {{ featureview.ttl }} second + {% endif %} + + {% for entity in featureview.entities %} + AND subquery.{{ entity }} = entity_dataframe.{{ entity }} + {% endfor %} +), + +/* + 2. If the `created_timestamp_column` has been set, we need to + deduplicate the data first. This is done by calculating the + `MAX(created_at_timestamp)` for each event_timestamp. + We then join the data on the next CTE +*/ +{% if featureview.created_timestamp_column %} +{{ featureview.name }}__dedup AS ( + SELECT + {{featureview.name}}__entity_row_unique_id, + event_timestamp, + MAX(created_timestamp) as created_timestamp + FROM {{ featureview.name }}__base + GROUP BY {{featureview.name}}__entity_row_unique_id, event_timestamp +), +{% endif %} + +/* + 3. The data has been filtered during the first CTE "*__base" + Thus we only need to compute the latest timestamp of each feature. +*/ +/* +{{ featureview.name }}__latest AS ( + SELECT + {{featureview.name}}__entity_row_unique_id, + MAX(event_timestamp) AS event_timestamp + {% if featureview.created_timestamp_column %} + ,MAX(created_timestamp) AS created_timestamp + {% endif %} + + FROM {{ featureview.name }}__base + {% if featureview.created_timestamp_column %} + INNER JOIN {{ featureview.name }}__dedup + USING ({{featureview.name}}__entity_row_unique_id, event_timestamp, created_timestamp) + {% endif %} + + GROUP BY {{featureview.name}}__entity_row_unique_id +), +*/ + +{{ featureview.name }}__latest AS ( + SELECT + event_timestamp, + {% if featureview.created_timestamp_column %}created_timestamp,{% endif %} + {{featureview.name}}__entity_row_unique_id + FROM + ( + SELECT *, + ROW_NUMBER() OVER( + PARTITION BY {{featureview.name}}__entity_row_unique_id + ORDER BY event_timestamp DESC{% if featureview.created_timestamp_column %},created_timestamp DESC{% endif %} + ) AS row_num + FROM {{ featureview.name }}__base + {% if featureview.created_timestamp_column %} + INNER JOIN {{ featureview.name }}__dedup + USING ({{featureview.name}}__entity_row_unique_id, event_timestamp, created_timestamp) + {% endif %} + ) AS sub + WHERE row_num = 1 +), + +/* + 4. Once we know the latest value of each feature for a given timestamp, + we can join again the data back to the original "base" dataset +*/ +{{ featureview.name }}__cleaned AS ( + SELECT base.* + FROM {{ featureview.name }}__base as base + INNER JOIN {{ featureview.name }}__latest + USING( + {{featureview.name}}__entity_row_unique_id, + event_timestamp + {% if featureview.created_timestamp_column %} + ,created_timestamp + {% endif %} + ) +){% if loop.last %}{% else %}, {% endif %} + + +{% endfor %} +/* + Joins the outputs of multiple time travel joins to a single table. + The entity_dataframe dataset being our source of truth here. + */ + +SELECT * +FROM entity_dataframe +{% for featureview in featureviews %} +LEFT JOIN ( + SELECT + {{featureview.name}}__entity_row_unique_id + {% for feature in featureview.features %} + ,{% if full_feature_names %}{{ featureview.name }}__{{feature}}{% else %}{{ feature }}{% endif %} + {% endfor %} + FROM {{ featureview.name }}__cleaned +) as tmp USING ({{featureview.name}}__entity_row_unique_id) +{% endfor %} +""" diff --git a/provider/sdk/dkubefs/mysqlserver_source.py b/provider/sdk/dkubefs/mysqlserver_source.py new file mode 100644 index 0000000..d8219e2 --- /dev/null +++ b/provider/sdk/dkubefs/mysqlserver_source.py @@ -0,0 +1,210 @@ +import json +from typing import Any, Callable, Dict, Iterable, Optional, Tuple + +import pandas as pd +from feast import RepoConfig, ValueType +from feast.data_source import DataSource +from feast.protos.feast.core.DataSource_pb2 import ( + DataSource as DataSourceProto, +) +from mysql.connector import connect +from provider.sdk.dkubefs.utils import ( + get_mysql_connect_args, + get_offline_connection_str, +) + + +class MySQLOptions: + def __init__(self, connection_str, table_ref): + self._connection_str = connection_str + self._table_ref = table_ref + + @property + def connection_str(self): + return self._connection_str + + @connection_str.setter + def connection_str(self, connection_str): + self._connection_str = connection_str + + @property + def table_ref(self): + return self._table_ref + + @table_ref.setter + def table_ref(self, table_ref): + self._table_ref = table_ref + + @classmethod + def from_proto( + cls, mysql_options_proto: DataSourceProto.CustomSourceOptions + ) -> "MySQLOptions": + options = json.loads(mysql_options_proto.configuration) + mysql_options = cls( + connection_str=options["connection_str"], + table_ref=options["table_ref"], + ) + return mysql_options + + def to_proto(self) -> DataSourceProto.CustomSourceOptions: + mysql_options_proto = DataSourceProto.CustomSourceOptions( + configuration=json.dumps( + { + "connection_str": self._connection_str, + "table_ref": self._table_ref, + } + ).encode("utf-8") + ) + return mysql_options_proto + + +class MySQLServerSource(DataSource): + def __init__( + self, + event_timestamp_column: Optional[str] = "", + table_ref: Optional[str] = None, + created_timestamp_column: Optional[str] = "", + field_mapping: Optional[Dict[str, str]] = None, + date_partition_column: Optional[str] = "", + tags: Optional[Dict[str, str]] = None, + owner: Optional[str] = None, + name: Optional[str] = None, + description: Optional[str] = None, + timestamp_field: Optional[str] = None, + **kwargs + ): + connection_str = get_offline_connection_str(project=kwargs["project"]) + self._mysql_options = MySQLOptions(connection_str, table_ref) + self._connection_str = connection_str + self._table_ref = table_ref + _timestamp_field = timestamp_field or event_timestamp_column or "" + super().__init__( + timestamp_field=_timestamp_field, + created_timestamp_column=created_timestamp_column, + field_mapping=field_mapping, + date_partition_column=date_partition_column, + tags=tags, + owner=owner, + name=name, + description=description, + ) + + def __hash__(self): + return super().__hash__() + + def __eq__(self, other): + if not isinstance(other, MySQLServerSource): + raise TypeError( + "Comparisons should only involve MySQLServerSource class objects." + ) + + return ( + self._mysql_options.connection_str + == other._mysql_options.connection_str + and self.timestamp_field == other.timestamp_field + and self.created_timestamp_column == other.created_timestamp_column + and self.field_mapping == other.field_mapping + ) + + @property + def connection_str(self): + return self._mysql_options.connection_str + + @property + def table_ref(self): + return self._mysql_options.table_ref + + @property + def mysql_options(self): + return self._mysql_options + + @mysql_options.setter + def mysql_options(self, _options): + self._mysql_options = _options + + @staticmethod + def from_proto(data_source: DataSourceProto, **kwargs) -> Any: + options = json.loads(data_source.custom_options.configuration) + return MySQLServerSource( + field_mapping=dict(data_source.field_mapping), + table_ref=options["table_ref"], + timestamp_field=data_source.timestamp_field, + created_timestamp_column=data_source.created_timestamp_column, + date_partition_column=data_source.date_partition_column, + **kwargs + ) + + def to_proto(self) -> DataSourceProto: + data_source_proto = DataSourceProto( + type=DataSourceProto.CUSTOM_SOURCE, + field_mapping=self.field_mapping, + custom_options=self._mysql_options.to_proto(), + ) + data_source_proto.timestamp_field = self.timestamp_field + data_source_proto.created_timestamp_column = ( + self.created_timestamp_column + ) + data_source_proto.date_partition_column = self.date_partition_column + return data_source_proto + + def validate(self, config: RepoConfig): + # REVISIT(VK) + return None + + @staticmethod + def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]: + return mysql_to_feast_value_type + + def get_table_column_names_and_types( + self, config: RepoConfig + ) -> Iterable[Tuple[str, str]]: + _connect_args = get_mysql_connect_args(self._connection_str) + name_type_pairs = list() + _db, _table = _connect_args["database"], self._table_ref + _query = ( + f"select column_name, data_type from " + f"INFORMATION_SCHEMA.COLUMNS where TABLE_SCHEMA = '{_db}' " + f"and TABLE_NAME = '{_table}'" + ) + with connect(**_connect_args) as conn: + table_schema = pd.read_sql(_query, conn) + name_type_pairs.extend( + list( + zip( + table_schema["COLUMN_NAME"].to_list(), + table_schema["DATA_TYPE"].to_list(), + ) + ) + ) + return name_type_pairs + + def get_table_query_string(self) -> str: + return f"{self.table_ref}" + + +def mysql_to_feast_value_type(mysql_type_as_str: str) -> ValueType: + _type_map = { + "char": ValueType.STRING, + "varchar": ValueType.STRING, + "binary": ValueType.STRING, + "varbinary": ValueType.STRING, + "text": ValueType.STRING, + "str": ValueType.STRING, + "blob": ValueType.STRING, + "mediumblob": ValueType.STRING, + "longblob": ValueType.STRING, + "tinyint": ValueType.INT32, + "smallint": ValueType.INT32, + "mediumint": ValueType.INT64, + "bigint": ValueType.INT64, + "int": ValueType.INT64, + "float": ValueType.FLOAT, + "double": ValueType.DOUBLE, + "bool": ValueType.BOOL, + "boolean": ValueType.INT32, + "bytes": ValueType.BYTES, + "timestamp": ValueType.UNIX_TIMESTAMP, + "datetime": ValueType.UNIX_TIMESTAMP, + "null": ValueType.NULL, + } + return _type_map[mysql_type_as_str] diff --git a/provider/sdk/dkubefs/utils.py b/provider/sdk/dkubefs/utils.py new file mode 100644 index 0000000..028292d --- /dev/null +++ b/provider/sdk/dkubefs/utils.py @@ -0,0 +1,159 @@ +import os +import sys +from pathlib import Path + +from dkube.sdk import DkubeApi +from online_server.common.utils.utils import ( + get_user_info, + get_user_info_by_project +) + + +def get_dkube_client(token: str): + reg_conf = get_registry_config() + DKUBE_URL = os.getenv("DKUBE_URL", reg_conf["url"]) + if not DKUBE_URL: + sys.exit("Dkube access url not set.") + # DKUBE_TOKEN = os.getenv("DKUBE_USER_ACCESS_TOKEN", reg_conf["token"]) + DKUBE_TOKEN = token + if not DKUBE_TOKEN: + sys.exit("Dkube access token not set.") + dkube = DkubeApi(URL=DKUBE_URL, token=DKUBE_TOKEN) + return dkube + + +def get_offline_store_conf(offline_user=None, offline_dataset=None): + if offline_user: + USER = offline_user + if offline_dataset: + offline_ds = offline_dataset + token = get_user_token(USER) + dclient = get_dkube_client(token) + ods = dclient.get_dataset(USER, offline_ds) + return { + "user": ods["datum"]["sql"]["username"], + "host": ods["datum"]["sql"]["host"], + "port": ods["datum"]["sql"]["port"], + "password": ods["datum"]["sql"]["password"], + "db": ods["datum"]["sql"]["database"], + "autocommit": True, + } + + +def get_offline_store_conf_by_project(project): + user_info = get_user_info_by_project(project) + token = user_info["token"] + offline_ds = user_info["offline_dataset"] + + dclient = get_dkube_client(token) + ods = dclient.get_dataset(user_info["user"], offline_ds) + return { + "user": ods["datum"]["sql"]["username"], + "host": ods["datum"]["sql"]["host"], + "port": ods["datum"]["sql"]["port"], + "password": ods["datum"]["sql"]["password"], + "db": ods["datum"]["sql"]["database"], + "autocommit": True, + } + + +def get_mysql_connect_args(connection_str=None): + if connection_str: + mysql_config = connection_str.split(":") + mysql_ip = mysql_config[0] + mysql_port = mysql_config[1] + _creds = mysql_config[2].split("@") + _user = _creds[0] + _password = _creds[1] + db = mysql_config[3] + conf = { + "host": mysql_ip, + "port": mysql_port, + "user": _user, + "password": _password, + "database": db, + } + else: + conf = get_offline_store_conf() + conf.update(autocommit=True) + return conf + + +def get_offline_connection_str(user=None, offline_dataset=None, project=None): + if not user and not offline_dataset and project is not None: + offline_conf = get_offline_store_conf_by_project(project) + else: + offline_conf = get_offline_store_conf(user, offline_dataset) + return f"""{offline_conf['host']}:{offline_conf['port']}: + {offline_conf['user']}@{offline_conf['password']}: + {offline_conf['db']}""" + + +def get_dkube_server_config(): + feast_ol_url = os.getenv("FEAST_ONLINE_SERVER_URL") + if feast_ol_url: + return feast_ol_url + else: + print("Using default server config.") + return "http://knative-local-gateway.istio-system.svc.cluster.local" + + +def get_dkube_server_host(): + feast_ol_url = os.getenv("FEAST_ONLINE_SERVER_HOST") + if feast_ol_url: + return feast_ol_url + else: + return {"Host": "feast-online-server.default.svc"} + + +def get_mysql_url(_connect_args=None, user=None, offline_dataset=None): + if not _connect_args: + if user and offline_dataset: + _connect_args = get_offline_store_conf(user, offline_dataset) + else: + raise Exception("MYSQL url cannot be retreived.") + if "db" in _connect_args: + db = _connect_args["db"] + elif "database" in _connect_args: + db = _connect_args["database"] + else: + sys.exit("db or database not set") + return f"""mysql+pymysql://{_connect_args['user']}:{ + _connect_args['password']}@{_connect_args['host']}:{ + _connect_args['port']}/{db}""" + + +def get_dkube_db_config(user): + dds = os.getenv("ONLINE_DATASET") + if not dds: + print("online dataset not found. using default: online-dataset") + dds = "online-dataset" + print("using default dataset: online-dataset") + + token = get_user_token(user) + dclient = get_dkube_client(token) + ods = dclient.get_dataset(user, dds) + return { + "host": ods["datum"]["sql"]["host"], + "port": ods["datum"]["sql"]["port"], + "user": ods["datum"]["sql"]["username"], + "password": ods["datum"]["sql"]["password"], + "db": ods["datum"]["sql"]["database"], + "autocommit": True, + } + + +def get_registry_config(): + dkube_url = os.getenv("DKUBE_URL") + if not dkube_url: + sys.exit("DKUBE_URL not set.") + return {"url": dkube_url} + + +def get_user_token(user: str): + user_info = get_user_info(user) + if not user_info: + raise Exception("User details not found.") + if "token" not in user_info: + raise Exception("User token not found.") + return user_info["token"]