diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index ea3a4d29e11..3ee63d42092 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -11,15 +11,24 @@ # 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 Dict, List, Optional, Tuple, Union 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 RetrievalJob, get_offline_store_for_retrieval +from feast.offline_store import ( + RetrievalJob, + 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, @@ -27,6 +36,7 @@ RepoConfig, load_repo_config, ) +from feast.type_map import python_value_to_proto_value class FeatureStore: @@ -153,6 +163,88 @@ def get_historical_features( ) return job + def materialize( + 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 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: + 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 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() + if feature_views is None: + 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) + feature_views_to_materialize.append(feature_view) + + # TODO paging large loads + 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" + ) + 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, + event_timestamp_column, + created_timestamp_column, + ) = _run_reverse_field_mapping(feature_view) + + offline_store = get_offline_store(self.config) + table = offline_store.pull_latest_from_table( + feature_view.input, + 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( + self.config.project, feature_view, rows_to_write + ) + def _get_requested_feature_views( feature_refs: List[str], all_feature_views: List[FeatureView] @@ -176,3 +268,102 @@ def _get_requested_feature_views( 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]]: + """ + 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 + 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, Optional[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 = 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 = 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 + ) + 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..b99ea9a4154 100644 --- a/sdk/python/feast/infra/gcp.py +++ b/sdk/python/feast/infra/gcp.py @@ -106,14 +106,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 +126,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 @@ -139,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/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/feast/offline_store.py b/sdk/python/feast/offline_store.py index dbd1503ded3..3b1f9a9a50e 100644 --- a/sdk/python/feast/offline_store.py +++ b/sdk/python/feast/offline_store.py @@ -15,14 +15,14 @@ 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, Type, Union import pandas as pd import pyarrow 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,8 +87,19 @@ class OfflineStore(ABC): @staticmethod @abstractmethod def pull_latest_from_table( - feature_view: FeatureView, start_date: datetime, end_date: datetime, - ) -> Optional[pyarrow.Table]: + data_source: DataSource, + 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: + """ + 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 @@ -105,21 +116,21 @@ def get_historical_features( class BigQueryOfflineStore(OfflineStore): @staticmethod def pull_latest_from_table( - feature_view: FeatureView, start_date: datetime, end_date: datetime, + data_source: DataSource, + 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: + 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" ) - ( - 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 +145,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 @@ -278,7 +288,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, + data_source: DataSource, + 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 @@ -385,69 +401,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 +565,10 @@ 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/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 eba16015d35..317001885b5 100644 --- a/sdk/python/feast/value_type.py +++ b/sdk/python/feast/value_type.py @@ -11,7 +11,6 @@ # 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 tensorflow_metadata.proto.v0 import schema_pb2 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( diff --git a/sdk/python/tests/test_bigquery_ingestion.py b/sdk/python/tests/test_bigquery_ingestion.py new file mode 100644 index 00000000000..db6637bba1c --- /dev/null +++ b/sdk/python/tests/test_bigquery_ingestion.py @@ -0,0 +1,91 @@ +import random +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.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 + + +@pytest.mark.integration +class TestMaterializeFromBigQueryToDatastore: + 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"]) + + 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) + + # 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() + + # 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]) + + # 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