From d01836b81f5e7433f9732bb9096dfd56dc332da0 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Thu, 11 Mar 2021 11:26:30 -0500 Subject: [PATCH 01/12] WIP Ingest into Firestore Signed-off-by: Jacob Klegar --- sdk/python/feast/online_store.py | 66 ++++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 sdk/python/feast/online_store.py diff --git a/sdk/python/feast/online_store.py b/sdk/python/feast/online_store.py new file mode 100644 index 00000000000..9de08c43269 --- /dev/null +++ b/sdk/python/feast/online_store.py @@ -0,0 +1,66 @@ +# Copyright 2019 The Feast Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from abc import ABC, abstractmethod +from typing import Any + +import pyarrow + +from feast.feature_view import FeatureView +from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.types.Value_pb2 import Value as ValueProto + + +class OnlineStore(ABC): + @abstractmethod + def ingest(self, table: pyarrow.Table, feature_view: FeatureView): + pass + + +class FirestoreOnlineStore(OnlineStore): + def ingest(self, table: pyarrow.Table, feature_view: FeatureView): + rows_to_write = [] + for row in zip(*table.to_pydict().values()): + entity_key = EntityKeyProto() + for entity in feature_view.entities: + entity_key.entity_names.append(entity.name) + idx = table.column_names.index(entity.name) + value = _convert_to_proto(row[idx]) + entity_key.entity_values.append(value) + feature_dict = {} + for feature in feature_view.features: + idx = table.column_names.index(feature.name) + value = _convert_to_proto(row[idx]) + feature_dict[feature.name] = value + event_timestamp_idx = table.column_names.index( + feature_view.inputs.event_timestamp_column + ) + rows_to_write.append((entity_key, feature_dict, row[event_timestamp_idx])) + print(rows_to_write) + + +def _convert_to_proto(value: Any) -> ValueProto: + value_proto = ValueProto() + if isinstance(value, str): + value_proto.string_val = value + elif isinstance(value, bool): + value_proto.bool_val = value + elif isinstance(value, int): + value_proto.int32_val = value + elif isinstance(value, float): + value_proto.double_val = value + elif isinstance(value, bytes): + value_proto.bytes_val = value + else: + raise ValueError(f"Cannot convert value {value} of type {type(value)}.") + return value_proto From 8d429dc8ae07defaa4ed61baf8c174cfd99a6a6b Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Mon, 15 Mar 2021 19:50:27 -0400 Subject: [PATCH 02/12] Full materialize function Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 159 +++++++++++++++++++++++++++++- sdk/python/feast/offline_store.py | 100 +++++-------------- sdk/python/feast/online_store.py | 66 ------------- 3 files changed, 180 insertions(+), 145 deletions(-) delete mode 100644 sdk/python/feast/online_store.py diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index ea3a4d29e11..78ce72569cb 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -11,15 +11,17 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +from datetime import datetime from pathlib import Path -from typing import List, Optional, Union +from typing import Any, Dict, List, Optional, Tuple, Type, Union import pandas as pd +import pyarrow from feast.entity import Entity from feast.feature_view import FeatureView from feast.infra.provider import Provider, get_provider -from feast.offline_store import RetrievalJob, get_offline_store_for_retrieval +from feast.offline_store import OfflineStore, get_offline_store, RetrievalJob, get_offline_store_for_retrieval from feast.registry import Registry from feast.repo_config import ( LocalOnlineStoreConfig, @@ -27,6 +29,8 @@ RepoConfig, load_repo_config, ) +from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.types.Value_pb2 import Value as ValueProto class FeatureStore: @@ -58,6 +62,9 @@ def __init__( def _get_provider(self) -> Provider: return get_provider(self.config) + def _get_offline_store(self) -> Type[OfflineStore]: + return get_offline_store(self.config) + def _get_registry(self) -> Registry: return Registry(self.config.metadata_store) @@ -176,3 +183,151 @@ def _get_requested_feature_views( feature_views_list.append(view) return feature_views_list + + @property + def project(self) -> str: + return "default" + + def materialize( + self, feature_views: List[str], start_date: datetime, end_date: datetime + ): + full_feature_views = [] + registry = self._get_registry() + for name in feature_views: + feature_view = registry.get_feature_view(name, self.project) + full_feature_views.append(feature_view) + if feature_view.input.table_ref is None: + raise NotImplementedError( + f"This function is only implemented for FeatureViews with a table_ref; {feature_view.name} does not have one." + ) + + # TODO paging large loads + for feature_view in full_feature_views: + ( + entity_names, + feature_names, + event_timestamp_column, + created_timestamp_column, + ) = _run_reverse_field_mapping(feature_view) + + offline_store = self._get_offline_store() + table = offline_store.pull_latest_from_table( + feature_view.input.table_ref, + entity_names, + feature_names, + event_timestamp_column, + created_timestamp_column, + start_date, + end_date, + ) + + if feature_view.input.field_mapping is not None: + table = _run_forward_field_mapping( + table, feature_view.input.field_mapping + ) + + rows_to_write = _convert_arrow_to_proto(table, feature_view) + + provider = self._get_provider() + provider.online_write_batch( + "default", feature_view, rows_to_write, created_timestamp_column + ) + + +def _run_reverse_field_mapping( + feature_view: FeatureView, +) -> Tuple[List[str], List[str], str, Optional[str]]: + """ + If a field mapping exists, run it in reverse on the entity names, feature names, event timestamp column, and created timestamp column to get the names of the relevant columns in the BigQuery table. + + Args: + feature_view: FeatureView object containing the field mapping as well as the names to reverse-map. + Returns: + Tuple containing the list of reverse-mapped entity names, reverse-mapped feature names, reverse-mapped event timestamp column, and reverse-mapped created timestamp column that will be passed into the query to the offline store. + """ + # if we have mapped fields, use the original field names in the call to the offline store + event_timestamp_column = feature_view.input.event_timestamp_column + entity_names = [entity for entity in feature_view.entities] + feature_names = [feature.name for feature in feature_view.features] + created_timestamp_column = feature_view.input.created_timestamp_column + if feature_view.input.field_mapping is not None: + reverse_field_mapping = { + v: k for k, v in feature_view.input.field_mapping.items() + } + event_timestamp_column = ( + reverse_field_mapping[event_timestamp_column] + if event_timestamp_column in reverse_field_mapping.keys() + else event_timestamp_column + ) + created_timestamp_column = ( + reverse_field_mapping[created_timestamp_column] + if created_timestamp_column is not None + and created_timestamp_column in reverse_field_mapping.keys() + else created_timestamp_column + ) + entity_names = [ + reverse_field_mapping[col] if col in reverse_field_mapping.keys() else col + for col in entity_names + ] + feature_names = [ + reverse_field_mapping[col] if col in reverse_field_mapping.keys() else col + for col in feature_names + ] + return ( + entity_names, + feature_names, + event_timestamp_column, + created_timestamp_column, + ) + + +def _run_forward_field_mapping( + table: pyarrow.Table, field_mapping: Dict[str, str], +) -> pyarrow.Table: + # run field mapping in the forward direction + cols = table.column_names + mapped_cols = [ + field_mapping[col] if col in field_mapping.keys() else col for col in cols + ] + table = table.rename_columns(mapped_cols) + return table + + +def _convert_arrow_to_proto( + table: pyarrow.Table, feature_view: FeatureView +) -> List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime]]: + rows_to_write = [] + for row in zip(*table.to_pydict().values()): + entity_key = EntityKeyProto() + for entity_name in feature_view.entities: + entity_key.entity_names.append(entity_name) + idx = table.column_names.index(entity_name) + value = _convert_to_proto(row[idx]) + entity_key.entity_values.append(value) + feature_dict = {} + for feature in feature_view.features: + idx = table.column_names.index(feature.name) + value = _convert_to_proto(row[idx]) + feature_dict[feature.name] = value + event_timestamp_idx = table.column_names.index( + feature_view.input.event_timestamp_column + ) + rows_to_write.append((entity_key, feature_dict, row[event_timestamp_idx])) + return rows_to_write + + +def _convert_to_proto(value: Any) -> ValueProto: + value_proto = ValueProto() + if isinstance(value, str): + value_proto.string_val = value + elif isinstance(value, bool): + value_proto.bool_val = value + elif isinstance(value, int): + value_proto.int32_val = value + elif isinstance(value, float): + value_proto.double_val = value + elif isinstance(value, bytes): + value_proto.bytes_val = value + else: + raise ValueError(f"Cannot convert value {value} of type {type(value)}.") + return value_proto diff --git a/sdk/python/feast/offline_store.py b/sdk/python/feast/offline_store.py index dbd1503ded3..7effb6e7e96 100644 --- a/sdk/python/feast/offline_store.py +++ b/sdk/python/feast/offline_store.py @@ -15,7 +15,7 @@ from abc import ABC, abstractmethod from dataclasses import asdict, dataclass from datetime import datetime, timedelta -from typing import Callable, Dict, List, Optional, Tuple, Union +from typing import Callable, Dict, List, Optional, Tuple, Type, Union import pandas as pd import pyarrow @@ -87,7 +87,13 @@ class OfflineStore(ABC): @staticmethod @abstractmethod def pull_latest_from_table( - feature_view: FeatureView, start_date: datetime, end_date: datetime, + table_ref: str, + entity_names: List[str], + feature_names: List[str], + event_timestamp_column: str, + created_timestamp_column: Optional[str], + start_date: datetime, + end_date: datetime, ) -> Optional[pyarrow.Table]: pass @@ -105,21 +111,19 @@ def get_historical_features( class BigQueryOfflineStore(OfflineStore): @staticmethod def pull_latest_from_table( - feature_view: FeatureView, start_date: datetime, end_date: datetime, + table_ref: str, + entity_names: List[str], + feature_names: List[str], + event_timestamp_column: str, + created_timestamp_column: Optional[str], + start_date: datetime, + end_date: datetime, ) -> pyarrow.Table: - assert isinstance(feature_view.input, BigQuerySource) - if feature_view.input.table_ref is None: + if table_ref is None: raise ValueError( "This function can only be called on a FeatureView with a table_ref" ) - ( - entity_names, - feature_names, - event_timestamp_column, - created_timestamp_column, - ) = run_reverse_field_mapping(feature_view) - partition_by_entity_string = ", ".join(entity_names) if partition_by_entity_string != "": partition_by_entity_string = "PARTITION BY " + partition_by_entity_string @@ -134,14 +138,13 @@ def pull_latest_from_table( FROM ( SELECT {field_string}, ROW_NUMBER() OVER({partition_by_entity_string} ORDER BY {timestamp_desc_string}) AS _feast_row - FROM `{feature_view.input.table_ref}` + FROM `{table_ref}` WHERE {event_timestamp_column} BETWEEN TIMESTAMP('{start_date}') AND TIMESTAMP('{end_date}') ) WHERE _feast_row = 1 """ table = BigQueryOfflineStore._pull_query(query) - table = run_forward_field_mapping(table, feature_view) return table @staticmethod @@ -385,69 +388,6 @@ def evaluate_historical_retrieval(): return job -def run_reverse_field_mapping( - feature_view: FeatureView, -) -> Tuple[List[str], List[str], str, Optional[str]]: - """ - If a field mapping exists, run it in reverse on the entity names, feature names, event timestamp column, and created timestamp column to get the names of the relevant columns in the BigQuery table. - - Args: - feature_view: FeatureView object containing the field mapping as well as the names to reverse-map. - Returns: - Tuple containing the list of reverse-mapped entity names, reverse-mapped feature names, reverse-mapped event timestamp column, and reverse-mapped created timestamp column that will be passed into the query to the offline store. - """ - # if we have mapped fields, use the original field names in the call to the offline store - event_timestamp_column = feature_view.input.event_timestamp_column - entity_names = [entity for entity in feature_view.entities] - feature_names = [feature.name for feature in feature_view.features] - created_timestamp_column = feature_view.input.created_timestamp_column - if feature_view.input.field_mapping is not None: - reverse_field_mapping = { - v: k for k, v in feature_view.input.field_mapping.items() - } - event_timestamp_column = ( - reverse_field_mapping[event_timestamp_column] - if event_timestamp_column in reverse_field_mapping.keys() - else event_timestamp_column - ) - created_timestamp_column = ( - reverse_field_mapping[created_timestamp_column] - if created_timestamp_column is not None - and created_timestamp_column in reverse_field_mapping.keys() - else created_timestamp_column - ) - entity_names = [ - reverse_field_mapping[col] if col in reverse_field_mapping.keys() else col - for col in entity_names - ] - feature_names = [ - reverse_field_mapping[col] if col in reverse_field_mapping.keys() else col - for col in feature_names - ] - return ( - entity_names, - feature_names, - event_timestamp_column, - created_timestamp_column, - ) - - -def run_forward_field_mapping( - table: pyarrow.Table, feature_view: FeatureView -) -> pyarrow.Table: - # run field mapping in the forward direction - if table is not None and feature_view.input.field_mapping is not None: - cols = table.column_names - mapped_cols = [ - feature_view.input.field_mapping[col] - if col in feature_view.input.field_mapping.keys() - else col - for col in cols - ] - table = table.rename_columns(mapped_cols) - return table - - def get_offline_store_for_retrieval(feature_views: List[FeatureView],) -> OfflineStore: """Detect which offline store should be used for retrieving historical features""" @@ -612,3 +552,9 @@ def _get_requested_feature_views_to_features_dict( {% endfor %} ORDER BY event_timestamp """ + +def get_offline_store(config: RepoConfig) -> Type[OfflineStore]: + if config.provider == "gcp": + return BigQueryOfflineStore + else: + raise ValueError(config) diff --git a/sdk/python/feast/online_store.py b/sdk/python/feast/online_store.py deleted file mode 100644 index 9de08c43269..00000000000 --- a/sdk/python/feast/online_store.py +++ /dev/null @@ -1,66 +0,0 @@ -# Copyright 2019 The Feast Authors -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# https://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -from abc import ABC, abstractmethod -from typing import Any - -import pyarrow - -from feast.feature_view import FeatureView -from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto -from feast.types.Value_pb2 import Value as ValueProto - - -class OnlineStore(ABC): - @abstractmethod - def ingest(self, table: pyarrow.Table, feature_view: FeatureView): - pass - - -class FirestoreOnlineStore(OnlineStore): - def ingest(self, table: pyarrow.Table, feature_view: FeatureView): - rows_to_write = [] - for row in zip(*table.to_pydict().values()): - entity_key = EntityKeyProto() - for entity in feature_view.entities: - entity_key.entity_names.append(entity.name) - idx = table.column_names.index(entity.name) - value = _convert_to_proto(row[idx]) - entity_key.entity_values.append(value) - feature_dict = {} - for feature in feature_view.features: - idx = table.column_names.index(feature.name) - value = _convert_to_proto(row[idx]) - feature_dict[feature.name] = value - event_timestamp_idx = table.column_names.index( - feature_view.inputs.event_timestamp_column - ) - rows_to_write.append((entity_key, feature_dict, row[event_timestamp_idx])) - print(rows_to_write) - - -def _convert_to_proto(value: Any) -> ValueProto: - value_proto = ValueProto() - if isinstance(value, str): - value_proto.string_val = value - elif isinstance(value, bool): - value_proto.bool_val = value - elif isinstance(value, int): - value_proto.int32_val = value - elif isinstance(value, float): - value_proto.double_val = value - elif isinstance(value, bytes): - value_proto.bytes_val = value - else: - raise ValueError(f"Cannot convert value {value} of type {type(value)}.") - return value_proto From 4e5ae8874929def2323b3d1a4da9e758876d974d Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Tue, 16 Mar 2021 11:40:54 -0400 Subject: [PATCH 03/12] Rebase Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 48 +++++++++++++++---------------- 1 file changed, 24 insertions(+), 24 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 78ce72569cb..11fead95785 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -160,30 +160,6 @@ def get_historical_features( ) return job - -def _get_requested_feature_views( - feature_refs: List[str], all_feature_views: List[FeatureView] -) -> List[FeatureView]: - """Get list of feature views based on feature references""" - - feature_views_dict = {} - for ref in feature_refs: - ref_parts = ref.split(":") - found = False - for feature_view in all_feature_views: - if feature_view.name == ref_parts[0]: - found = True - feature_views_dict[feature_view.name] = feature_view - continue - - if not found: - raise ValueError(f"Could not find feature view from reference {ref}") - feature_views_list = [] - for view in feature_views_dict.values(): - feature_views_list.append(view) - - return feature_views_list - @property def project(self) -> str: return "default" @@ -234,6 +210,30 @@ def materialize( ) +def _get_requested_feature_views( + feature_refs: List[str], all_feature_views: List[FeatureView] +) -> List[FeatureView]: + """Get list of feature views based on feature references""" + + feature_views_dict = {} + for ref in feature_refs: + ref_parts = ref.split(":") + found = False + for feature_view in all_feature_views: + if feature_view.name == ref_parts[0]: + found = True + feature_views_dict[feature_view.name] = feature_view + continue + + if not found: + raise ValueError(f"Could not find feature view from reference {ref}") + feature_views_list = [] + for view in feature_views_dict.values(): + feature_views_list.append(view) + + return feature_views_list + + def _run_reverse_field_mapping( feature_view: FeatureView, ) -> Tuple[List[str], List[str], str, Optional[str]]: From 969d5568e3ec37c66361582b693645af32e53bef Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Tue, 16 Mar 2021 13:30:01 -0400 Subject: [PATCH 04/12] Add basic ingestion integration test Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 20 ++++-- sdk/python/feast/offline_store.py | 13 +++- sdk/python/tests/test_bigquery_ingestion.py | 76 +++++++++++++++++++++ 3 files changed, 101 insertions(+), 8 deletions(-) create mode 100644 sdk/python/tests/test_bigquery_ingestion.py diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 11fead95785..05c42291070 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -18,10 +18,16 @@ import pandas as pd import pyarrow +from feast.data_source import FileSource from feast.entity import Entity from feast.feature_view import FeatureView from feast.infra.provider import Provider, get_provider -from feast.offline_store import OfflineStore, get_offline_store, RetrievalJob, get_offline_store_for_retrieval +from feast.offline_store import ( + OfflineStore, + RetrievalJob, + get_offline_store, + get_offline_store_for_retrieval, +) from feast.registry import Registry from feast.repo_config import ( LocalOnlineStoreConfig, @@ -172,13 +178,17 @@ def materialize( for name in feature_views: feature_view = registry.get_feature_view(name, self.project) full_feature_views.append(feature_view) + + # TODO paging large loads + for feature_view in full_feature_views: + if isinstance(feature_view.input, FileSource): + raise NotImplementedError( + "This function is not yet implemented for File data sources" + ) if feature_view.input.table_ref is None: raise NotImplementedError( f"This function is only implemented for FeatureViews with a table_ref; {feature_view.name} does not have one." ) - - # TODO paging large loads - for feature_view in full_feature_views: ( entity_names, feature_names, @@ -206,7 +216,7 @@ def materialize( provider = self._get_provider() provider.online_write_batch( - "default", feature_view, rows_to_write, created_timestamp_column + "default", feature_view, rows_to_write, datetime.utcnow() ) diff --git a/sdk/python/feast/offline_store.py b/sdk/python/feast/offline_store.py index 7effb6e7e96..ab3a5a51161 100644 --- a/sdk/python/feast/offline_store.py +++ b/sdk/python/feast/offline_store.py @@ -15,7 +15,7 @@ from abc import ABC, abstractmethod from dataclasses import asdict, dataclass from datetime import datetime, timedelta -from typing import Callable, Dict, List, Optional, Tuple, Type, Union +from typing import Callable, Dict, List, Optional, Type, Union import pandas as pd import pyarrow @@ -94,7 +94,7 @@ def pull_latest_from_table( created_timestamp_column: Optional[str], start_date: datetime, end_date: datetime, - ) -> Optional[pyarrow.Table]: + ) -> pyarrow.Table: pass @staticmethod @@ -281,7 +281,13 @@ def build_point_in_time_query( class FileOfflineStore(OfflineStore): @staticmethod def pull_latest_from_table( - feature_view: FeatureView, start_date: datetime, end_date: datetime, + table_ref: str, + entity_names: List[str], + feature_names: List[str], + event_timestamp_column: str, + created_timestamp_column: Optional[str], + start_date: datetime, + end_date: datetime, ) -> pyarrow.Table: pass @@ -553,6 +559,7 @@ def _get_requested_feature_views_to_features_dict( ORDER BY event_timestamp """ + def get_offline_store(config: RepoConfig) -> Type[OfflineStore]: if config.provider == "gcp": return BigQueryOfflineStore diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_bigquery_ingestion.py new file mode 100644 index 00000000000..5dccaa05568 --- /dev/null +++ b/sdk/python/tests/test_bigquery_ingestion.py @@ -0,0 +1,76 @@ +import time +from datetime import datetime, timedelta + +import pandas as pd +import pytest +from google.cloud import bigquery + +from feast.data_source import BigQuerySource +from feast.feature import Feature +from feast.feature_store import FeatureStore +from feast.feature_view import FeatureView +from feast.repo_config import LocalOnlineStoreConfig, OnlineStoreConfig, RepoConfig +from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.types.Value_pb2 import Value as ValueProto +from feast.value_type import ValueType + + +@pytest.mark.integration +def test_bigquery_ingestion(): + # create dataset + ts = pd.Timestamp.now(tz="UTC").round("ms") + data = { + "id": [1, 2, 1], + "value": [0.1, 0.2, 0.3], + "ts": [ts - timedelta(minutes=2), ts, ts], + "created_ts": [ts, ts, ts], + } + df = pd.DataFrame.from_dict(data) + + # load dataset into BigQuery + client = bigquery.Client() + job_config = bigquery.LoadJobConfig() + gcp_project = client.project + bigquery_dataset = "test_ingestion" + dataset = bigquery.Dataset(f"{gcp_project}.{bigquery_dataset}") + client.create_dataset(dataset, exists_ok=True) + table_id = f"{gcp_project}.{bigquery_dataset}.table_{int(time.time())}" + job = client.load_table_from_dataframe(df, table_id, job_config=job_config) + job.result() + + # create FeatureView + fv = FeatureView( + "test_fv", + ["id_mapped"], + [Feature("value", ValueType.FLOAT)], + timedelta(minutes=5), + BigQuerySource( + "timestamp_mapped", + table_id, + "created_ts", + {"ts": "timestamp_mapped", "id": "id_mapped"}, + "", + ), + ) + config = RepoConfig( + metadata_store="./metadata.db", + project="default", + provider="gcp", + online_store=OnlineStoreConfig(local=LocalOnlineStoreConfig("online_store.db")), + ) + fs = FeatureStore(config=config) + fs.apply([fv]) + + # run materialize() + fs.materialize( + ["test_fv"], + datetime.utcnow() - timedelta(minutes=5), + datetime.utcnow() - timedelta(minutes=0), + ) + + # check result of materialize() + entity_key = EntityKeyProto( + entity_names=["id_mapped"], entity_values=[ValueProto(int32_val=1)] + ) + _, val = fs._get_provider().online_read("default", fv, entity_key) + assert abs(val["value"].double_val - 0.3) < 1e-6 From 3ffdb19f3bee8feb52d593ed2e99cccde4448445 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Tue, 16 Mar 2021 18:58:09 -0400 Subject: [PATCH 05/12] Update created_ts to use column or null instead of current ts Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 19 +++++++++++++----- sdk/python/feast/infra/gcp.py | 20 ++++++++++++------- sdk/python/feast/infra/local_sqlite.py | 10 ++++++---- sdk/python/feast/infra/provider.py | 12 +++++------ .../tests/cli/online_read_write_test.py | 2 +- 5 files changed, 40 insertions(+), 23 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 05c42291070..f7e1d397943 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -215,9 +215,7 @@ def materialize( rows_to_write = _convert_arrow_to_proto(table, feature_view) provider = self._get_provider() - provider.online_write_batch( - "default", feature_view, rows_to_write, datetime.utcnow() - ) + provider.online_write_batch(self.project, feature_view, rows_to_write) def _get_requested_feature_views( @@ -305,7 +303,7 @@ def _run_forward_field_mapping( def _convert_arrow_to_proto( table: pyarrow.Table, feature_view: FeatureView -) -> List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime]]: +) -> List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]]: rows_to_write = [] for row in zip(*table.to_pydict().values()): entity_key = EntityKeyProto() @@ -322,7 +320,18 @@ def _convert_arrow_to_proto( event_timestamp_idx = table.column_names.index( feature_view.input.event_timestamp_column ) - rows_to_write.append((entity_key, feature_dict, row[event_timestamp_idx])) + event_timestamp = row[event_timestamp_idx] + if feature_view.input.created_timestamp_column is not None: + created_timestamp_idx = table.column_names.index( + feature_view.input.created_timestamp_column + ) + created_timestamp = row[created_timestamp_idx] + else: + created_timestamp = None + + rows_to_write.append( + (entity_key, feature_dict, event_timestamp, created_timestamp) + ) return rows_to_write diff --git a/sdk/python/feast/infra/gcp.py b/sdk/python/feast/infra/gcp.py index d97a89f416a..7604eebed9f 100644 --- a/sdk/python/feast/infra/gcp.py +++ b/sdk/python/feast/infra/gcp.py @@ -38,7 +38,9 @@ def compute_datastore_entity_id(entity_key: EntityKeyProto) -> str: return mmh3.hash_bytes(serialize_entity_key(entity_key)).hex() -def _make_tzaware(t: datetime): +def _make_tzaware(t: Optional[datetime]): + if t is None: + return t """ We assume tz-naive datetimes are UTC """ if t.tzinfo is None: return t.replace(tzinfo=utc) @@ -106,14 +108,15 @@ def online_write_batch( self, project: str, table: Union[FeatureTable, FeatureView], - data: List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime]], - created_ts: datetime, + data: List[ + Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] + ], ) -> None: from google.cloud import datastore client = self._initialize_client() - for entity_key, features, timestamp in data: + for entity_key, features, timestamp, created_ts in data: document_id = compute_datastore_entity_id(entity_key) key = client.key( @@ -125,9 +128,12 @@ def online_write_batch( if entity["event_ts"] > _make_tzaware(timestamp): # Do not overwrite feature values computed from fresher data continue - elif entity["event_ts"] == _make_tzaware(timestamp) and entity[ - "created_ts" - ] > _make_tzaware(created_ts): + elif ( + entity["event_ts"] == _make_tzaware(timestamp) + and created_ts is not None + and entity["created_ts"] is not None + and entity["created_ts"] > _make_tzaware(created_ts) + ): # Do not overwrite feature values computed from the same data, but # computed later than this one continue diff --git a/sdk/python/feast/infra/local_sqlite.py b/sdk/python/feast/infra/local_sqlite.py index 9c35a7b0419..ae038f6e090 100644 --- a/sdk/python/feast/infra/local_sqlite.py +++ b/sdk/python/feast/infra/local_sqlite.py @@ -53,13 +53,14 @@ def online_write_batch( self, project: str, table: Union[FeatureTable, FeatureView], - data: List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime]], - created_ts: datetime, + data: List[ + Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] + ], ) -> None: conn = self._get_conn() with conn: - for entity_key, values, timestamp in data: + for entity_key, values, timestamp, created_ts in data: for feature_name, val in values.items(): entity_key_bin = serialize_entity_key(entity_key) @@ -67,7 +68,7 @@ def online_write_batch( f""" UPDATE {_table_id(project, table)} SET value = ?, event_ts = ?, created_ts = ? - WHERE (event_ts < ? OR (event_ts = ? AND created_ts < ?)) + WHERE (event_ts < ? OR (event_ts = ? AND (created_ts IS NULL OR ? IS NULL OR created_ts < ?))) AND (entity_key = ? AND feature_name = ?) """, ( @@ -79,6 +80,7 @@ def online_write_batch( timestamp, timestamp, created_ts, + created_ts, entity_key_bin, feature_name, ), diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py index 33834a92b16..9afc24ed5f7 100644 --- a/sdk/python/feast/infra/provider.py +++ b/sdk/python/feast/infra/provider.py @@ -44,8 +44,9 @@ def online_write_batch( self, project: str, table: Union[FeatureTable, FeatureView], - data: List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime]], - created_ts: datetime, + data: List[ + Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]] + ], ) -> None: """ Write a batch of feature rows to the online store. This is a low level interface, not @@ -56,10 +57,9 @@ def online_write_batch( Args: project: Feast project name table: Feast FeatureTable - data: a list of triplets containing Feature data. Each triplet contains an Entity Key, - a dict containing feature values, and event timestamp for the row. - created_ts: the created timestamp (typically set to current time), same value used for - all rows. + data: a list of quadruplets containing Feature data. Each quadruplet contains an Entity Key, + a dict containing feature values, an event timestamp for the row, and + the created timestamp for the row if it exists. """ ... diff --git a/sdk/python/tests/cli/online_read_write_test.py b/sdk/python/tests/cli/online_read_write_test.py index 4535db012fe..a80b64485ef 100644 --- a/sdk/python/tests/cli/online_read_write_test.py +++ b/sdk/python/tests/cli/online_read_write_test.py @@ -36,9 +36,9 @@ def _driver_rw_test(event_ts, created_ts, write, expect_read): "lon": ValueProto(string_val=write_lon), }, event_ts, + created_ts, ) ], - created_ts=created_ts, ) _, val = provider.online_read( From 072ac726b9ea84fa469ae9aba00f38ff326a17bc Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Tue, 16 Mar 2021 19:50:17 -0400 Subject: [PATCH 06/12] Update feast types import Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 4 ++-- sdk/python/tests/test_bigquery_ingestion.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index f7e1d397943..12e2ac239a2 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -35,8 +35,8 @@ RepoConfig, load_repo_config, ) -from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto -from feast.types.Value_pb2 import Value as ValueProto +from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.protos.feast.types.Value_pb2 import Value as ValueProto class FeatureStore: diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_bigquery_ingestion.py index 5dccaa05568..19ad645832b 100644 --- a/sdk/python/tests/test_bigquery_ingestion.py +++ b/sdk/python/tests/test_bigquery_ingestion.py @@ -10,8 +10,8 @@ from feast.feature_store import FeatureStore from feast.feature_view import FeatureView from feast.repo_config import LocalOnlineStoreConfig, OnlineStoreConfig, RepoConfig -from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto -from feast.types.Value_pb2 import Value as ValueProto +from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.protos.feast.types.Value_pb2 import Value as ValueProto from feast.value_type import ValueType From 9948c5f2d5fb84c49ae8af8c5c9e50c6cc9f6134 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Tue, 16 Mar 2021 20:04:24 -0400 Subject: [PATCH 07/12] lint Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 4 ++-- sdk/python/tests/test_bigquery_ingestion.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 12e2ac239a2..2a86a9b0e60 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -28,6 +28,8 @@ get_offline_store, get_offline_store_for_retrieval, ) +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 ( LocalOnlineStoreConfig, @@ -35,8 +37,6 @@ RepoConfig, load_repo_config, ) -from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto -from feast.protos.feast.types.Value_pb2 import Value as ValueProto class FeatureStore: diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_bigquery_ingestion.py index 19ad645832b..6f68f5deaf0 100644 --- a/sdk/python/tests/test_bigquery_ingestion.py +++ b/sdk/python/tests/test_bigquery_ingestion.py @@ -9,9 +9,9 @@ from feast.feature import Feature from feast.feature_store import FeatureStore from feast.feature_view import FeatureView -from feast.repo_config import LocalOnlineStoreConfig, OnlineStoreConfig, RepoConfig 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 LocalOnlineStoreConfig, OnlineStoreConfig, RepoConfig from feast.value_type import ValueType From e63ed48f82325b337239c5a4e391ebf60c89df73 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Wed, 17 Mar 2021 19:29:02 -0400 Subject: [PATCH 08/12] Address comments Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 79 +++++++++++---------- sdk/python/feast/infra/gcp.py | 10 +-- sdk/python/feast/offline_store.py | 10 +-- sdk/python/feast/value_type.py | 20 +++++- sdk/python/tests/test_bigquery_ingestion.py | 30 ++++---- 5 files changed, 91 insertions(+), 58 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 2a86a9b0e60..b6bc09b88b7 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -13,7 +13,7 @@ # limitations under the License. from datetime import datetime from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple, Type, Union +from typing import Dict, List, Optional, Tuple, Type, Union import pandas as pd import pyarrow @@ -37,6 +37,7 @@ RepoConfig, load_repo_config, ) +from feast.value_type import convert_to_proto class FeatureStore: @@ -166,18 +167,33 @@ def get_historical_features( ) return job - @property - def project(self) -> str: - return "default" - def materialize( - self, feature_views: List[str], start_date: datetime, end_date: datetime - ): + self, + feature_views: Optional[List[str]], + start_date: datetime, + end_date: datetime, + ) -> None: + """ + Materialize data from the offline store into the online store. + + This method materializes feature data in the specified interval from either + the specified feature views or all feature views if none are specified + into the online store where it is available for online serving. + + Args: + feature_views (List[str]): Optional list of feature view names. If selected, will only run + materialization for the specified feature views. + start_date (datetime): Start date of query + end_date (datetime): End date of query + """ full_feature_views = [] registry = self._get_registry() - for name in feature_views: - feature_view = registry.get_feature_view(name, self.project) - full_feature_views.append(feature_view) + if feature_views is None: + full_feature_views = registry.list_feature_views(self.config.project) + else: + for name in feature_views: + feature_view = registry.get_feature_view(name, self.config.project) + full_feature_views.append(feature_view) # TODO paging large loads for feature_view in full_feature_views: @@ -185,7 +201,7 @@ def materialize( raise NotImplementedError( "This function is not yet implemented for File data sources" ) - if feature_view.input.table_ref is None: + if not feature_view.input.table_ref: raise NotImplementedError( f"This function is only implemented for FeatureViews with a table_ref; {feature_view.name} does not have one." ) @@ -198,7 +214,7 @@ def materialize( offline_store = self._get_offline_store() table = offline_store.pull_latest_from_table( - feature_view.input.table_ref, + feature_view.input, entity_names, feature_names, event_timestamp_column, @@ -215,7 +231,9 @@ def materialize( rows_to_write = _convert_arrow_to_proto(table, feature_view) provider = self._get_provider() - provider.online_write_batch(self.project, feature_view, rows_to_write) + provider.online_write_batch( + self.config.project, feature_view, rows_to_write + ) def _get_requested_feature_views( @@ -246,12 +264,18 @@ def _run_reverse_field_mapping( feature_view: FeatureView, ) -> Tuple[List[str], List[str], str, Optional[str]]: """ - If a field mapping exists, run it in reverse on the entity names, feature names, event timestamp column, and created timestamp column to get the names of the relevant columns in the BigQuery table. + If a field mapping exists, run it in reverse on the entity names, + feature names, event timestamp column, and created timestamp column + to get the names of the relevant columns in the BigQuery table. Args: - feature_view: FeatureView object containing the field mapping as well as the names to reverse-map. + feature_view: FeatureView object containing the field mapping + as well as the names to reverse-map. Returns: - Tuple containing the list of reverse-mapped entity names, reverse-mapped feature names, reverse-mapped event timestamp column, and reverse-mapped created timestamp column that will be passed into the query to the offline store. + Tuple containing the list of reverse-mapped entity names, + reverse-mapped feature names, reverse-mapped event timestamp column, + and reverse-mapped created timestamp column that will be passed into + the query to the offline store. """ # if we have mapped fields, use the original field names in the call to the offline store event_timestamp_column = feature_view.input.event_timestamp_column @@ -269,7 +293,7 @@ def _run_reverse_field_mapping( ) created_timestamp_column = ( reverse_field_mapping[created_timestamp_column] - if created_timestamp_column is not None + if created_timestamp_column and created_timestamp_column in reverse_field_mapping.keys() else created_timestamp_column ) @@ -310,12 +334,12 @@ def _convert_arrow_to_proto( for entity_name in feature_view.entities: entity_key.entity_names.append(entity_name) idx = table.column_names.index(entity_name) - value = _convert_to_proto(row[idx]) + value = convert_to_proto(row[idx]) entity_key.entity_values.append(value) feature_dict = {} for feature in feature_view.features: idx = table.column_names.index(feature.name) - value = _convert_to_proto(row[idx]) + value = convert_to_proto(row[idx]) feature_dict[feature.name] = value event_timestamp_idx = table.column_names.index( feature_view.input.event_timestamp_column @@ -333,20 +357,3 @@ def _convert_arrow_to_proto( (entity_key, feature_dict, event_timestamp, created_timestamp) ) return rows_to_write - - -def _convert_to_proto(value: Any) -> ValueProto: - value_proto = ValueProto() - if isinstance(value, str): - value_proto.string_val = value - elif isinstance(value, bool): - value_proto.bool_val = value - elif isinstance(value, int): - value_proto.int32_val = value - elif isinstance(value, float): - value_proto.double_val = value - elif isinstance(value, bytes): - value_proto.bytes_val = value - else: - raise ValueError(f"Cannot convert value {value} of type {type(value)}.") - return value_proto diff --git a/sdk/python/feast/infra/gcp.py b/sdk/python/feast/infra/gcp.py index 7604eebed9f..b99ea9a4154 100644 --- a/sdk/python/feast/infra/gcp.py +++ b/sdk/python/feast/infra/gcp.py @@ -38,9 +38,7 @@ def compute_datastore_entity_id(entity_key: EntityKeyProto) -> str: return mmh3.hash_bytes(serialize_entity_key(entity_key)).hex() -def _make_tzaware(t: Optional[datetime]): - if t is None: - return t +def _make_tzaware(t: datetime): """ We assume tz-naive datetimes are UTC """ if t.tzinfo is None: return t.replace(tzinfo=utc) @@ -145,7 +143,11 @@ def online_write_batch( key=entity_key.SerializeToString(), values={k: v.SerializeToString() for k, v in features.items()}, event_ts=_make_tzaware(timestamp), - created_ts=_make_tzaware(created_ts), + created_ts=( + _make_tzaware(created_ts) + if created_ts is not None + else None + ), ) ) client.put(entity) diff --git a/sdk/python/feast/offline_store.py b/sdk/python/feast/offline_store.py index ab3a5a51161..fa698cf1354 100644 --- a/sdk/python/feast/offline_store.py +++ b/sdk/python/feast/offline_store.py @@ -22,7 +22,7 @@ from google.cloud import bigquery from jinja2 import BaseLoader, Environment -from feast.data_source import BigQuerySource, FileSource +from feast.data_source import BigQuerySource, DataSource, FileSource from feast.feature_view import FeatureView from feast.repo_config import RepoConfig @@ -87,7 +87,7 @@ class OfflineStore(ABC): @staticmethod @abstractmethod def pull_latest_from_table( - table_ref: str, + data_source: DataSource, entity_names: List[str], feature_names: List[str], event_timestamp_column: str, @@ -111,7 +111,7 @@ def get_historical_features( class BigQueryOfflineStore(OfflineStore): @staticmethod def pull_latest_from_table( - table_ref: str, + data_source: DataSource, entity_names: List[str], feature_names: List[str], event_timestamp_column: str, @@ -119,6 +119,8 @@ def pull_latest_from_table( start_date: datetime, end_date: datetime, ) -> pyarrow.Table: + assert isinstance(data_source, BigQuerySource) + table_ref = data_source.table_ref if table_ref is None: raise ValueError( "This function can only be called on a FeatureView with a table_ref" @@ -281,7 +283,7 @@ def build_point_in_time_query( class FileOfflineStore(OfflineStore): @staticmethod def pull_latest_from_table( - table_ref: str, + data_source: DataSource, entity_names: List[str], feature_names: List[str], event_timestamp_column: str, diff --git a/sdk/python/feast/value_type.py b/sdk/python/feast/value_type.py index eba16015d35..f6d493ac7f5 100644 --- a/sdk/python/feast/value_type.py +++ b/sdk/python/feast/value_type.py @@ -11,9 +11,10 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. - import enum +from typing import Any +from feast.protos.feast.types.Value_pb2 import Value as ValueProto from tensorflow_metadata.proto.v0 import schema_pb2 @@ -58,3 +59,20 @@ def to_tfx_schema_feature_type(self): return schema_pb2.FeatureType.FLOAT else: return schema_pb2.FeatureType.TYPE_UNKNOWN + + +def convert_to_proto(value: Any) -> ValueProto: + value_proto = ValueProto() + if isinstance(value, str): + value_proto.string_val = value + elif isinstance(value, bool): + value_proto.bool_val = value + elif isinstance(value, int): + value_proto.int32_val = value + elif isinstance(value, float): + value_proto.double_val = value + elif isinstance(value, bytes): + value_proto.bytes_val = value + else: + raise ValueError(f"Cannot convert value {value} of type {type(value)}.") + return value_proto diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_bigquery_ingestion.py index 6f68f5deaf0..f5338b8b0cc 100644 --- a/sdk/python/tests/test_bigquery_ingestion.py +++ b/sdk/python/tests/test_bigquery_ingestion.py @@ -16,13 +16,13 @@ @pytest.mark.integration -def test_bigquery_ingestion(): +def test_bigquery_ingestion_correctness(): # create dataset ts = pd.Timestamp.now(tz="UTC").round("ms") data = { "id": [1, 2, 1], "value": [0.1, 0.2, 0.3], - "ts": [ts - timedelta(minutes=2), ts, ts], + "ts_1": [ts - timedelta(minutes=2), ts, ts], "created_ts": [ts, ts, ts], } df = pd.DataFrame.from_dict(data) @@ -34,22 +34,26 @@ def test_bigquery_ingestion(): bigquery_dataset = "test_ingestion" dataset = bigquery.Dataset(f"{gcp_project}.{bigquery_dataset}") client.create_dataset(dataset, exists_ok=True) + dataset.default_table_expiration_ms = ( + 1000 * 60 * 60 * 24 * 14 + ) # 2 weeks in milliseconds + client.update_dataset(dataset, ["default_table_expiration_ms"]) table_id = f"{gcp_project}.{bigquery_dataset}.table_{int(time.time())}" job = client.load_table_from_dataframe(df, table_id, job_config=job_config) job.result() # create FeatureView fv = FeatureView( - "test_fv", - ["id_mapped"], - [Feature("value", ValueType.FLOAT)], - timedelta(minutes=5), - BigQuerySource( - "timestamp_mapped", - table_id, - "created_ts", - {"ts": "timestamp_mapped", "id": "id_mapped"}, - "", + name="test_fv", + entities=["driver_id"], + features=[Feature("value", ValueType.FLOAT)], + ttl=timedelta(minutes=5), + input=BigQuerySource( + event_timestamp_column="ts", + table_ref=table_id, + created_timestamp_column="created_ts", + field_mapping={"ts_1": "ts", "id": "driver_id"}, + date_partition_column="", ), ) config = RepoConfig( @@ -70,7 +74,7 @@ def test_bigquery_ingestion(): # check result of materialize() entity_key = EntityKeyProto( - entity_names=["id_mapped"], entity_values=[ValueProto(int32_val=1)] + entity_names=["driver_id"], entity_values=[ValueProto(int32_val=1)] ) _, val = fs._get_provider().online_read("default", fv, entity_key) assert abs(val["value"].double_val - 0.3) < 1e-6 From 14ba58d8dd231a26347743ca13a133eb8df502cb Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Wed, 17 Mar 2021 20:17:16 -0400 Subject: [PATCH 09/12] Use existing type map function Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 6 +++--- sdk/python/feast/type_map.py | 5 +++++ sdk/python/feast/value_type.py | 19 ------------------- 3 files changed, 8 insertions(+), 22 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index b6bc09b88b7..93d5e61963a 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -37,7 +37,7 @@ RepoConfig, load_repo_config, ) -from feast.value_type import convert_to_proto +from feast.type_map import python_value_to_proto_value class FeatureStore: @@ -334,12 +334,12 @@ def _convert_arrow_to_proto( for entity_name in feature_view.entities: entity_key.entity_names.append(entity_name) idx = table.column_names.index(entity_name) - value = convert_to_proto(row[idx]) + value = python_value_to_proto_value(row[idx]) entity_key.entity_values.append(value) feature_dict = {} for feature in feature_view.features: idx = table.column_names.index(feature.name) - value = convert_to_proto(row[idx]) + value = python_value_to_proto_value(row[idx]) feature_dict[feature.name] = value event_timestamp_idx = table.column_names.index( feature_view.input.event_timestamp_column diff --git a/sdk/python/feast/type_map.py b/sdk/python/feast/type_map.py index 58811afe1ca..5afb5f2e408 100644 --- a/sdk/python/feast/type_map.py +++ b/sdk/python/feast/type_map.py @@ -313,6 +313,11 @@ def _python_value_to_proto_value(feast_value_type, value) -> ProtoValue: raise Exception(f"Unsupported data type: ${str(type(value))}") +def python_value_to_proto_value(value: Any) -> ProtoValue: + value_type = python_type_to_feast_value_type("", value) + return _python_value_to_proto_value(value_type, value) + + def _proto_str_to_value_type(proto_str: str) -> ValueType: """ Returns Feast ValueType given Feast ValueType string. diff --git a/sdk/python/feast/value_type.py b/sdk/python/feast/value_type.py index f6d493ac7f5..317001885b5 100644 --- a/sdk/python/feast/value_type.py +++ b/sdk/python/feast/value_type.py @@ -12,9 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import enum -from typing import Any -from feast.protos.feast.types.Value_pb2 import Value as ValueProto from tensorflow_metadata.proto.v0 import schema_pb2 @@ -59,20 +57,3 @@ def to_tfx_schema_feature_type(self): return schema_pb2.FeatureType.FLOAT else: return schema_pb2.FeatureType.TYPE_UNKNOWN - - -def convert_to_proto(value: Any) -> ValueProto: - value_proto = ValueProto() - if isinstance(value, str): - value_proto.string_val = value - elif isinstance(value, bool): - value_proto.bool_val = value - elif isinstance(value, int): - value_proto.int32_val = value - elif isinstance(value, float): - value_proto.double_val = value - elif isinstance(value, bytes): - value_proto.bytes_val = value - else: - raise ValueError(f"Cannot convert value {value} of type {type(value)}.") - return value_proto From c8ddd762b02626b95142782b0de5bb79ed2fbc30 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Fri, 19 Mar 2021 18:47:02 -0400 Subject: [PATCH 10/12] Address comments Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 14 ++- sdk/python/tests/test_bigquery_ingestion.py | 127 +++++++++++--------- 2 files changed, 77 insertions(+), 64 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 93d5e61963a..e1d4916375d 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -176,8 +176,8 @@ def materialize( """ Materialize data from the offline store into the online store. - This method materializes feature data in the specified interval from either - the specified feature views or all feature views if none are specified + This method loads feature data in the specified interval from either + the specified feature views, or all feature views if none are specified, into the online store where it is available for online serving. Args: @@ -186,17 +186,19 @@ def materialize( start_date (datetime): Start date of query end_date (datetime): End date of query """ - full_feature_views = [] + feature_views_to_materialize = [] registry = self._get_registry() if feature_views is None: - full_feature_views = registry.list_feature_views(self.config.project) + feature_views_to_materialize = registry.list_feature_views( + self.config.project + ) else: for name in feature_views: feature_view = registry.get_feature_view(name, self.config.project) - full_feature_views.append(feature_view) + feature_views_to_materialize.append(feature_view) # TODO paging large loads - for feature_view in full_feature_views: + for feature_view in feature_views_to_materialize: if isinstance(feature_view.input, FileSource): raise NotImplementedError( "This function is not yet implemented for File data sources" diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_bigquery_ingestion.py index f5338b8b0cc..47f8e70ed58 100644 --- a/sdk/python/tests/test_bigquery_ingestion.py +++ b/sdk/python/tests/test_bigquery_ingestion.py @@ -1,3 +1,4 @@ +import random import time from datetime import datetime, timedelta @@ -16,65 +17,75 @@ @pytest.mark.integration -def test_bigquery_ingestion_correctness(): - # create dataset - ts = pd.Timestamp.now(tz="UTC").round("ms") - data = { - "id": [1, 2, 1], - "value": [0.1, 0.2, 0.3], - "ts_1": [ts - timedelta(minutes=2), ts, ts], - "created_ts": [ts, ts, ts], - } - df = pd.DataFrame.from_dict(data) +class TestBigQueryIngestion: + def setup_method(self): + self.client = bigquery.Client() + self.gcp_project = self.client.project + self.bigquery_dataset = "test_ingestion" + dataset = bigquery.Dataset(f"{self.gcp_project}.{self.bigquery_dataset}") + self.client.create_dataset(dataset, exists_ok=True) + dataset.default_table_expiration_ms = ( + 1000 * 60 * 60 * 24 * 14 + ) # 2 weeks in milliseconds + self.client.update_dataset(dataset, ["default_table_expiration_ms"]) - # load dataset into BigQuery - client = bigquery.Client() - job_config = bigquery.LoadJobConfig() - gcp_project = client.project - bigquery_dataset = "test_ingestion" - dataset = bigquery.Dataset(f"{gcp_project}.{bigquery_dataset}") - client.create_dataset(dataset, exists_ok=True) - dataset.default_table_expiration_ms = ( - 1000 * 60 * 60 * 24 * 14 - ) # 2 weeks in milliseconds - client.update_dataset(dataset, ["default_table_expiration_ms"]) - table_id = f"{gcp_project}.{bigquery_dataset}.table_{int(time.time())}" - job = client.load_table_from_dataframe(df, table_id, job_config=job_config) - job.result() + def test_bigquery_ingestion_correctness(self): + # create dataset + ts = pd.Timestamp.now(tz="UTC").round("ms") + checked_value = ( + random.random() + ) # random value so test doesn't still work if no values written to online store + data = { + "id": [1, 2, 1], + "value": [0.1, 0.2, checked_value], + "ts_1": [ts - timedelta(minutes=2), ts, ts], + "created_ts": [ts, ts, ts], + } + df = pd.DataFrame.from_dict(data) - # create FeatureView - fv = FeatureView( - name="test_fv", - entities=["driver_id"], - features=[Feature("value", ValueType.FLOAT)], - ttl=timedelta(minutes=5), - input=BigQuerySource( - event_timestamp_column="ts", - table_ref=table_id, - created_timestamp_column="created_ts", - field_mapping={"ts_1": "ts", "id": "driver_id"}, - date_partition_column="", - ), - ) - config = RepoConfig( - metadata_store="./metadata.db", - project="default", - provider="gcp", - online_store=OnlineStoreConfig(local=LocalOnlineStoreConfig("online_store.db")), - ) - fs = FeatureStore(config=config) - fs.apply([fv]) + # load dataset into BigQuery + job_config = bigquery.LoadJobConfig() + table_id = ( + f"{self.gcp_project}.{self.bigquery_dataset}.correctness_{int(time.time())}" + ) + job = self.client.load_table_from_dataframe(df, table_id, job_config=job_config) + job.result() - # run materialize() - fs.materialize( - ["test_fv"], - datetime.utcnow() - timedelta(minutes=5), - datetime.utcnow() - timedelta(minutes=0), - ) + # create FeatureView + fv = FeatureView( + name="test_bq_correctness", + entities=["driver_id"], + features=[Feature("value", ValueType.FLOAT)], + ttl=timedelta(minutes=5), + input=BigQuerySource( + event_timestamp_column="ts", + table_ref=table_id, + created_timestamp_column="created_ts", + field_mapping={"ts_1": "ts", "id": "driver_id"}, + date_partition_column="", + ), + ) + config = RepoConfig( + metadata_store="./metadata.db", + project="default", + provider="gcp", + online_store=OnlineStoreConfig( + local=LocalOnlineStoreConfig("online_store.db") + ), + ) + fs = FeatureStore(config=config) + fs.apply([fv]) - # check result of materialize() - entity_key = EntityKeyProto( - entity_names=["driver_id"], entity_values=[ValueProto(int32_val=1)] - ) - _, val = fs._get_provider().online_read("default", fv, entity_key) - assert abs(val["value"].double_val - 0.3) < 1e-6 + # run materialize() + fs.materialize( + ["test_bq_correctness"], + datetime.utcnow() - timedelta(minutes=5), + datetime.utcnow() - timedelta(minutes=0), + ) + + # check result of materialize() + entity_key = EntityKeyProto( + entity_names=["driver_id"], entity_values=[ValueProto(int64_val=1)] + ) + t, val = fs._get_provider().online_read("default", fv, entity_key) + assert abs(val["value"].double_val - checked_value) < 1e-6 From 5d063409536656207c54596aeaaa076227156e2d Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Mon, 22 Mar 2021 11:31:17 -0400 Subject: [PATCH 11/12] Address comments Signed-off-by: Jacob Klegar --- sdk/python/feast/feature_store.py | 24 ++++++++++++++------- sdk/python/tests/test_bigquery_ingestion.py | 2 +- 2 files changed, 17 insertions(+), 9 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index e1d4916375d..3ee63d42092 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -13,7 +13,7 @@ # limitations under the License. from datetime import datetime from pathlib import Path -from typing import Dict, List, Optional, Tuple, Type, Union +from typing import Dict, List, Optional, Tuple, Union import pandas as pd import pyarrow @@ -23,7 +23,6 @@ from feast.feature_view import FeatureView from feast.infra.provider import Provider, get_provider from feast.offline_store import ( - OfflineStore, RetrievalJob, get_offline_store, get_offline_store_for_retrieval, @@ -69,9 +68,6 @@ def __init__( def _get_provider(self) -> Provider: return get_provider(self.config) - def _get_offline_store(self) -> Type[OfflineStore]: - return get_offline_store(self.config) - def _get_registry(self) -> Registry: return Registry(self.config.metadata_store) @@ -183,8 +179,20 @@ def materialize( Args: feature_views (List[str]): Optional list of feature view names. If selected, will only run materialization for the specified feature views. - start_date (datetime): Start date of query - end_date (datetime): End date of query + start_date (datetime): Start date for time range of data to materialize into the online store + end_date (datetime): End date for time range of data to materialize into the online store + + Examples: + Materialize all features into the online store over the interval + from 3 hours ago to 10 minutes ago. + >>> from datetime import datetime, timedelta + >>> from feast.feature_store import FeatureStore + >>> + >>> fs = FeatureStore(config=RepoConfig(provider="gcp")) + >>> fs.materialize( + >>> start_date=datetime.utcnow() - timedelta(hours=3), + >>> end_date=datetime.utcnow() - timedelta(minutes=10) + >>> ) """ feature_views_to_materialize = [] registry = self._get_registry() @@ -214,7 +222,7 @@ def materialize( created_timestamp_column, ) = _run_reverse_field_mapping(feature_view) - offline_store = self._get_offline_store() + offline_store = get_offline_store(self.config) table = offline_store.pull_latest_from_table( feature_view.input, entity_names, diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_bigquery_ingestion.py index 47f8e70ed58..db6637bba1c 100644 --- a/sdk/python/tests/test_bigquery_ingestion.py +++ b/sdk/python/tests/test_bigquery_ingestion.py @@ -17,7 +17,7 @@ @pytest.mark.integration -class TestBigQueryIngestion: +class TestMaterializeFromBigQueryToDatastore: def setup_method(self): self.client = bigquery.Client() self.gcp_project = self.client.project From d43cb086ee7e87a3e16b94b19e46d4deb5f65841 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Mon, 22 Mar 2021 17:33:22 -0400 Subject: [PATCH 12/12] Add comment Signed-off-by: Jacob Klegar --- sdk/python/feast/offline_store.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/sdk/python/feast/offline_store.py b/sdk/python/feast/offline_store.py index fa698cf1354..3b1f9a9a50e 100644 --- a/sdk/python/feast/offline_store.py +++ b/sdk/python/feast/offline_store.py @@ -95,6 +95,11 @@ def pull_latest_from_table( start_date: datetime, end_date: datetime, ) -> pyarrow.Table: + """ + Note that entity_names, feature_names, event_timestamp_column, and created_timestamp_column + have all already been mapped back to column names of the source table + and those column names are the values passed into this function. + """ pass @staticmethod