Skip to content

Commit 5507787

Browse files
woopwoop
committed
Add basic e2e tests for Feast (excludes batch retrieval)
1 parent e8a9fae commit 5507787

7 files changed

Lines changed: 363 additions & 0 deletions

File tree

‎tests/e2e/basic/cust_trans_fs.yaml‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
name: customer_transactions
2+
kind: feature_set
3+
entities:
4+
- name: customer_id
5+
valueType: INT64
6+
features:
7+
- name: daily_transactions
8+
valueType: FLOAT
9+
- name: total_transactions
10+
valueType: FLOAT
11+
maxAge: 3600s

‎tests/e2e/basic/data.csv‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
datetime,customer,daily_transactions,total_transactions
2+
1570366527,1001,1.3,500
3+
1570366536,1002,1.4,600

‎tests/e2e/conftest.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
def pytest_addoption(parser):
2+
parser.addoption("--core_url", action="store", default="localhost:6565")
3+
parser.addoption("--serving_url", action="store", default="localhost:6565")
4+
parser.addoption("--allow_dirty", action="store", default="false")

‎tests/e2e/large_volume/cust_trans_large_fs.yaml‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
name: customer_transactions_large
2+
kind: feature_set
3+
entities:
4+
- name: customer_id
5+
valueType: INT64
6+
features:
7+
- name: daily_transactions
8+
valueType: FLOAT
9+
- name: total_transactions
10+
valueType: FLOAT
11+
maxAge: 3600s

‎tests/e2e/pytest.ini‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
[pytest]
2+
filterwarnings =
3+
ignore::DeprecationWarning

‎tests/e2e/requirements.txt‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
mock==2.0.0
2+
numpy==1.16.4
3+
pandas==0.24.2
4+
pytest==5.2.1
5+
pytest-benchmark==3.2.2
6+
pytest-mock==1.10.4
7+
pytest-timeout==1.3.3

‎tests/e2e/test_e2e.py‎

Lines changed: 324 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,324 @@
1+
import math
2+
import random
3+
import time
4+
from feast.entity import Entity
5+
from feast.serving.ServingService_pb2 import (
6+
GetOnlineFeaturesRequest,
7+
GetOnlineFeaturesResponse,
8+
)
9+
from feast.types.Value_pb2 import Value as Value
10+
from feast.client import Client
11+
from feast.feature_set import FeatureSet
12+
from feast.type_map import ValueType
13+
from google.protobuf.duration_pb2 import Duration
14+
import pytest
15+
from datetime import datetime
16+
import pytz
17+
18+
import pandas as pd
19+
import numpy as np
20+
21+
from feast.feature import Feature
22+
23+
24+
FLOAT_TOLERANCE = 0.00001
25+
26+
27+
@pytest.fixture()
28+
def core_url(pytestconfig):
29+
return pytestconfig.getoption("core_url")
30+
31+
32+
@pytest.fixture()
33+
def serving_url(pytestconfig):
34+
return pytestconfig.getoption("serving_url")
35+
36+
37+
@pytest.fixture()
38+
def allow_dirty(pytestconfig):
39+
return True if pytestconfig.getoption("allow_dirty").lower() == "true" else False
40+
41+
42+
@pytest.fixture
43+
def client(core_url, serving_url, allow_dirty):
44+
# Get client for core and serving
45+
client = Client(core_url=core_url, serving_url=serving_url)
46+
47+
# Ensure Feast core is active, but empty
48+
if not allow_dirty:
49+
feature_sets = client.list_feature_sets()
50+
if len(feature_sets) > 0:
51+
raise Exception(
52+
"Feast cannot have existing feature sets registered. Exiting tests."
53+
)
54+
55+
return client
56+
57+
58+
@pytest.mark.timeout(300)
59+
def test_basic(client):
60+
61+
cust_trans_fs = client.get_feature_set(name="customer_transactions", version=1)
62+
63+
# TODO: Fix source handling in Feast Core to support true idempotent
64+
# applies. In this case, applying a feature set without a source will
65+
# create a new feature set every time.
66+
67+
if cust_trans_fs is None:
68+
# Load feature set from file
69+
cust_trans_fs = FeatureSet.from_yaml("basic/cust_trans_fs.yaml")
70+
71+
# Register feature set
72+
client.apply(cust_trans_fs)
73+
74+
cust_trans_fs = client.get_feature_set(name="customer_transactions", version=1)
75+
76+
offset = random.randint(1000, 100000) # ensure a unique key space is used
77+
customer_data = pd.DataFrame(
78+
{
79+
"datetime": [datetime.utcnow().replace(tzinfo=pytz.utc) for _ in range(5)],
80+
"customer_id": [offset + inc for inc in range(5)],
81+
"daily_transactions": [np.random.rand() for _ in range(5)],
82+
"total_transactions": [512 for _ in range(5)],
83+
}
84+
)
85+
86+
# Ingest customer transaction data
87+
cust_trans_fs.ingest(dataframe=customer_data)
88+
89+
# Poll serving for feature values until the correct values are returned
90+
while True:
91+
response = client.get_online_features(
92+
entity_rows=[
93+
GetOnlineFeaturesRequest.EntityRow(
94+
fields={
95+
"customer_id": Value(
96+
int64_val=customer_data.iloc[0]["customer_id"]
97+
)
98+
}
99+
)
100+
],
101+
feature_ids=[
102+
"customer_transactions:1:daily_transactions",
103+
"customer_transactions:1:total_transactions",
104+
],
105+
) # type: GetOnlineFeaturesResponse
106+
if response is None:
107+
time.sleep(1)
108+
continue
109+
110+
returned_daily_transactions = float(
111+
response.field_values[0]
112+
.fields["customer_transactions:1:daily_transactions"]
113+
.float_val
114+
)
115+
sent_daily_transactions = float(customer_data.iloc[0]["daily_transactions"])
116+
117+
if math.isclose(
118+
sent_daily_transactions,
119+
returned_daily_transactions,
120+
abs_tol=FLOAT_TOLERANCE,
121+
):
122+
break
123+
124+
125+
@pytest.mark.timeout(300)
126+
def test_all_types(client):
127+
all_types_fs = client.get_feature_set(name="all_types", version="1")
128+
129+
if all_types_fs is None:
130+
# Register new feature set if it doesnt exist
131+
all_types_fs = FeatureSet(
132+
name="all_types",
133+
entities=[Entity(name="user_id", dtype=ValueType.INT64)],
134+
features=[
135+
Feature(name="float_feature", dtype=ValueType.FLOAT),
136+
Feature(name="int64_feature", dtype=ValueType.INT64),
137+
Feature(name="int32_feature", dtype=ValueType.INT32),
138+
Feature(name="string_feature", dtype=ValueType.STRING),
139+
Feature(name="bytes_feature", dtype=ValueType.BYTES),
140+
Feature(name="bool_feature", dtype=ValueType.BOOL),
141+
Feature(name="double_feature", dtype=ValueType.DOUBLE),
142+
Feature(name="float_list_feature", dtype=ValueType.FLOAT_LIST),
143+
Feature(name="int64_list_feature", dtype=ValueType.INT64_LIST),
144+
Feature(name="int32_list_feature", dtype=ValueType.INT32_LIST),
145+
Feature(name="string_list_feature", dtype=ValueType.STRING_LIST),
146+
Feature(name="bytes_list_feature", dtype=ValueType.BYTES_LIST),
147+
Feature(name="bool_list_feature", dtype=ValueType.BOOL_LIST),
148+
Feature(name="double_list_feature", dtype=ValueType.DOUBLE_LIST),
149+
],
150+
max_age=Duration(seconds=3600),
151+
)
152+
153+
# Register feature set
154+
client.apply(all_types_fs)
155+
all_types_fs = client.get_feature_set(name="all_types", version="1")
156+
157+
all_types_df = pd.DataFrame(
158+
{
159+
"datetime": [datetime.utcnow().replace(tzinfo=pytz.utc) for _ in range(3)],
160+
"user_id": [1001, 1002, 1003],
161+
"int32_feature": [np.int32(1), np.int32(2), np.int32(3)],
162+
"int64_feature": [np.int64(1), np.int64(2), np.int64(3)],
163+
"float_feature": [np.float(0.1), np.float(0.2), np.float(0.3)],
164+
"double_feature": [np.float64(0.1), np.float64(0.2), np.float64(0.3)],
165+
"string_feature": ["one", "two", "three"],
166+
"bytes_feature": [b"one", b"two", b"three"],
167+
"bool_feature": [True, False, False],
168+
"int32_list_feature": [
169+
np.array([1, 2, 3, 4], dtype=np.int32),
170+
np.array([1, 2, 3, 4], dtype=np.int32),
171+
np.array([1, 2, 3, 4], dtype=np.int32),
172+
],
173+
"int64_list_feature": [
174+
np.array([1, 2, 3, 4], dtype=np.int64),
175+
np.array([1, 2, 3, 4], dtype=np.int64),
176+
np.array([1, 2, 3, 4], dtype=np.int64),
177+
],
178+
"float_list_feature": [
179+
np.array([1.1, 1.2, 1.3, 1.4], dtype=np.float32),
180+
np.array([1.1, 1.2, 1.3, 1.4], dtype=np.float32),
181+
np.array([1.1, 1.2, 1.3, 1.4], dtype=np.float32),
182+
],
183+
"double_list_feature": [
184+
np.array([1.1, 1.2, 1.3, 1.4], dtype=np.float64),
185+
np.array([1.1, 1.2, 1.3, 1.4], dtype=np.float64),
186+
np.array([1.1, 1.2, 1.3, 1.4], dtype=np.float64),
187+
],
188+
"string_list_feature": [
189+
np.array(["one", "two", "three"]),
190+
np.array(["one", "two", "three"]),
191+
np.array(["one", "two", "three"]),
192+
],
193+
"bytes_list_feature": [
194+
np.array([b"one", b"two", b"three"]),
195+
np.array([b"one", b"two", b"three"]),
196+
np.array([b"one", b"two", b"three"]),
197+
],
198+
"bool_list_feature": [
199+
np.array([True, False, True]),
200+
np.array([True, False, True]),
201+
np.array([True, False, True]),
202+
],
203+
}
204+
)
205+
206+
# Ingest user embedding data
207+
all_types_fs.ingest(dataframe=all_types_df)
208+
209+
# Poll serving for feature values until the correct values are returned
210+
while True:
211+
response = client.get_online_features(
212+
entity_rows=[
213+
GetOnlineFeaturesRequest.EntityRow(
214+
fields={"user_id": Value(int64_val=all_types_df.iloc[0]["user_id"])}
215+
)
216+
],
217+
feature_ids=[
218+
"all_types:1:float_feature",
219+
"all_types:1:int64_feature",
220+
"all_types:1:int32_feature",
221+
"all_types:1:string_feature",
222+
"all_types:1:bytes_feature",
223+
"all_types:1:bool_feature",
224+
"all_types:1:double_feature",
225+
"all_types:1:float_list_feature",
226+
"all_types:1:int64_list_feature",
227+
"all_types:1:int32_list_feature",
228+
"all_types:1:string_list_feature",
229+
"all_types:1:bytes_list_feature",
230+
"all_types:1:bool_list_feature",
231+
"all_types:1:double_list_feature",
232+
],
233+
) # type: GetOnlineFeaturesResponse
234+
235+
if response is None:
236+
time.sleep(1)
237+
continue
238+
239+
returned_float_list = (
240+
response.field_values[0]
241+
.fields["all_types:1:float_list_feature"]
242+
.float_list_val.val
243+
)
244+
245+
sent_float_list = all_types_df.iloc[0]["float_list_feature"]
246+
247+
# TODO: Add tests for each value and type
248+
if math.isclose(
249+
returned_float_list[0], sent_float_list[0], abs_tol=FLOAT_TOLERANCE
250+
):
251+
break
252+
253+
# Wait for values to appear in Serving
254+
time.sleep(1)
255+
256+
257+
@pytest.mark.timeout(600)
258+
def test_large_volume(client):
259+
ROW_COUNT = 50000
260+
261+
cust_trans_fs = client.get_feature_set(
262+
name="customer_transactions_large", version=1
263+
)
264+
if cust_trans_fs is None:
265+
# Load feature set from file
266+
cust_trans_fs = FeatureSet.from_yaml("large_volume/cust_trans_large_fs.yaml")
267+
268+
# Register feature set
269+
client.apply(cust_trans_fs)
270+
271+
cust_trans_fs = client.get_feature_set(
272+
name="customer_transactions_large", version=1
273+
)
274+
275+
offset = random.randint(1000000, 10000000) # ensure a unique key space
276+
customer_data = pd.DataFrame(
277+
{
278+
"datetime": [
279+
datetime.utcnow().replace(tzinfo=pytz.utc) for _ in range(ROW_COUNT)
280+
],
281+
"customer_id": [offset + inc for inc in range(ROW_COUNT)],
282+
"daily_transactions": [np.random.rand() for _ in range(ROW_COUNT)],
283+
"total_transactions": [256 for _ in range(ROW_COUNT)],
284+
}
285+
)
286+
287+
# Ingest customer transaction data
288+
cust_trans_fs.ingest(dataframe=customer_data)
289+
290+
# Poll serving for feature values until the correct values are returned
291+
while True:
292+
response = client.get_online_features(
293+
entity_rows=[
294+
GetOnlineFeaturesRequest.EntityRow(
295+
fields={
296+
"customer_id": Value(
297+
int64_val=customer_data.iloc[0]["customer_id"]
298+
)
299+
}
300+
)
301+
],
302+
feature_ids=[
303+
"customer_transactions_large:1:daily_transactions",
304+
"customer_transactions_large:1:total_transactions",
305+
],
306+
) # type: GetOnlineFeaturesResponse
307+
308+
if response is None:
309+
time.sleep(1)
310+
continue
311+
312+
returned_daily_transactions = float(
313+
response.field_values[0]
314+
.fields["customer_transactions_large:1:daily_transactions"]
315+
.float_val
316+
)
317+
sent_daily_transactions = float(customer_data.iloc[0]["daily_transactions"])
318+
319+
if math.isclose(
320+
sent_daily_transactions,
321+
returned_daily_transactions,
322+
abs_tol=FLOAT_TOLERANCE,
323+
):
324+
break

0 commit comments

Comments
 (0)