Skip to content

Commit c8ddd76

Browse files
jklegarjklegar
committed
Address comments
Signed-off-by: Jacob Klegar <jacob@tecton.ai>
1 parent 14ba58d commit c8ddd76

2 files changed

Lines changed: 77 additions & 64 deletions

File tree

‎sdk/python/feast/feature_store.py‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -176,8 +176,8 @@ def materialize(
176176
"""
177177
Materialize data from the offline store into the online store.
178178
179-
This method materializes feature data in the specified interval from either
180-
the specified feature views or all feature views if none are specified
179+
This method loads feature data in the specified interval from either
180+
the specified feature views, or all feature views if none are specified,
181181
into the online store where it is available for online serving.
182182
183183
Args:
@@ -186,17 +186,19 @@ def materialize(
186186
start_date (datetime): Start date of query
187187
end_date (datetime): End date of query
188188
"""
189-
full_feature_views = []
189+
feature_views_to_materialize = []
190190
registry = self._get_registry()
191191
if feature_views is None:
192-
full_feature_views = registry.list_feature_views(self.config.project)
192+
feature_views_to_materialize = registry.list_feature_views(
193+
self.config.project
194+
)
193195
else:
194196
for name in feature_views:
195197
feature_view = registry.get_feature_view(name, self.config.project)
196-
full_feature_views.append(feature_view)
198+
feature_views_to_materialize.append(feature_view)
197199

198200
# TODO paging large loads
199-
for feature_view in full_feature_views:
201+
for feature_view in feature_views_to_materialize:
200202
if isinstance(feature_view.input, FileSource):
201203
raise NotImplementedError(
202204
"This function is not yet implemented for File data sources"

‎sdk/python/tests/test_bigquery_ingestion.py‎

Lines changed: 69 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import random
12
import time
23
from datetime import datetime, timedelta
34

@@ -16,65 +17,75 @@
1617

1718

1819
@pytest.mark.integration
19-
def test_bigquery_ingestion_correctness():
20-
# create dataset
21-
ts = pd.Timestamp.now(tz="UTC").round("ms")
22-
data = {
23-
"id": [1, 2, 1],
24-
"value": [0.1, 0.2, 0.3],
25-
"ts_1": [ts - timedelta(minutes=2), ts, ts],
26-
"created_ts": [ts, ts, ts],
27-
}
28-
df = pd.DataFrame.from_dict(data)
20+
class TestBigQueryIngestion:
21+
def setup_method(self):
22+
self.client = bigquery.Client()
23+
self.gcp_project = self.client.project
24+
self.bigquery_dataset = "test_ingestion"
25+
dataset = bigquery.Dataset(f"{self.gcp_project}.{self.bigquery_dataset}")
26+
self.client.create_dataset(dataset, exists_ok=True)
27+
dataset.default_table_expiration_ms = (
28+
1000 * 60 * 60 * 24 * 14
29+
) # 2 weeks in milliseconds
30+
self.client.update_dataset(dataset, ["default_table_expiration_ms"])
2931

30-
# load dataset into BigQuery
31-
client = bigquery.Client()
32-
job_config = bigquery.LoadJobConfig()
33-
gcp_project = client.project
34-
bigquery_dataset = "test_ingestion"
35-
dataset = bigquery.Dataset(f"{gcp_project}.{bigquery_dataset}")
36-
client.create_dataset(dataset, exists_ok=True)
37-
dataset.default_table_expiration_ms = (
38-
1000 * 60 * 60 * 24 * 14
39-
) # 2 weeks in milliseconds
40-
client.update_dataset(dataset, ["default_table_expiration_ms"])
41-
table_id = f"{gcp_project}.{bigquery_dataset}.table_{int(time.time())}"
42-
job = client.load_table_from_dataframe(df, table_id, job_config=job_config)
43-
job.result()
32+
def test_bigquery_ingestion_correctness(self):
33+
# create dataset
34+
ts = pd.Timestamp.now(tz="UTC").round("ms")
35+
checked_value = (
36+
random.random()
37+
) # random value so test doesn't still work if no values written to online store
38+
data = {
39+
"id": [1, 2, 1],
40+
"value": [0.1, 0.2, checked_value],
41+
"ts_1": [ts - timedelta(minutes=2), ts, ts],
42+
"created_ts": [ts, ts, ts],
43+
}
44+
df = pd.DataFrame.from_dict(data)
4445

45-
# create FeatureView
46-
fv = FeatureView(
47-
name="test_fv",
48-
entities=["driver_id"],
49-
features=[Feature("value", ValueType.FLOAT)],
50-
ttl=timedelta(minutes=5),
51-
input=BigQuerySource(
52-
event_timestamp_column="ts",
53-
table_ref=table_id,
54-
created_timestamp_column="created_ts",
55-
field_mapping={"ts_1": "ts", "id": "driver_id"},
56-
date_partition_column="",
57-
),
58-
)
59-
config = RepoConfig(
60-
metadata_store="./metadata.db",
61-
project="default",
62-
provider="gcp",
63-
online_store=OnlineStoreConfig(local=LocalOnlineStoreConfig("online_store.db")),
64-
)
65-
fs = FeatureStore(config=config)
66-
fs.apply([fv])
46+
# load dataset into BigQuery
47+
job_config = bigquery.LoadJobConfig()
48+
table_id = (
49+
f"{self.gcp_project}.{self.bigquery_dataset}.correctness_{int(time.time())}"
50+
)
51+
job = self.client.load_table_from_dataframe(df, table_id, job_config=job_config)
52+
job.result()
6753

68-
# run materialize()
69-
fs.materialize(
70-
["test_fv"],
71-
datetime.utcnow() - timedelta(minutes=5),
72-
datetime.utcnow() - timedelta(minutes=0),
73-
)
54+
# create FeatureView
55+
fv = FeatureView(
56+
name="test_bq_correctness",
57+
entities=["driver_id"],
58+
features=[Feature("value", ValueType.FLOAT)],
59+
ttl=timedelta(minutes=5),
60+
input=BigQuerySource(
61+
event_timestamp_column="ts",
62+
table_ref=table_id,
63+
created_timestamp_column="created_ts",
64+
field_mapping={"ts_1": "ts", "id": "driver_id"},
65+
date_partition_column="",
66+
),
67+
)
68+
config = RepoConfig(
69+
metadata_store="./metadata.db",
70+
project="default",
71+
provider="gcp",
72+
online_store=OnlineStoreConfig(
73+
local=LocalOnlineStoreConfig("online_store.db")
74+
),
75+
)
76+
fs = FeatureStore(config=config)
77+
fs.apply([fv])
7478

75-
# check result of materialize()
76-
entity_key = EntityKeyProto(
77-
entity_names=["driver_id"], entity_values=[ValueProto(int32_val=1)]
78-
)
79-
_, val = fs._get_provider().online_read("default", fv, entity_key)
80-
assert abs(val["value"].double_val - 0.3) < 1e-6
79+
# run materialize()
80+
fs.materialize(
81+
["test_bq_correctness"],
82+
datetime.utcnow() - timedelta(minutes=5),
83+
datetime.utcnow() - timedelta(minutes=0),
84+
)
85+
86+
# check result of materialize()
87+
entity_key = EntityKeyProto(
88+
entity_names=["driver_id"], entity_values=[ValueProto(int64_val=1)]
89+
)
90+
t, val = fs._get_provider().online_read("default", fv, entity_key)
91+
assert abs(val["value"].double_val - checked_value) < 1e-6

0 commit comments

Comments
 (0)