From 7bf741135872d3e49b75b2494b7acb4bc5fd5a56 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 16 Sep 2021 15:55:40 -0700 Subject: [PATCH 1/8] Refactor providers to remove duplicate implementations Signed-off-by: Achal Shah --- sdk/python/feast/infra/aws.py | 145 ++-------------------------- sdk/python/feast/infra/gcp.py | 147 ++--------------------------- sdk/python/feast/infra/local.py | 142 ++-------------------------- sdk/python/feast/infra/provider.py | 128 +++++++++++++++++++++++++ 4 files changed, 150 insertions(+), 412 deletions(-) diff --git a/sdk/python/feast/infra/aws.py b/sdk/python/feast/infra/aws.py index f93553bf9f4..17971c3c0c4 100644 --- a/sdk/python/feast/infra/aws.py +++ b/sdk/python/feast/infra/aws.py @@ -3,152 +3,21 @@ from datetime import datetime from pathlib import Path from tempfile import TemporaryFile -from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union from urllib.parse import urlparse -import pandas -from tqdm import tqdm - -from feast import FeatureTable -from feast.entity import Entity from feast.errors import S3RegistryBucketForbiddenAccess, S3RegistryBucketNotExist -from feast.feature_view import FeatureView -from feast.infra.offline_stores.offline_utils import get_offline_store_from_config -from feast.infra.online_stores.helpers import get_online_store_from_config -from feast.infra.provider import ( - Provider, - RetrievalJob, - _convert_arrow_to_proto, - _get_column_names, - _run_field_mapping, -) +from feast.infra.provider import PassthroughProvider from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto -from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto -from feast.protos.feast.types.Value_pb2 import Value as ValueProto -from feast.registry import Registry from feast.registry_store import RegistryStore -from feast.repo_config import RegistryConfig, RepoConfig - - -class AwsProvider(Provider): - def __init__(self, config: RepoConfig): - self.repo_config = config - self.offline_store = get_offline_store_from_config(config.offline_store) - self.online_store = get_online_store_from_config(config.online_store) - - def update_infra( - self, - project: str, - tables_to_delete: Sequence[Union[FeatureTable, FeatureView]], - tables_to_keep: Sequence[Union[FeatureTable, FeatureView]], - entities_to_delete: Sequence[Entity], - entities_to_keep: Sequence[Entity], - partial: bool, - ): - self.online_store.update( - config=self.repo_config, - tables_to_delete=tables_to_delete, - tables_to_keep=tables_to_keep, - entities_to_keep=entities_to_keep, - entities_to_delete=entities_to_delete, - partial=partial, - ) - - def teardown_infra( - self, - project: str, - tables: Sequence[Union[FeatureTable, FeatureView]], - entities: Sequence[Entity], - ) -> None: - self.online_store.teardown(self.repo_config, tables, entities) - - def online_write_batch( - self, - config: RepoConfig, - table: Union[FeatureTable, FeatureView], - data: List[ - Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] - ], - progress: Optional[Callable[[int], Any]], - ) -> None: - self.online_store.online_write_batch(config, table, data, progress) - - def online_read( - self, - config: RepoConfig, - table: Union[FeatureTable, FeatureView], - entity_keys: List[EntityKeyProto], - requested_features: List[str] = None, - ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]: - result = self.online_store.online_read(config, table, entity_keys) +from feast.repo_config import RegistryConfig - return result - def materialize_single_feature_view( - self, - config: RepoConfig, - feature_view: FeatureView, - start_date: datetime, - end_date: datetime, - registry: Registry, - project: str, - tqdm_builder: Callable[[int], tqdm], - ) -> None: - entities = [] - for entity_name in feature_view.entities: - entities.append(registry.get_entity(entity_name, project)) +class AwsProvider(PassthroughProvider): + """ + This class only exists for backwards compatibility. + """ - ( - join_key_columns, - feature_name_columns, - event_timestamp_column, - 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, - event_timestamp_column=event_timestamp_column, - created_timestamp_column=created_timestamp_column, - start_date=start_date, - end_date=end_date, - ) - - table = offline_job.to_arrow() - - if feature_view.batch_source.field_mapping is not None: - table = _run_field_mapping(table, feature_view.batch_source.field_mapping) - - join_keys = [entity.join_key for entity in entities] - rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys) - - with tqdm_builder(len(rows_to_write)) as pbar: - self.online_write_batch( - self.repo_config, feature_view, rows_to_write, lambda x: pbar.update(x) - ) - - def get_historical_features( - self, - config: RepoConfig, - feature_views: List[FeatureView], - feature_refs: List[str], - entity_df: Union[pandas.DataFrame, str], - registry: Registry, - project: str, - full_feature_names: bool, - ) -> RetrievalJob: - job = self.offline_store.get_historical_features( - config=config, - feature_views=feature_views, - feature_refs=feature_refs, - entity_df=entity_df, - registry=registry, - project=project, - full_feature_names=full_feature_names, - ) - return job + pass class S3RegistryStore(RegistryStore): diff --git a/sdk/python/feast/infra/gcp.py b/sdk/python/feast/infra/gcp.py index c57450c8760..8db4c5fd66e 100644 --- a/sdk/python/feast/infra/gcp.py +++ b/sdk/python/feast/infra/gcp.py @@ -2,153 +2,20 @@ from datetime import datetime from pathlib import Path from tempfile import TemporaryFile -from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union from urllib.parse import urlparse -import pandas -from tqdm import tqdm - -from feast import FeatureTable -from feast.entity import Entity -from feast.feature_view import FeatureView -from feast.infra.offline_stores.offline_utils import get_offline_store_from_config -from feast.infra.online_stores.helpers import get_online_store_from_config -from feast.infra.provider import ( - Provider, - RetrievalJob, - _convert_arrow_to_proto, - _get_column_names, - _run_field_mapping, -) +from feast.infra.provider import PassthroughProvider from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto -from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto -from feast.protos.feast.types.Value_pb2 import Value as ValueProto -from feast.registry import Registry from feast.registry_store import RegistryStore -from feast.repo_config import RegistryConfig, RepoConfig - - -class GcpProvider(Provider): - _gcp_project_id: Optional[str] - _namespace: Optional[str] - - def __init__(self, config: RepoConfig): - self.repo_config = config - self.offline_store = get_offline_store_from_config(config.offline_store) - self.online_store = get_online_store_from_config(config.online_store) - - def update_infra( - self, - project: str, - tables_to_delete: Sequence[Union[FeatureTable, FeatureView]], - tables_to_keep: Sequence[Union[FeatureTable, FeatureView]], - entities_to_delete: Sequence[Entity], - entities_to_keep: Sequence[Entity], - partial: bool, - ): - self.online_store.update( - config=self.repo_config, - tables_to_delete=tables_to_delete, - tables_to_keep=tables_to_keep, - entities_to_keep=entities_to_keep, - entities_to_delete=entities_to_delete, - partial=partial, - ) - - def teardown_infra( - self, - project: str, - tables: Sequence[Union[FeatureTable, FeatureView]], - entities: Sequence[Entity], - ) -> None: - self.online_store.teardown(self.repo_config, tables, entities) - - def online_write_batch( - self, - config: RepoConfig, - table: Union[FeatureTable, FeatureView], - data: List[ - Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] - ], - progress: Optional[Callable[[int], Any]], - ) -> None: - self.online_store.online_write_batch(config, table, data, progress) - - def online_read( - self, - config: RepoConfig, - table: Union[FeatureTable, FeatureView], - entity_keys: List[EntityKeyProto], - requested_features: List[str] = None, - ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]: - result = self.online_store.online_read(config, table, entity_keys) +from feast.repo_config import RegistryConfig - return result - def materialize_single_feature_view( - self, - config: RepoConfig, - feature_view: FeatureView, - start_date: datetime, - end_date: datetime, - registry: Registry, - project: str, - tqdm_builder: Callable[[int], tqdm], - ) -> None: - entities = [] - for entity_name in feature_view.entities: - entities.append(registry.get_entity(entity_name, project)) +class GcpProvider(PassthroughProvider): + """ + This class only exists for backwards compatibility. + """ - ( - join_key_columns, - feature_name_columns, - event_timestamp_column, - 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, - event_timestamp_column=event_timestamp_column, - created_timestamp_column=created_timestamp_column, - start_date=start_date, - end_date=end_date, - ) - table = offline_job.to_arrow() - - if feature_view.batch_source.field_mapping is not None: - table = _run_field_mapping(table, feature_view.batch_source.field_mapping) - - join_keys = [entity.join_key for entity in entities] - rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys) - - with tqdm_builder(len(rows_to_write)) as pbar: - self.online_write_batch( - self.repo_config, feature_view, rows_to_write, lambda x: pbar.update(x) - ) - - def get_historical_features( - self, - config: RepoConfig, - feature_views: List[FeatureView], - feature_refs: List[str], - entity_df: Union[pandas.DataFrame, str], - registry: Registry, - project: str, - full_feature_names: bool, - ) -> RetrievalJob: - job = self.offline_store.get_historical_features( - config=config, - feature_views=feature_views, - feature_refs=feature_refs, - entity_df=entity_df, - registry=registry, - project=project, - full_feature_names=full_feature_names, - ) - return job + pass class GCSRegistryStore(RegistryStore): diff --git a/sdk/python/feast/infra/local.py b/sdk/python/feast/infra/local.py index 7304e262b16..65b84a41175 100644 --- a/sdk/python/feast/infra/local.py +++ b/sdk/python/feast/infra/local.py @@ -1,150 +1,24 @@ import uuid from datetime import datetime from pathlib import Path -from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union +from typing import Union -import pandas as pd import pytz -from tqdm import tqdm from feast import FeatureTable -from feast.entity import Entity from feast.feature_view import FeatureView -from feast.infra.offline_stores.offline_utils import get_offline_store_from_config -from feast.infra.online_stores.helpers import get_online_store_from_config -from feast.infra.provider import ( - Provider, - RetrievalJob, - _convert_arrow_to_proto, - _get_column_names, - _run_field_mapping, -) +from feast.infra.provider import PassthroughProvider from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto -from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto -from feast.protos.feast.types.Value_pb2 import Value as ValueProto -from feast.registry import Registry from feast.registry_store import RegistryStore -from feast.repo_config import RegistryConfig, RepoConfig +from feast.repo_config import RegistryConfig -class LocalProvider(Provider): - def __init__(self, config: RepoConfig): - assert config is not None - self.config = config - self.offline_store = get_offline_store_from_config(config.offline_store) - self.online_store = get_online_store_from_config(config.online_store) +class LocalProvider(PassthroughProvider): + """ + This class only exists for backwards compatibility. + """ - def update_infra( - self, - project: str, - tables_to_delete: Sequence[Union[FeatureTable, FeatureView]], - tables_to_keep: Sequence[Union[FeatureTable, FeatureView]], - entities_to_delete: Sequence[Entity], - entities_to_keep: Sequence[Entity], - partial: bool, - ): - self.online_store.update( - self.config, - tables_to_delete, - tables_to_keep, - entities_to_delete, - entities_to_keep, - partial, - ) - - def teardown_infra( - self, - project: str, - tables: Sequence[Union[FeatureTable, FeatureView]], - entities: Sequence[Entity], - ) -> None: - self.online_store.teardown(self.config, tables, entities) - - def online_write_batch( - self, - config: RepoConfig, - table: Union[FeatureTable, FeatureView], - data: List[ - Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] - ], - progress: Optional[Callable[[int], Any]], - ) -> None: - self.online_store.online_write_batch(config, table, data, progress) - - def online_read( - self, - config: RepoConfig, - table: Union[FeatureTable, FeatureView], - entity_keys: List[EntityKeyProto], - requested_features: List[str] = None, - ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]: - result = self.online_store.online_read(config, table, entity_keys) - - return result - - def materialize_single_feature_view( - self, - config: RepoConfig, - feature_view: FeatureView, - start_date: datetime, - end_date: datetime, - registry: Registry, - project: str, - tqdm_builder: Callable[[int], tqdm], - ) -> None: - entities = [] - for entity_name in feature_view.entities: - entities.append(registry.get_entity(entity_name, project)) - - ( - join_key_columns, - feature_name_columns, - event_timestamp_column, - created_timestamp_column, - ) = _get_column_names(feature_view, entities) - - offline_job = self.offline_store.pull_latest_from_table_or_query( - data_source=feature_view.batch_source, - join_key_columns=join_key_columns, - feature_name_columns=feature_name_columns, - event_timestamp_column=event_timestamp_column, - created_timestamp_column=created_timestamp_column, - start_date=start_date, - end_date=end_date, - config=config, - ) - table = offline_job.to_arrow() - - if feature_view.batch_source.field_mapping is not None: - table = _run_field_mapping(table, feature_view.batch_source.field_mapping) - - join_keys = [entity.join_key for entity in entities] - rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys) - - with tqdm_builder(len(rows_to_write)) as pbar: - self.online_write_batch( - self.config, feature_view, rows_to_write, lambda x: pbar.update(x) - ) - - def get_historical_features( - self, - config: RepoConfig, - feature_views: List[FeatureView], - feature_refs: List[str], - entity_df: Union[pd.DataFrame, str], - registry: Registry, - project: str, - full_feature_names: bool, - ) -> RetrievalJob: - return self.offline_store.get_historical_features( - config=config, - feature_views=feature_views, - feature_refs=feature_refs, - entity_df=entity_df, - registry=registry, - project=project, - full_feature_names=full_feature_names, - ) + pass def _table_id(project: str, table: Union[FeatureTable, FeatureView]) -> str: diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py index 54c2ee94fbd..60e20f996a8 100644 --- a/sdk/python/feast/infra/provider.py +++ b/sdk/python/feast/infra/provider.py @@ -13,6 +13,8 @@ from feast.feature_table import FeatureTable from feast.feature_view import DUMMY_ENTITY_ID, FeatureView from feast.infra.offline_stores.offline_store import RetrievalJob +from feast.infra.offline_stores.offline_utils import get_offline_store_from_config +from feast.infra.online_stores.helpers import get_online_store_from_config from feast.on_demand_feature_view import OnDemandFeatureView from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto @@ -144,6 +146,132 @@ def online_read( ... +class PassthroughProvider(Provider): + """ + The Passthrough provider delegates all operations to the underlying online and offline stores. + """ + + def __init__(self, config: RepoConfig): + super().__init__(config) + self.repo_config = config + self.offline_store = get_offline_store_from_config(config.offline_store) + self.online_store = get_online_store_from_config(config.online_store) + + def update_infra( + self, + project: str, + tables_to_delete: Sequence[Union[FeatureTable, FeatureView]], + tables_to_keep: Sequence[Union[FeatureTable, FeatureView]], + entities_to_delete: Sequence[Entity], + entities_to_keep: Sequence[Entity], + partial: bool, + ): + self.online_store.update( + config=self.repo_config, + tables_to_delete=tables_to_delete, + tables_to_keep=tables_to_keep, + entities_to_keep=entities_to_keep, + entities_to_delete=entities_to_delete, + partial=partial, + ) + + def teardown_infra( + self, + project: str, + tables: Sequence[Union[FeatureTable, FeatureView]], + entities: Sequence[Entity], + ) -> None: + self.online_store.teardown(self.repo_config, tables, entities) + + def online_write_batch( + self, + config: RepoConfig, + table: Union[FeatureTable, FeatureView], + data: List[ + Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] + ], + progress: Optional[Callable[[int], Any]], + ) -> None: + self.online_store.online_write_batch(config, table, data, progress) + + def online_read( + self, + config: RepoConfig, + table: Union[FeatureTable, FeatureView], + entity_keys: List[EntityKeyProto], + requested_features: List[str] = None, + ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]: + result = self.online_store.online_read(config, table, entity_keys) + + return result + + def materialize_single_feature_view( + self, + config: RepoConfig, + feature_view: FeatureView, + start_date: datetime, + end_date: datetime, + registry: Registry, + project: str, + tqdm_builder: Callable[[int], tqdm], + ) -> None: + entities = [] + for entity_name in feature_view.entities: + entities.append(registry.get_entity(entity_name, project)) + + ( + join_key_columns, + feature_name_columns, + event_timestamp_column, + 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, + event_timestamp_column=event_timestamp_column, + created_timestamp_column=created_timestamp_column, + start_date=start_date, + end_date=end_date, + ) + + table = offline_job.to_arrow() + + if feature_view.batch_source.field_mapping is not None: + table = _run_field_mapping(table, feature_view.batch_source.field_mapping) + + join_keys = [entity.join_key for entity in entities] + rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys) + + with tqdm_builder(len(rows_to_write)) as pbar: + self.online_write_batch( + self.repo_config, feature_view, rows_to_write, lambda x: pbar.update(x) + ) + + def get_historical_features( + self, + config: RepoConfig, + feature_views: List[FeatureView], + feature_refs: List[str], + entity_df: Union[pandas.DataFrame, str], + registry: Registry, + project: str, + full_feature_names: bool, + ) -> RetrievalJob: + job = self.offline_store.get_historical_features( + config=config, + feature_views=feature_views, + feature_refs=feature_refs, + entity_df=entity_df, + registry=registry, + project=project, + full_feature_names=full_feature_names, + ) + return job + + def get_provider(config: RepoConfig, repo_path: Path) -> Provider: if "." not in config.provider: if config.provider == "gcp": From b1192218bf23e59ce4b691bf522d09213c597a71 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 16 Sep 2021 16:06:33 -0700 Subject: [PATCH 2/8] Refactor Signed-off-by: Achal Shah --- sdk/python/feast/infra/provider.py | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py index 60e20f996a8..bf49118c076 100644 --- a/sdk/python/feast/infra/provider.py +++ b/sdk/python/feast/infra/provider.py @@ -274,18 +274,8 @@ def get_historical_features( def get_provider(config: RepoConfig, repo_path: Path) -> Provider: if "." not in config.provider: - if config.provider == "gcp": - from feast.infra.gcp import GcpProvider - - return GcpProvider(config) - elif config.provider == "aws": - from feast.infra.aws import AwsProvider - - return AwsProvider(config) - elif config.provider == "local": - from feast.infra.local import LocalProvider - - return LocalProvider(config) + if config.provider in {"gcp", "aws", "local"}: + return PassthroughProvider(config) else: raise errors.FeastProviderNotImplementedError(config.provider) else: From 3c55be40b8eb40dd6594fe87a7dd56175fbd1316 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 16 Sep 2021 16:14:43 -0700 Subject: [PATCH 3/8] refactor Signed-off-by: Achal Shah --- sdk/python/feast/infra/provider.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py index bf49118c076..71dee202871 100644 --- a/sdk/python/feast/infra/provider.py +++ b/sdk/python/feast/infra/provider.py @@ -13,8 +13,6 @@ from feast.feature_table import FeatureTable from feast.feature_view import DUMMY_ENTITY_ID, FeatureView from feast.infra.offline_stores.offline_store import RetrievalJob -from feast.infra.offline_stores.offline_utils import get_offline_store_from_config -from feast.infra.online_stores.helpers import get_online_store_from_config from feast.on_demand_feature_view import OnDemandFeatureView from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto @@ -153,6 +151,11 @@ class PassthroughProvider(Provider): def __init__(self, config: RepoConfig): super().__init__(config) + from feast.infra.offline_stores.offline_utils import ( + get_offline_store_from_config, + ) + from feast.infra.online_stores.helpers import get_online_store_from_config + self.repo_config = config self.offline_store = get_offline_store_from_config(config.offline_store) self.online_store = get_online_store_from_config(config.online_store) From 748a77ce394137cb0a5ebd328de9aacba30b3b45 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 16 Sep 2021 16:20:00 -0700 Subject: [PATCH 4/8] refactor Signed-off-by: Achal Shah --- sdk/__init__.py | 0 sdk/python/__init__.py | 0 sdk/python/feast/infra/aws.py | 2 +- sdk/python/feast/infra/gcp.py | 2 +- sdk/python/feast/infra/local.py | 2 +- .../feast/infra/passthrough_provider.py | 146 ++++++++++++++++++ sdk/python/feast/infra/provider.py | 132 +--------------- 7 files changed, 150 insertions(+), 134 deletions(-) create mode 100644 sdk/__init__.py create mode 100644 sdk/python/__init__.py create mode 100644 sdk/python/feast/infra/passthrough_provider.py diff --git a/sdk/__init__.py b/sdk/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/sdk/python/__init__.py b/sdk/python/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/sdk/python/feast/infra/aws.py b/sdk/python/feast/infra/aws.py index 17971c3c0c4..0f4f2e07381 100644 --- a/sdk/python/feast/infra/aws.py +++ b/sdk/python/feast/infra/aws.py @@ -6,7 +6,7 @@ from urllib.parse import urlparse from feast.errors import S3RegistryBucketForbiddenAccess, S3RegistryBucketNotExist -from feast.infra.provider import PassthroughProvider +from feast.infra.passthrough_provider import PassthroughProvider from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto from feast.registry_store import RegistryStore from feast.repo_config import RegistryConfig diff --git a/sdk/python/feast/infra/gcp.py b/sdk/python/feast/infra/gcp.py index 8db4c5fd66e..1dd1eefe2dd 100644 --- a/sdk/python/feast/infra/gcp.py +++ b/sdk/python/feast/infra/gcp.py @@ -4,7 +4,7 @@ from tempfile import TemporaryFile from urllib.parse import urlparse -from feast.infra.provider import PassthroughProvider +from feast.infra.passthrough_provider import PassthroughProvider from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto from feast.registry_store import RegistryStore from feast.repo_config import RegistryConfig diff --git a/sdk/python/feast/infra/local.py b/sdk/python/feast/infra/local.py index 65b84a41175..6ac3ce72591 100644 --- a/sdk/python/feast/infra/local.py +++ b/sdk/python/feast/infra/local.py @@ -7,7 +7,7 @@ from feast import FeatureTable from feast.feature_view import FeatureView -from feast.infra.provider import PassthroughProvider +from feast.infra.passthrough_provider import PassthroughProvider from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto from feast.registry_store import RegistryStore from feast.repo_config import RegistryConfig diff --git a/sdk/python/feast/infra/passthrough_provider.py b/sdk/python/feast/infra/passthrough_provider.py new file mode 100644 index 00000000000..ddcc11fc10c --- /dev/null +++ b/sdk/python/feast/infra/passthrough_provider.py @@ -0,0 +1,146 @@ +from datetime import datetime +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union + +import pandas +from tqdm import tqdm + +from feast import Entity, FeatureTable, FeatureView, RepoConfig +from feast.infra.offline_stores.offline_store import RetrievalJob +from feast.infra.offline_stores.offline_utils import get_offline_store_from_config +from feast.infra.online_stores.helpers import get_online_store_from_config +from feast.infra.provider import ( + Provider, + _convert_arrow_to_proto, + _get_column_names, + _run_field_mapping, +) +from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.protos.feast.types.Value_pb2 import Value as ValueProto +from feast.registry import Registry + + +class PassthroughProvider(Provider): + """ + The Passthrough provider delegates all operations to the underlying online and offline stores. + """ + + def __init__(self, config: RepoConfig): + super().__init__(config) + + self.repo_config = config + self.offline_store = get_offline_store_from_config(config.offline_store) + self.online_store = get_online_store_from_config(config.online_store) + + def update_infra( + self, + project: str, + tables_to_delete: Sequence[Union[FeatureTable, FeatureView]], + tables_to_keep: Sequence[Union[FeatureTable, FeatureView]], + entities_to_delete: Sequence[Entity], + entities_to_keep: Sequence[Entity], + partial: bool, + ): + self.online_store.update( + config=self.repo_config, + tables_to_delete=tables_to_delete, + tables_to_keep=tables_to_keep, + entities_to_keep=entities_to_keep, + entities_to_delete=entities_to_delete, + partial=partial, + ) + + def teardown_infra( + self, + project: str, + tables: Sequence[Union[FeatureTable, FeatureView]], + entities: Sequence[Entity], + ) -> None: + self.online_store.teardown(self.repo_config, tables, entities) + + def online_write_batch( + self, + config: RepoConfig, + table: Union[FeatureTable, FeatureView], + data: List[ + Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] + ], + progress: Optional[Callable[[int], Any]], + ) -> None: + self.online_store.online_write_batch(config, table, data, progress) + + def online_read( + self, + config: RepoConfig, + table: Union[FeatureTable, FeatureView], + entity_keys: List[EntityKeyProto], + requested_features: List[str] = None, + ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]: + result = self.online_store.online_read(config, table, entity_keys) + + return result + + def materialize_single_feature_view( + self, + config: RepoConfig, + feature_view: FeatureView, + start_date: datetime, + end_date: datetime, + registry: Registry, + project: str, + tqdm_builder: Callable[[int], tqdm], + ) -> None: + entities = [] + for entity_name in feature_view.entities: + entities.append(registry.get_entity(entity_name, project)) + + ( + join_key_columns, + feature_name_columns, + event_timestamp_column, + 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, + event_timestamp_column=event_timestamp_column, + created_timestamp_column=created_timestamp_column, + start_date=start_date, + end_date=end_date, + ) + + table = offline_job.to_arrow() + + if feature_view.batch_source.field_mapping is not None: + table = _run_field_mapping(table, feature_view.batch_source.field_mapping) + + join_keys = [entity.join_key for entity in entities] + rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys) + + with tqdm_builder(len(rows_to_write)) as pbar: + self.online_write_batch( + self.repo_config, feature_view, rows_to_write, lambda x: pbar.update(x) + ) + + def get_historical_features( + self, + config: RepoConfig, + feature_views: List[FeatureView], + feature_refs: List[str], + entity_df: Union[pandas.DataFrame, str], + registry: Registry, + project: str, + full_feature_names: bool, + ) -> RetrievalJob: + job = self.offline_store.get_historical_features( + config=config, + feature_views=feature_views, + feature_refs=feature_refs, + entity_df=entity_df, + registry=registry, + project=project, + full_feature_names=full_feature_names, + ) + return job diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py index 71dee202871..0c8d8e08365 100644 --- a/sdk/python/feast/infra/provider.py +++ b/sdk/python/feast/infra/provider.py @@ -13,6 +13,7 @@ from feast.feature_table import FeatureTable from feast.feature_view import DUMMY_ENTITY_ID, FeatureView from feast.infra.offline_stores.offline_store import RetrievalJob +from feast.infra.passthrough_provider import PassthroughProvider from feast.on_demand_feature_view import OnDemandFeatureView from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto @@ -144,137 +145,6 @@ def online_read( ... -class PassthroughProvider(Provider): - """ - The Passthrough provider delegates all operations to the underlying online and offline stores. - """ - - def __init__(self, config: RepoConfig): - super().__init__(config) - from feast.infra.offline_stores.offline_utils import ( - get_offline_store_from_config, - ) - from feast.infra.online_stores.helpers import get_online_store_from_config - - self.repo_config = config - self.offline_store = get_offline_store_from_config(config.offline_store) - self.online_store = get_online_store_from_config(config.online_store) - - def update_infra( - self, - project: str, - tables_to_delete: Sequence[Union[FeatureTable, FeatureView]], - tables_to_keep: Sequence[Union[FeatureTable, FeatureView]], - entities_to_delete: Sequence[Entity], - entities_to_keep: Sequence[Entity], - partial: bool, - ): - self.online_store.update( - config=self.repo_config, - tables_to_delete=tables_to_delete, - tables_to_keep=tables_to_keep, - entities_to_keep=entities_to_keep, - entities_to_delete=entities_to_delete, - partial=partial, - ) - - def teardown_infra( - self, - project: str, - tables: Sequence[Union[FeatureTable, FeatureView]], - entities: Sequence[Entity], - ) -> None: - self.online_store.teardown(self.repo_config, tables, entities) - - def online_write_batch( - self, - config: RepoConfig, - table: Union[FeatureTable, FeatureView], - data: List[ - Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] - ], - progress: Optional[Callable[[int], Any]], - ) -> None: - self.online_store.online_write_batch(config, table, data, progress) - - def online_read( - self, - config: RepoConfig, - table: Union[FeatureTable, FeatureView], - entity_keys: List[EntityKeyProto], - requested_features: List[str] = None, - ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]: - result = self.online_store.online_read(config, table, entity_keys) - - return result - - def materialize_single_feature_view( - self, - config: RepoConfig, - feature_view: FeatureView, - start_date: datetime, - end_date: datetime, - registry: Registry, - project: str, - tqdm_builder: Callable[[int], tqdm], - ) -> None: - entities = [] - for entity_name in feature_view.entities: - entities.append(registry.get_entity(entity_name, project)) - - ( - join_key_columns, - feature_name_columns, - event_timestamp_column, - 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, - event_timestamp_column=event_timestamp_column, - created_timestamp_column=created_timestamp_column, - start_date=start_date, - end_date=end_date, - ) - - table = offline_job.to_arrow() - - if feature_view.batch_source.field_mapping is not None: - table = _run_field_mapping(table, feature_view.batch_source.field_mapping) - - join_keys = [entity.join_key for entity in entities] - rows_to_write = _convert_arrow_to_proto(table, feature_view, join_keys) - - with tqdm_builder(len(rows_to_write)) as pbar: - self.online_write_batch( - self.repo_config, feature_view, rows_to_write, lambda x: pbar.update(x) - ) - - def get_historical_features( - self, - config: RepoConfig, - feature_views: List[FeatureView], - feature_refs: List[str], - entity_df: Union[pandas.DataFrame, str], - registry: Registry, - project: str, - full_feature_names: bool, - ) -> RetrievalJob: - job = self.offline_store.get_historical_features( - config=config, - feature_views=feature_views, - feature_refs=feature_refs, - entity_df=entity_df, - registry=registry, - project=project, - full_feature_names=full_feature_names, - ) - return job - - def get_provider(config: RepoConfig, repo_path: Path) -> Provider: if "." not in config.provider: if config.provider in {"gcp", "aws", "local"}: From 8d030fd0499ebea645fc3d4907138c3fcf1cba11 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 16 Sep 2021 16:31:51 -0700 Subject: [PATCH 5/8] fix imports Signed-off-by: Achal Shah --- sdk/python/feast/infra/passthrough_provider.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/sdk/python/feast/infra/passthrough_provider.py b/sdk/python/feast/infra/passthrough_provider.py index ddcc11fc10c..3e4c0a34858 100644 --- a/sdk/python/feast/infra/passthrough_provider.py +++ b/sdk/python/feast/infra/passthrough_provider.py @@ -4,7 +4,9 @@ import pandas from tqdm import tqdm -from feast import Entity, FeatureTable, FeatureView, RepoConfig +from feast.entity import Entity +from feast.feature_table import FeatureTable +from feast.feature_view import FeatureView from feast.infra.offline_stores.offline_store import RetrievalJob from feast.infra.offline_stores.offline_utils import get_offline_store_from_config from feast.infra.online_stores.helpers import get_online_store_from_config @@ -17,6 +19,7 @@ from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto from feast.registry import Registry +from feast.repo_config import RepoConfig class PassthroughProvider(Provider): From 71b21bdacb5cce532001e96237ae12e5e2b27f0c Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 16 Sep 2021 16:34:43 -0700 Subject: [PATCH 6/8] Dynamic import for passthru Signed-off-by: Achal Shah --- sdk/python/feast/infra/provider.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py index 0c8d8e08365..fe0a14b9921 100644 --- a/sdk/python/feast/infra/provider.py +++ b/sdk/python/feast/infra/provider.py @@ -13,7 +13,6 @@ from feast.feature_table import FeatureTable from feast.feature_view import DUMMY_ENTITY_ID, FeatureView from feast.infra.offline_stores.offline_store import RetrievalJob -from feast.infra.passthrough_provider import PassthroughProvider from feast.on_demand_feature_view import OnDemandFeatureView from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto @@ -148,6 +147,7 @@ def online_read( def get_provider(config: RepoConfig, repo_path: Path) -> Provider: if "." not in config.provider: if config.provider in {"gcp", "aws", "local"}: + from feast.infra.passthrough_provider import PassthroughProvider return PassthroughProvider(config) else: raise errors.FeastProviderNotImplementedError(config.provider) From 99ec0e2614e8aea1e8d64a470f96e5ad7d323a98 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 16 Sep 2021 22:48:09 -0700 Subject: [PATCH 7/8] Dynamic import for passthru Signed-off-by: Achal Shah --- sdk/python/feast/infra/provider.py | 1 + 1 file changed, 1 insertion(+) diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py index fe0a14b9921..6147f19b9af 100644 --- a/sdk/python/feast/infra/provider.py +++ b/sdk/python/feast/infra/provider.py @@ -148,6 +148,7 @@ def get_provider(config: RepoConfig, repo_path: Path) -> Provider: if "." not in config.provider: if config.provider in {"gcp", "aws", "local"}: from feast.infra.passthrough_provider import PassthroughProvider + return PassthroughProvider(config) else: raise errors.FeastProviderNotImplementedError(config.provider) From 6ab8c3f3c672474dd0af80fdcb920f250deb6c73 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Fri, 17 Sep 2021 09:57:23 -0700 Subject: [PATCH 8/8] remove init files Signed-off-by: Achal Shah --- sdk/__init__.py | 0 sdk/python/__init__.py | 0 2 files changed, 0 insertions(+), 0 deletions(-) delete mode 100644 sdk/__init__.py delete mode 100644 sdk/python/__init__.py diff --git a/sdk/__init__.py b/sdk/__init__.py deleted file mode 100644 index e69de29bb2d..00000000000 diff --git a/sdk/python/__init__.py b/sdk/python/__init__.py deleted file mode 100644 index e69de29bb2d..00000000000