From 00a55e5726e7e6d56cf20cc0779d5676803b1303 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Mon, 22 Mar 2021 10:51:25 -0400 Subject: [PATCH 1/5] WIP pull query from BQ Signed-off-by: Jacob Klegar --- sdk/python/feast/data_source.py | 4 +++- sdk/python/feast/feature_store.py | 4 ---- sdk/python/feast/offline_store.py | 11 +++++------ 3 files changed, 8 insertions(+), 11 deletions(-) diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 2a2f67d301a..9479630829f 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -172,6 +172,7 @@ def to_proto(self) -> DataSourceProto.BigQueryOptions: bigquery_options_proto = DataSourceProto.BigQueryOptions( table_ref=self.table_ref, + query=self.query, ) return bigquery_options_proto @@ -461,13 +462,14 @@ def from_proto(data_source): created_timestamp_column=data_source.created_timestamp_column, date_partition_column=data_source.date_partition_column, ) - elif data_source.bigquery_options.table_ref: + elif (data_source.bigquery_options.table_ref or data_source.bigquery_options.query): data_source_obj = BigQuerySource( field_mapping=data_source.field_mapping, table_ref=data_source.bigquery_options.table_ref, event_timestamp_column=data_source.event_timestamp_column, created_timestamp_column=data_source.created_timestamp_column, date_partition_column=data_source.date_partition_column, + query=data_source.bigquery_options.query, ) elif ( data_source.kafka_options.bootstrap_servers diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index e27a7860831..73f0ef5e660 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -217,10 +217,6 @@ def materialize( raise NotImplementedError( "This function is not yet implemented for File data sources" ) - 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." - ) ( entity_names, feature_names, diff --git a/sdk/python/feast/offline_store.py b/sdk/python/feast/offline_store.py index 3b1f9a9a50e..ed9dda1cd96 100644 --- a/sdk/python/feast/offline_store.py +++ b/sdk/python/feast/offline_store.py @@ -125,11 +125,10 @@ def pull_latest_from_table( 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" - ) + if data_source.table_ref: + from_table = f"`{data_source.table_ref}`" + else: + from_table = f"({data_source.query})" partition_by_entity_string = ", ".join(entity_names) if partition_by_entity_string != "": @@ -145,7 +144,7 @@ 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 `{table_ref}` + FROM {from_table} WHERE {event_timestamp_column} BETWEEN TIMESTAMP('{start_date}') AND TIMESTAMP('{end_date}') ) WHERE _feast_row = 1 From de8fdb91e0f7a8810772a0257fced45a34d22f2c Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Mon, 22 Mar 2021 12:18:55 -0400 Subject: [PATCH 2/5] Add test Signed-off-by: Jacob Klegar --- sdk/python/feast/data_source.py | 7 ++- sdk/python/feast/feature_store.py | 2 +- sdk/python/feast/offline_store.py | 12 ++-- sdk/python/tests/test_bigquery_ingestion.py | 70 +++++++++++++++++++-- 4 files changed, 75 insertions(+), 16 deletions(-) diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 9479630829f..227cfcab12d 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -171,8 +171,7 @@ def to_proto(self) -> DataSourceProto.BigQueryOptions: """ bigquery_options_proto = DataSourceProto.BigQueryOptions( - table_ref=self.table_ref, - query=self.query, + table_ref=self.table_ref, query=self.query, ) return bigquery_options_proto @@ -462,7 +461,9 @@ def from_proto(data_source): created_timestamp_column=data_source.created_timestamp_column, date_partition_column=data_source.date_partition_column, ) - elif (data_source.bigquery_options.table_ref or data_source.bigquery_options.query): + elif ( + data_source.bigquery_options.table_ref or data_source.bigquery_options.query + ): data_source_obj = BigQuerySource( field_mapping=data_source.field_mapping, table_ref=data_source.bigquery_options.table_ref, diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 73f0ef5e660..7dc45ac3239 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -225,7 +225,7 @@ def materialize( ) = _run_reverse_field_mapping(feature_view) offline_store = get_offline_store(self.config) - table = offline_store.pull_latest_from_table( + table = offline_store.pull_latest_from_table_or_query( feature_view.input, entity_names, feature_names, diff --git a/sdk/python/feast/offline_store.py b/sdk/python/feast/offline_store.py index ed9dda1cd96..08c8cb38082 100644 --- a/sdk/python/feast/offline_store.py +++ b/sdk/python/feast/offline_store.py @@ -86,7 +86,7 @@ class OfflineStore(ABC): @staticmethod @abstractmethod - def pull_latest_from_table( + def pull_latest_from_table_or_query( data_source: DataSource, entity_names: List[str], feature_names: List[str], @@ -115,7 +115,7 @@ def get_historical_features( class BigQueryOfflineStore(OfflineStore): @staticmethod - def pull_latest_from_table( + def pull_latest_from_table_or_query( data_source: DataSource, entity_names: List[str], feature_names: List[str], @@ -126,9 +126,9 @@ def pull_latest_from_table( ) -> pyarrow.Table: assert isinstance(data_source, BigQuerySource) if data_source.table_ref: - from_table = f"`{data_source.table_ref}`" + from_expression = f"`{data_source.table_ref}`" else: - from_table = f"({data_source.query})" + from_expression = f"({data_source.query})" partition_by_entity_string = ", ".join(entity_names) if partition_by_entity_string != "": @@ -144,7 +144,7 @@ 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 {from_table} + FROM {from_expression} WHERE {event_timestamp_column} BETWEEN TIMESTAMP('{start_date}') AND TIMESTAMP('{end_date}') ) WHERE _feast_row = 1 @@ -286,7 +286,7 @@ def build_point_in_time_query( class FileOfflineStore(OfflineStore): @staticmethod - def pull_latest_from_table( + def pull_latest_from_table_or_query( data_source: DataSource, entity_names: List[str], feature_names: List[str], diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_bigquery_ingestion.py index 8f5199ecf55..a7ef3cb43b8 100644 --- a/sdk/python/tests/test_bigquery_ingestion.py +++ b/sdk/python/tests/test_bigquery_ingestion.py @@ -29,7 +29,7 @@ def setup_method(self): ) # 2 weeks in milliseconds self.client.update_dataset(dataset, ["default_table_expiration_ms"]) - def test_bigquery_ingestion_correctness(self): + def test_bigquery_table_to_datastore_correctness(self): # create dataset ts = pd.Timestamp.now(tz="UTC").round("ms") checked_value = ( @@ -45,15 +45,13 @@ def test_bigquery_ingestion_correctness(self): # load dataset into BigQuery job_config = bigquery.LoadJobConfig() - table_id = ( - f"{self.gcp_project}.{self.bigquery_dataset}.correctness_{int(time.time())}" - ) + table_id = f"{self.gcp_project}.{self.bigquery_dataset}.table_correctness_{int(time.time())}" job = self.client.load_table_from_dataframe(df, table_id, job_config=job_config) job.result() # create FeatureView fv = FeatureView( - name="test_bq_correctness", + name="test_bq_table_correctness", entities=["driver_id"], features=[Feature("value", ValueType.FLOAT)], ttl=timedelta(minutes=5), @@ -78,7 +76,67 @@ def test_bigquery_ingestion_correctness(self): # run materialize() fs.materialize( - ["test_bq_correctness"], + [fv.name], + 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 + + def test_bigquery_query_to_datastore_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) + + # load dataset into BigQuery + job_config = bigquery.LoadJobConfig() + table_id = f"{self.gcp_project}.{self.bigquery_dataset}.query_correctness_{int(time.time())}" + query = f"SELECT * FROM `{table_id}`" + job = self.client.load_table_from_dataframe(df, table_id, job_config=job_config) + job.result() + + # create FeatureView + fv = FeatureView( + name="test_bq_query_correctness", + entities=["driver_id"], + features=[Feature("value", ValueType.FLOAT)], + ttl=timedelta(minutes=5), + input=BigQuerySource( + event_timestamp_column="ts", + created_timestamp_column="created_ts", + field_mapping={"ts_1": "ts", "id": "driver_id"}, + date_partition_column="", + query=query, + ), + ) + 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( + [fv.name], datetime.utcnow() - timedelta(minutes=5), datetime.utcnow() - timedelta(minutes=0), ) From 796805c8fd0b982499307aabfb9a9d21b3312241 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Tue, 23 Mar 2021 18:31:52 -0400 Subject: [PATCH 3/5] Use read API instead of internal method Signed-off-by: Jacob Klegar --- ...materialize_from_bigquery_to_datastore.py} | 22 +++++++------------ 1 file changed, 8 insertions(+), 14 deletions(-) rename sdk/python/tests/{test_bigquery_ingestion.py => test_materialize_from_bigquery_to_datastore.py} (86%) diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py similarity index 86% rename from sdk/python/tests/test_bigquery_ingestion.py rename to sdk/python/tests/test_materialize_from_bigquery_to_datastore.py index a7ef3cb43b8..35086a155ca 100644 --- a/sdk/python/tests/test_bigquery_ingestion.py +++ b/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py @@ -10,8 +10,6 @@ from feast.feature import Feature from feast.feature_store import FeatureStore from feast.feature_view import FeatureView -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 @@ -82,11 +80,10 @@ def test_bigquery_table_to_datastore_correctness(self): ) # 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 + response_dict = fs.get_online_features( + [f"{fv.name}:value"], [{"driver_id": 1}] + ).to_dict() + assert abs(response_dict[f"{fv.name}:value"][0] - checked_value) < 1e-6 def test_bigquery_query_to_datastore_correctness(self): # create dataset @@ -142,10 +139,7 @@ def test_bigquery_query_to_datastore_correctness(self): ) # check result of materialize() - entity_key = EntityKeyProto( - entity_names=["driver_id"], entity_values=[ValueProto(int64_val=1)] - ) - read_rows = fs._get_provider().online_read("default", fv, [entity_key]) - assert len(read_rows) == 1 - _, val = read_rows[0] - assert abs(val["value"].double_val - checked_value) < 1e-6 + response_dict = fs.get_online_features( + [f"{fv.name}:value"], [{"driver_id": 1}] + ).to_dict() + assert abs(response_dict[f"{fv.name}:value"][0] - checked_value) < 1e-6 From 968766c0e12e93fb11eaef1b799ca578ec3d10e6 Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Wed, 24 Mar 2021 13:08:18 -0400 Subject: [PATCH 4/5] Use separate project per test instance Signed-off-by: Jacob Klegar --- ..._materialize_from_bigquery_to_datastore.py | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-) diff --git a/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py b/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py index 35086a155ca..ce9d134a40f 100644 --- a/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py +++ b/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py @@ -1,4 +1,3 @@ -import random import time from datetime import datetime, timedelta @@ -30,12 +29,9 @@ def setup_method(self): def test_bigquery_table_to_datastore_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], + "value": [0.1, 0.2, 0.3], "ts_1": [ts - timedelta(minutes=2), ts, ts], "created_ts": [ts, ts, ts], } @@ -63,7 +59,7 @@ def test_bigquery_table_to_datastore_correctness(self): ) config = RepoConfig( metadata_store="./metadata.db", - project="default", + project=f"test_bq_table_correctness_{int(time.time())}", provider="gcp", online_store=OnlineStoreConfig( local=LocalOnlineStoreConfig("online_store.db") @@ -83,17 +79,14 @@ def test_bigquery_table_to_datastore_correctness(self): response_dict = fs.get_online_features( [f"{fv.name}:value"], [{"driver_id": 1}] ).to_dict() - assert abs(response_dict[f"{fv.name}:value"][0] - checked_value) < 1e-6 + assert abs(response_dict[f"{fv.name}:value"][0] - 0.3) < 1e-6 def test_bigquery_query_to_datastore_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], + "value": [0.1, 0.2, 0.3], "ts_1": [ts - timedelta(minutes=2), ts, ts], "created_ts": [ts, ts, ts], } @@ -122,7 +115,7 @@ def test_bigquery_query_to_datastore_correctness(self): ) config = RepoConfig( metadata_store="./metadata.db", - project="default", + project=f"test_bq_query_correctness_{int(time.time())}", provider="gcp", online_store=OnlineStoreConfig( local=LocalOnlineStoreConfig("online_store.db") @@ -142,4 +135,4 @@ def test_bigquery_query_to_datastore_correctness(self): response_dict = fs.get_online_features( [f"{fv.name}:value"], [{"driver_id": 1}] ).to_dict() - assert abs(response_dict[f"{fv.name}:value"][0] - checked_value) < 1e-6 + assert abs(response_dict[f"{fv.name}:value"][0] - 0.3) < 1e-6 From 5fe704bc183add4010d85975d52a08cafc5fc8cc Mon Sep 17 00:00:00 2001 From: Jacob Klegar Date: Wed, 24 Mar 2021 13:22:38 -0400 Subject: [PATCH 5/5] Remove extraneous online store config Signed-off-by: Jacob Klegar --- .../tests/test_materialize_from_bigquery_to_datastore.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py b/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py index ce9d134a40f..3a6788130bc 100644 --- a/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py +++ b/sdk/python/tests/test_materialize_from_bigquery_to_datastore.py @@ -9,7 +9,7 @@ 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.repo_config import RepoConfig from feast.value_type import ValueType @@ -61,9 +61,6 @@ def test_bigquery_table_to_datastore_correctness(self): metadata_store="./metadata.db", project=f"test_bq_table_correctness_{int(time.time())}", provider="gcp", - online_store=OnlineStoreConfig( - local=LocalOnlineStoreConfig("online_store.db") - ), ) fs = FeatureStore(config=config) fs.apply([fv]) @@ -117,9 +114,6 @@ def test_bigquery_query_to_datastore_correctness(self): metadata_store="./metadata.db", project=f"test_bq_query_correctness_{int(time.time())}", provider="gcp", - online_store=OnlineStoreConfig( - local=LocalOnlineStoreConfig("online_store.db") - ), ) fs = FeatureStore(config=config) fs.apply([fv])