From 19cad3df9e311420c4c5382020333945452b61c1 Mon Sep 17 00:00:00 2001 From: Terence Date: Tue, 9 Jun 2020 21:31:15 +0800 Subject: [PATCH 01/17] Update CoreService proto --- protos/feast/core/CoreService.proto | 2 ++ 1 file changed, 2 insertions(+) diff --git a/protos/feast/core/CoreService.proto b/protos/feast/core/CoreService.proto index 05755b701d8..6a9bfd8ff28 100644 --- a/protos/feast/core/CoreService.proto +++ b/protos/feast/core/CoreService.proto @@ -140,6 +140,8 @@ message ListFeatureSetsRequest { // - my-feature-set* can be used to match all features prefixed by "my-feature-set" // - my-feature-set-6 can be used to select a single feature set string feature_set_name = 1; + + map labels = 4; } } From 43466360078f3cac48155f90141ee758a0a76454 Mon Sep 17 00:00:00 2001 From: Terence Date: Tue, 9 Jun 2020 21:32:32 +0800 Subject: [PATCH 02/17] Update SpecService and add unit test --- .../java/feast/core/service/SpecService.java | 14 +++- .../feast/core/service/SpecServiceTest.java | 64 ++++++++++++++++++- 2 files changed, 75 insertions(+), 3 deletions(-) diff --git a/core/src/main/java/feast/core/service/SpecService.java b/core/src/main/java/feast/core/service/SpecService.java index 01cd264c761..d8808a0e993 100644 --- a/core/src/main/java/feast/core/service/SpecService.java +++ b/core/src/main/java/feast/core/service/SpecService.java @@ -47,6 +47,7 @@ import feast.proto.core.StoreProto.Store.Subscription; import java.util.ArrayList; import java.util.List; +import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -133,6 +134,7 @@ public ListFeatureSetsResponse listFeatureSets(ListFeatureSetsRequest.Filter fil throws InvalidProtocolBufferException { String name = filter.getFeatureSetName(); String project = filter.getProject(); + Map labels = filter.getLabelsMap(); if (name.isEmpty()) { throw new IllegalArgumentException( @@ -189,7 +191,17 @@ public ListFeatureSetsResponse listFeatureSets(ListFeatureSetsRequest.Filter fil ListFeatureSetsResponse.Builder response = ListFeatureSetsResponse.newBuilder(); if (featureSets.size() > 0) { for (FeatureSet featureSet : featureSets) { - response.addFeatureSets(featureSet.toProto()); + if (labels != null && !labels.isEmpty()) { + Map tempLabels = featureSet.toProto().getSpec().getLabelsMap(); + for (Map.Entry entry : labels.entrySet()) { + if (tempLabels.containsKey(entry.getKey()) + && tempLabels.get(entry.getKey()).equals(entry.getValue())) { + response.addFeatureSets(featureSet.toProto()); + } + } + } else { + response.addFeatureSets(featureSet.toProto()); + } } } diff --git a/core/src/test/java/feast/core/service/SpecServiceTest.java b/core/src/test/java/feast/core/service/SpecServiceTest.java index b5fd03fc7fc..e9eaf30d3ce 100644 --- a/core/src/test/java/feast/core/service/SpecServiceTest.java +++ b/core/src/test/java/feast/core/service/SpecServiceTest.java @@ -96,7 +96,7 @@ public class SpecServiceTest { // TODO: Updates update features in place, so if tests follow the wrong order they might break. // Refactor this maybe? @Before - public void setUp() { + public void setUp() throws InvalidProtocolBufferException { initMocks(this); defaultSource = TestObjectFactory.defaultSource; @@ -111,7 +111,50 @@ public void setUp() { "f3", "project1", Arrays.asList(f3e1), Arrays.asList(f3f2, f3f1)); FeatureSet featureSet4 = newDummyFeatureSet("f4", Project.DEFAULT_NAME); - featureSets = Arrays.asList(featureSet1, featureSet2, featureSet3, featureSet4); + Map singleFeatureSetLabels = + new HashMap<>() { + { + put("fsLabel1", "fsValue1"); + } + }; + Map duoFeatureSetLabels = + new HashMap<>() { + { + put("fsLabel1", "fsValue1"); + put("fsLabel2", "fsValue2"); + } + }; + FeatureSet featureSet5 = newDummyFeatureSet("f5", Project.DEFAULT_NAME); + FeatureSet featureSet6 = newDummyFeatureSet("f6", Project.DEFAULT_NAME); + FeatureSetSpec featureSetSpec5 = featureSet5.toProto().getSpec().toBuilder().build(); + FeatureSetSpec featureSetSpec6 = featureSet6.toProto().getSpec().toBuilder().build(); + FeatureSetProto.FeatureSet fs5 = + FeatureSetProto.FeatureSet.newBuilder() + .setSpec( + featureSetSpec5 + .toBuilder() + .setSource(defaultSource.toProto()) + .putAllLabels(singleFeatureSetLabels) + .build()) + .build(); + FeatureSetProto.FeatureSet fs6 = + FeatureSetProto.FeatureSet.newBuilder() + .setSpec( + featureSetSpec6 + .toBuilder() + .setSource(defaultSource.toProto()) + .putAllLabels(duoFeatureSetLabels) + .build()) + .build(); + + featureSets = + Arrays.asList( + featureSet1, + featureSet2, + featureSet3, + featureSet4, + FeatureSet.fromProto(fs5), + FeatureSet.fromProto(fs6)); when(featureSetRepository.findAll()).thenReturn(featureSets); when(featureSetRepository.findAllByOrderByNameAsc()).thenReturn(featureSets); @@ -626,6 +669,23 @@ public void applyFeatureSetShouldAcceptFeatureSetLabels() throws InvalidProtocol assertEquals(featureSetLabels, appliedLabels); } + @Test + public void shouldFilterByFeatureSetLabels() throws InvalidProtocolBufferException { + ListFeatureSetsResponse actual = + specService.listFeatureSets( + Filter.newBuilder() + .setFeatureSetName("*") + .setProject("*") + .putLabels("fsLabel2", "fsValue2") + .build()); + List list = new ArrayList<>(); + list.add(featureSets.get(featureSets.size() - 1).toProto()); + + ListFeatureSetsResponse expected = + ListFeatureSetsResponse.newBuilder().addAllFeatureSets(list).build(); + assertThat(actual, equalTo(expected)); + } + @Test public void shouldUpdateStoreIfConfigChanges() throws InvalidProtocolBufferException { when(storeRepository.findById("SERVING")).thenReturn(Optional.of(stores.get(0))); From 452764968af3269f73b7cb2b1faae267806c52bd Mon Sep 17 00:00:00 2001 From: Terence Date: Tue, 9 Jun 2020 21:33:38 +0800 Subject: [PATCH 03/17] Update Python sdk and cli --- sdk/python/feast/cli.py | 38 ++++++++++++++++++++++++++++++++++++-- sdk/python/feast/client.py | 9 +++++++-- 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/sdk/python/feast/cli.py b/sdk/python/feast/cli.py index 4db41f92f18..750591e9e9e 100644 --- a/sdk/python/feast/cli.py +++ b/sdk/python/feast/cli.py @@ -121,14 +121,48 @@ def feature_set(): @feature_set.command(name="list") -def feature_set_list(): +@click.option( + "--project", + "-p", + help="Project that feature set belongs to", + type=click.STRING, + default="*", +) +@click.option( + "--name", + "-n", + help="Name that feature set belongs to, may include * to match multiple feature sets eg. driver_*", + type=click.STRING, + default="*", +) +@click.option( + "--labels", + "-l", + help="Labels to filter for feature sets", + type=click.STRING, + default="", +) +def feature_set_list(project: str, name: str, labels: str): """ List all feature sets """ feast_client = Client() # type: Client + labels_kv = labels.split(",") + labels_dict = {} + if labels == "": + pass + elif len(labels_kv) % 2 == 0: + for k, v in zip(labels_kv[0::2], labels_kv[1::2]): + labels_dict[k] = v + else: + print(f"Uneven key-value label pairs were entered") + return + table = [] - for fs in feast_client.list_feature_sets(project="*", name="*"): + for fs in feast_client.list_feature_sets( + project=project, name=name, labels=labels_dict + ): table.append([fs.name, repr(fs)]) from tabulate import tabulate diff --git a/sdk/python/feast/client.py b/sdk/python/feast/client.py index 316573ce93d..5ea39fa9884 100644 --- a/sdk/python/feast/client.py +++ b/sdk/python/feast/client.py @@ -425,7 +425,7 @@ def _apply_feature_set(self, feature_set: FeatureSet): feature_set._update_from_feature_set(applied_fs) def list_feature_sets( - self, project: str = None, name: str = None, + self, project: str = None, name: str = None, labels: Dict[str, str] = None ) -> List[FeatureSet]: """ Retrieve a list of feature sets from Feast Core @@ -448,7 +448,12 @@ def list_feature_sets( if name is None: name = "*" - filter = ListFeatureSetsRequest.Filter(project=project, feature_set_name=name) + if labels is None: + labels = {} + + filter = ListFeatureSetsRequest.Filter( + project=project, feature_set_name=name, labels=labels + ) # Get latest feature sets from Feast Core feature_set_protos = self._core_service_stub.ListFeatureSets( From 81322f9edc39798a3d186d30bc5c6a288695085c Mon Sep 17 00:00:00 2001 From: Terence Date: Tue, 9 Jun 2020 21:42:18 +0800 Subject: [PATCH 04/17] Fix python lint error --- sdk/python/feast/cli.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/python/feast/cli.py b/sdk/python/feast/cli.py index 750591e9e9e..a002ca74df4 100644 --- a/sdk/python/feast/cli.py +++ b/sdk/python/feast/cli.py @@ -156,7 +156,7 @@ def feature_set_list(project: str, name: str, labels: str): for k, v in zip(labels_kv[0::2], labels_kv[1::2]): labels_dict[k] = v else: - print(f"Uneven key-value label pairs were entered") + print("Uneven key-value label pairs were entered") return table = [] From 64ea3828dbe54d7cf5c99d431feab09b9ce0fdaa Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 10 Jun 2020 10:42:34 +0800 Subject: [PATCH 05/17] Add more tests --- .../feast/core/service/SpecServiceTest.java | 22 +++++-- sdk/python/tests/test_client.py | 62 +++++++++++++++++++ tests/e2e/redis/basic-ingest-redis-serving.py | 13 ++++ tests/e2e/redis/basic/driver_labels_fs.yaml | 14 +++++ 4 files changed, 106 insertions(+), 5 deletions(-) create mode 100644 tests/e2e/redis/basic/driver_labels_fs.yaml diff --git a/core/src/test/java/feast/core/service/SpecServiceTest.java b/core/src/test/java/feast/core/service/SpecServiceTest.java index e9eaf30d3ce..f367f915ad3 100644 --- a/core/src/test/java/feast/core/service/SpecServiceTest.java +++ b/core/src/test/java/feast/core/service/SpecServiceTest.java @@ -671,19 +671,31 @@ public void applyFeatureSetShouldAcceptFeatureSetLabels() throws InvalidProtocol @Test public void shouldFilterByFeatureSetLabels() throws InvalidProtocolBufferException { - ListFeatureSetsResponse actual = + List list = new ArrayList<>(); + ListFeatureSetsResponse actual1 = specService.listFeatureSets( Filter.newBuilder() .setFeatureSetName("*") .setProject("*") .putLabels("fsLabel2", "fsValue2") .build()); - List list = new ArrayList<>(); - list.add(featureSets.get(featureSets.size() - 1).toProto()); + list.add(featureSets.get(5).toProto()); + ListFeatureSetsResponse expected1 = + ListFeatureSetsResponse.newBuilder().addAllFeatureSets(list).build(); - ListFeatureSetsResponse expected = + ListFeatureSetsResponse actual2 = + specService.listFeatureSets( + Filter.newBuilder() + .setFeatureSetName("*") + .setProject("*") + .putLabels("fsLabel1", "fsValue1") + .build()); + list.add(0, featureSets.get(4).toProto()); + ListFeatureSetsResponse expected2 = ListFeatureSetsResponse.newBuilder().addAllFeatureSets(list).build(); - assertThat(actual, equalTo(expected)); + + assertThat(actual1, equalTo(expected1)); + assertThat(actual2, equalTo(expected2)); } @Test diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index ee89b83737a..85fea31dbb1 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -34,6 +34,7 @@ from feast.core.CoreService_pb2 import ( GetFeastCoreVersionResponse, GetFeatureSetResponse, + ListFeatureSetsResponse, ListIngestionJobsResponse, ) from feast.core.FeatureSet_pb2 import EntitySpec as EntitySpecProto @@ -321,6 +322,67 @@ def test_get_feature_set(self, mocked_client, mocker): and len(feature_set.entities) == 1 ) + @pytest.mark.parametrize( + "mocked_client", + [pytest.lazy_fixture("mock_client"), pytest.lazy_fixture("secure_mock_client")], + ) + def test_list_feature_sets(self, mocked_client, mocker): + mocker.patch.object( + mocked_client, + "_core_service_stub", + return_value=Core.CoreServiceStub(grpc.insecure_channel("")), + ) + + feature_set_1_proto = FeatureSetProto( + spec=FeatureSetSpecProto( + project="test", + name="driver_car", + max_age=Duration(seconds=3600), + labels={"key1": "val1", "key2": "val2"}, + features=[ + FeatureSpecProto( + name="feature_1", value_type=ValueProto.ValueType.FLOAT + ) + ], + ) + ) + feature_set_2_proto = FeatureSetProto( + spec=FeatureSetSpecProto( + project="test", + name="driver_ride", + max_age=Duration(seconds=3600), + labels={"key1": "val1"}, + features=[ + FeatureSpecProto( + name="feature_1", value_type=ValueProto.ValueType.FLOAT + ) + ], + ) + ) + + mocker.patch.object( + mocked_client._core_service_stub, + "ListFeatureSets", + return_value=ListFeatureSetsResponse( + feature_sets=[feature_set_1_proto, feature_set_2_proto] + ), + ) + + feature_sets = mocked_client.list_feature_sets(labels={"key1": "val1"}) + assert len(feature_sets) == 2 + + feature_set = feature_sets[0] + assert ( + feature_set.name == "driver_car" + and "key1" in feature_set.labels + and feature_set.labels["key1"] == "val1" + and "key2" in feature_set.labels + and feature_set.labels["key2"] == "val2" + and feature_set.fields["feature_1"].name == "feature_1" + and feature_set.fields["feature_1"].dtype == ValueType.FLOAT + and len(feature_set.features) == 1 + ) + @pytest.mark.parametrize( "mocked_client", [pytest.lazy_fixture("mock_client"), pytest.lazy_fixture("secure_mock_client")], diff --git a/tests/e2e/redis/basic-ingest-redis-serving.py b/tests/e2e/redis/basic-ingest-redis-serving.py index 63a3c794a5a..c893dca71ff 100644 --- a/tests/e2e/redis/basic-ingest-redis-serving.py +++ b/tests/e2e/redis/basic-ingest-redis-serving.py @@ -126,6 +126,19 @@ def test_basic_register_feature_set_success(client): project=PROJECT_NAME) assert cust_trans_fs_actual == cust_trans_fs_expected + # Register feature set with labels + driver_fs_expected = FeatureSet.from_yaml(f"{DIR_PATH}/basic/driver_fs.yaml") + driver_fs_labeled_expected = FeatureSet.from_yaml( + f"{DIR_PATH}/basic/driver_labels_fs.yaml" + ) + client.set_project(PROJECT_NAME) + client.apply(driver_fs_expected) + client.apply(driver_fs_labeled_expected) + driver_fs_actual = client.list_feature_sets( + project=PROJECT_NAME, labels={"key1": "val1"} + )[0] + assert driver_fs_actual == driver_fs_labeled_expected + # reset client's project for other tests client.set_project() diff --git a/tests/e2e/redis/basic/driver_labels_fs.yaml b/tests/e2e/redis/basic/driver_labels_fs.yaml new file mode 100644 index 00000000000..70aef796cc5 --- /dev/null +++ b/tests/e2e/redis/basic/driver_labels_fs.yaml @@ -0,0 +1,14 @@ +kind: feature_set +spec: + name: driver_labeled + entities: + - name: driver_id + valueType: INT64 + features: + - name: rating + valueType: FLOAT + - name: cost + valueType: FLOAT + maxAge: 3600s + labels: + key1: val1 From d82200f25383fed9a81ebd3b973fb4797208e5fe Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 10 Jun 2020 10:43:49 +0800 Subject: [PATCH 06/17] Black formatting changes --- tests/e2e/redis/basic-ingest-redis-serving.py | 224 +++++++++--------- 1 file changed, 107 insertions(+), 117 deletions(-) diff --git a/tests/e2e/redis/basic-ingest-redis-serving.py b/tests/e2e/redis/basic-ingest-redis-serving.py index c893dca71ff..444dc10c3d7 100644 --- a/tests/e2e/redis/basic-ingest-redis-serving.py +++ b/tests/e2e/redis/basic-ingest-redis-serving.py @@ -28,27 +28,26 @@ import uuid FLOAT_TOLERANCE = 0.00001 -PROJECT_NAME = 'basic_' + uuid.uuid4().hex.upper()[0:6] +PROJECT_NAME = "basic_" + uuid.uuid4().hex.upper()[0:6] DIR_PATH = os.path.dirname(os.path.realpath(__file__)) -@pytest.fixture(scope='module') +@pytest.fixture(scope="module") def core_url(pytestconfig): return pytestconfig.getoption("core_url") -@pytest.fixture(scope='module') +@pytest.fixture(scope="module") def serving_url(pytestconfig): return pytestconfig.getoption("serving_url") -@pytest.fixture(scope='module') +@pytest.fixture(scope="module") def allow_dirty(pytestconfig): - return True if pytestconfig.getoption( - "allow_dirty").lower() == "true" else False + return True if pytestconfig.getoption("allow_dirty").lower() == "true" else False -@pytest.fixture(scope='module') +@pytest.fixture(scope="module") def client(core_url, serving_url, allow_dirty): # Get client for core and serving client = Client(core_url=core_url, serving_url=serving_url) @@ -68,8 +67,7 @@ def client(core_url, serving_url, allow_dirty): def basic_dataframe(entities, features, ingest_time, n_size): offset = random.randint(1000, 100000) # ensure a unique key space is used df_dict = { - "datetime": [ingest_time.replace(tzinfo=pytz.utc) for _ in - range(n_size)], + "datetime": [ingest_time.replace(tzinfo=pytz.utc) for _ in range(n_size)], } for entity_name in entities: df_dict[entity_name] = list(range(1, n_size + 1)) @@ -85,31 +83,37 @@ def ingest_time(): @pytest.fixture(scope="module") def cust_trans_df(ingest_time): - return basic_dataframe(entities=["customer_id"], - features=["daily_transactions", "total_transactions"], - ingest_time=ingest_time, - n_size=5) + return basic_dataframe( + entities=["customer_id"], + features=["daily_transactions", "total_transactions"], + ingest_time=ingest_time, + n_size=5, + ) @pytest.fixture(scope="module") def driver_df(ingest_time): - return basic_dataframe(entities=["driver_id"], - features=["rating", "cost"], - ingest_time=ingest_time, - n_size=5) + return basic_dataframe( + entities=["driver_id"], + features=["rating", "cost"], + ingest_time=ingest_time, + n_size=5, + ) def test_version_returns_results(client): version_info = client.version() - assert not version_info['core'] is 'not configured' - assert not version_info['serving'] is 'not configured' + assert not version_info["core"] is "not configured" + assert not version_info["serving"] is "not configured" @pytest.mark.timeout(45) @pytest.mark.run(order=10) def test_basic_register_feature_set_success(client): # Register feature set without project - cust_trans_fs_expected = FeatureSet.from_yaml(f"{DIR_PATH}/basic/cust_trans_fs.yaml") + cust_trans_fs_expected = FeatureSet.from_yaml( + f"{DIR_PATH}/basic/cust_trans_fs.yaml" + ) driver_fs_expected = FeatureSet.from_yaml(f"{DIR_PATH}/basic/driver_fs.yaml") client.apply(cust_trans_fs_expected) client.apply(driver_fs_expected) @@ -119,11 +123,14 @@ def test_basic_register_feature_set_success(client): assert driver_fs_actual == driver_fs_expected # Register feature set with project - cust_trans_fs_expected = FeatureSet.from_yaml(f"{DIR_PATH}/basic/cust_trans_fs.yaml") + cust_trans_fs_expected = FeatureSet.from_yaml( + f"{DIR_PATH}/basic/cust_trans_fs.yaml" + ) client.set_project(PROJECT_NAME) client.apply(cust_trans_fs_expected) - cust_trans_fs_actual = client.get_feature_set("customer_transactions", - project=PROJECT_NAME) + cust_trans_fs_actual = client.get_feature_set( + "customer_transactions", project=PROJECT_NAME + ) assert cust_trans_fs_actual == cust_trans_fs_expected # Register feature set with labels @@ -172,27 +179,21 @@ def test_basic_retrieve_online_success(client, cust_trans_df): ) ], # Test retrieve with different variations of the string feature refs - feature_refs=[ - "daily_transactions", - "total_transactions", - ] + feature_refs=["daily_transactions", "total_transactions",], ) # type: GetOnlineFeaturesResponse if response is None: continue returned_daily_transactions = float( - response.field_values[0] - .fields["daily_transactions"] - .float_val + response.field_values[0].fields["daily_transactions"].float_val ) - sent_daily_transactions = float( - cust_trans_df.iloc[0]["daily_transactions"]) + sent_daily_transactions = float(cust_trans_df.iloc[0]["daily_transactions"]) if math.isclose( - sent_daily_transactions, - returned_daily_transactions, - abs_tol=FLOAT_TOLERANCE, + sent_daily_transactions, + returned_daily_transactions, + abs_tol=FLOAT_TOLERANCE, ): break @@ -217,9 +218,7 @@ def test_basic_retrieve_online_multiple_featureset(client, cust_trans_df, driver "customer_id": Value( int64_val=cust_trans_df.iloc[0]["customer_id"] ), - "driver_id": Value( - int64_val=driver_df.iloc[0]["driver_id"] - ) + "driver_id": Value(int64_val=driver_df.iloc[0]["driver_id"]), } ) ], @@ -231,9 +230,7 @@ def test_basic_retrieve_online_multiple_featureset(client, cust_trans_df, driver def check_response(ingest_df, response, feature_ref): returned_value = float( - response.field_values[0] - .fields[feature_ref] - .float_val + response.field_values[0].fields[feature_ref].float_val ) feature_ref_splits = feature_ref.split(":") if len(feature_ref_splits) == 1: @@ -241,16 +238,13 @@ def check_response(ingest_df, response, feature_ref): else: _, feature_name = feature_ref_splits - sent_value = float( - ingest_df.iloc[0][feature_name]) + sent_value = float(ingest_df.iloc[0][feature_name]) - return math.isclose( - sent_value, - returned_value, - abs_tol=FLOAT_TOLERANCE, - ) + return math.isclose(sent_value, returned_value, abs_tol=FLOAT_TOLERANCE,) - if all([check_response(df, response, ref) for ref, df in feature_ref_df_mapping]): + if all( + [check_response(df, response, ref) for ref, df in feature_ref_df_mapping] + ): break @@ -260,9 +254,12 @@ def test_basic_ingest_jobs(client): # list ingestion jobs given featureset cust_trans_fs = client.get_feature_set(name="customer_transactions") ingest_jobs = client.list_ingest_jobs( - feature_set_ref=FeatureSetRef.from_feature_set(cust_trans_fs)) + feature_set_ref=FeatureSetRef.from_feature_set(cust_trans_fs) + ) # filter ingestion jobs to only those that are running - ingest_jobs = [job for job in ingest_jobs if job.status == IngestionJobStatus.RUNNING] + ingest_jobs = [ + job for job in ingest_jobs if job.status == IngestionJobStatus.RUNNING + ] assert len(ingest_jobs) >= 1 for ingest_job in ingest_jobs: @@ -277,18 +274,16 @@ def test_basic_ingest_jobs(client): assert ingest_job.status == IngestionJobStatus.ABORTED -@pytest.fixture(scope='module') +@pytest.fixture(scope="module") def all_types_dataframe(): return pd.DataFrame( { - "datetime": [datetime.utcnow().replace(tzinfo=pytz.utc) for _ in - range(3)], + "datetime": [datetime.utcnow().replace(tzinfo=pytz.utc) for _ in range(3)], "user_id": [1001, 1002, 1003], "int32_feature": [np.int32(1), np.int32(2), np.int32(3)], "int64_feature": [np.int64(1), np.int64(2), np.int64(3)], "float_feature": [np.float(0.1), np.float(0.2), np.float(0.3)], - "double_feature": [np.float64(0.1), np.float64(0.2), - np.float64(0.3)], + "double_feature": [np.float64(0.1), np.float64(0.2), np.float64(0.3)], "string_feature": ["one", "two", "three"], "bytes_feature": [b"one", b"two", b"three"], "bool_feature": [True, False, False], @@ -350,8 +345,7 @@ def test_all_types_register_feature_set_success(client): Feature(name="float_list_feature", dtype=ValueType.FLOAT_LIST), Feature(name="int64_list_feature", dtype=ValueType.INT64_LIST), Feature(name="int32_list_feature", dtype=ValueType.INT32_LIST), - Feature(name="string_list_feature", - dtype=ValueType.STRING_LIST), + Feature(name="string_list_feature", dtype=ValueType.STRING_LIST), Feature(name="bytes_list_feature", dtype=ValueType.BYTES_LIST), ], max_age=Duration(seconds=3600), @@ -397,8 +391,11 @@ def test_all_types_retrieve_online_success(client, all_types_dataframe): response = client.get_online_features( entity_rows=[ GetOnlineFeaturesRequest.EntityRow( - fields={"user_id": Value( - int64_val=all_types_dataframe.iloc[0]["user_id"])} + fields={ + "user_id": Value( + int64_val=all_types_dataframe.iloc[0]["user_id"] + ) + } ) ], feature_refs=[ @@ -422,15 +419,13 @@ def test_all_types_retrieve_online_success(client, all_types_dataframe): continue returned_float_list = ( - response.field_values[0] - .fields["float_list_feature"] - .float_list_val.val + response.field_values[0].fields["float_list_feature"].float_list_val.val ) sent_float_list = all_types_dataframe.iloc[0]["float_list_feature"] if math.isclose( - returned_float_list[0], sent_float_list[0], abs_tol=FLOAT_TOLERANCE + returned_float_list[0], sent_float_list[0], abs_tol=FLOAT_TOLERANCE ): break @@ -441,9 +436,12 @@ def test_all_types_ingest_jobs(client, all_types_dataframe): # list ingestion jobs given featureset all_types_fs = client.get_feature_set(name="all_types") ingest_jobs = client.list_ingest_jobs( - feature_set_ref=FeatureSetRef.from_feature_set(all_types_fs)) + feature_set_ref=FeatureSetRef.from_feature_set(all_types_fs) + ) # filter ingestion jobs to only those that are running - ingest_jobs = [job for job in ingest_jobs if job.status == IngestionJobStatus.RUNNING] + ingest_jobs = [ + job for job in ingest_jobs if job.status == IngestionJobStatus.RUNNING + ] assert len(ingest_jobs) >= 1 for ingest_job in ingest_jobs: @@ -458,15 +456,14 @@ def test_all_types_ingest_jobs(client, all_types_dataframe): assert ingest_job.status == IngestionJobStatus.ABORTED -@pytest.fixture(scope='module') +@pytest.fixture(scope="module") def large_volume_dataframe(): ROW_COUNT = 100000 offset = random.randint(1000000, 10000000) # ensure a unique key space customer_data = pd.DataFrame( { "datetime": [ - datetime.utcnow().replace(tzinfo=pytz.utc) for _ in - range(ROW_COUNT) + datetime.utcnow().replace(tzinfo=pytz.utc) for _ in range(ROW_COUNT) ], "customer_id": [offset + inc for inc in range(ROW_COUNT)], "daily_transactions_large": [np.random.rand() for _ in range(ROW_COUNT)], @@ -480,7 +477,8 @@ def large_volume_dataframe(): @pytest.mark.run(order=30) def test_large_volume_register_feature_set_success(client): cust_trans_fs_expected = FeatureSet.from_yaml( - f"{DIR_PATH}/large_volume/cust_trans_large_fs.yaml") + f"{DIR_PATH}/large_volume/cust_trans_large_fs.yaml" + ) # Register feature set client.apply(cust_trans_fs_expected) @@ -488,8 +486,7 @@ def test_large_volume_register_feature_set_success(client): # Feast Core needs some time to fully commit the FeatureSet applied # when there is no existing job yet for the Featureset time.sleep(10) - cust_trans_fs_actual = client.get_feature_set( - name="customer_transactions_large") + cust_trans_fs_actual = client.get_feature_set(name="customer_transactions_large") assert cust_trans_fs_actual == cust_trans_fs_expected @@ -524,80 +521,73 @@ def test_large_volume_retrieve_online_success(client, large_volume_dataframe): GetOnlineFeaturesRequest.EntityRow( fields={ "customer_id": Value( - int64_val=large_volume_dataframe.iloc[0][ - "customer_id"] + int64_val=large_volume_dataframe.iloc[0]["customer_id"] ) } ) ], - feature_refs=[ - "daily_transactions_large", - "total_transactions_large", - ], + feature_refs=["daily_transactions_large", "total_transactions_large",], ) # type: GetOnlineFeaturesResponse if response is None: continue returned_daily_transactions = float( - response.field_values[0] - .fields["daily_transactions_large"] - .float_val + response.field_values[0].fields["daily_transactions_large"].float_val ) sent_daily_transactions = float( - large_volume_dataframe.iloc[0]["daily_transactions_large"]) + large_volume_dataframe.iloc[0]["daily_transactions_large"] + ) if math.isclose( - sent_daily_transactions, - returned_daily_transactions, - abs_tol=FLOAT_TOLERANCE, + sent_daily_transactions, + returned_daily_transactions, + abs_tol=FLOAT_TOLERANCE, ): break -@pytest.fixture(scope='module') +@pytest.fixture(scope="module") def all_types_parquet_file(): COUNT = 20000 df = pd.DataFrame( { "datetime": [datetime.utcnow() for _ in range(COUNT)], - "customer_id": [np.int32(random.randint(0, 10000)) for _ in - range(COUNT)], - "int32_feature_parquet": [np.int32(random.randint(0, 10000)) for _ in - range(COUNT)], - "int64_feature_parquet": [np.int64(random.randint(0, 10000)) for _ in - range(COUNT)], + "customer_id": [np.int32(random.randint(0, 10000)) for _ in range(COUNT)], + "int32_feature_parquet": [ + np.int32(random.randint(0, 10000)) for _ in range(COUNT) + ], + "int64_feature_parquet": [ + np.int64(random.randint(0, 10000)) for _ in range(COUNT) + ], "float_feature_parquet": [np.float(random.random()) for _ in range(COUNT)], - "double_feature_parquet": [np.float64(random.random()) for _ in - range(COUNT)], - "string_feature_parquet": ["one" + str(random.random()) for _ in - range(COUNT)], + "double_feature_parquet": [ + np.float64(random.random()) for _ in range(COUNT) + ], + "string_feature_parquet": [ + "one" + str(random.random()) for _ in range(COUNT) + ], "bytes_feature_parquet": [b"one" for _ in range(COUNT)], "int32_list_feature_parquet": [ np.array([1, 2, 3, random.randint(0, 10000)], dtype=np.int32) - for _ - in range(COUNT) + for _ in range(COUNT) ], "int64_list_feature_parquet": [ np.array([1, random.randint(0, 10000), 3, 4], dtype=np.int64) - for _ - in range(COUNT) + for _ in range(COUNT) ], "float_list_feature_parquet": [ - np.array([1.1, 1.2, 1.3, random.random()], dtype=np.float32) for - _ - in range(COUNT) + np.array([1.1, 1.2, 1.3, random.random()], dtype=np.float32) + for _ in range(COUNT) ], "double_list_feature_parquet": [ - np.array([1.1, 1.2, 1.3, random.random()], dtype=np.float64) for - _ - in range(COUNT) + np.array([1.1, 1.2, 1.3, random.random()], dtype=np.float64) + for _ in range(COUNT) ], "string_list_feature_parquet": [ - np.array(["one", "two" + str(random.random()), "three"]) for _ - in - range(COUNT) + np.array(["one", "two" + str(random.random()), "three"]) + for _ in range(COUNT) ], "bytes_list_feature_parquet": [ np.array([b"one", b"two", b"three"]) for _ in range(COUNT) @@ -608,7 +598,7 @@ def all_types_parquet_file(): # TODO: Boolean list is not being tested. # https://github.com/feast-dev/feast/issues/341 - file_path = os.path.join(tempfile.mkdtemp(), 'all_types.parquet') + file_path = os.path.join(tempfile.mkdtemp(), "all_types.parquet") df.to_parquet(file_path, allow_truncated_timestamps=True) return file_path @@ -618,7 +608,8 @@ def all_types_parquet_file(): def test_all_types_parquet_register_feature_set_success(client): # Load feature set from file all_types_parquet_expected = FeatureSet.from_yaml( - f"{DIR_PATH}/all_types_parquet/all_types_parquet.yaml") + f"{DIR_PATH}/all_types_parquet/all_types_parquet.yaml" + ) # Register feature set client.apply(all_types_parquet_expected) @@ -642,8 +633,7 @@ def test_all_types_parquet_register_feature_set_success(client): @pytest.mark.timeout(600) @pytest.mark.run(order=41) -def test_all_types_infer_register_ingest_file_success(client, - all_types_parquet_file): +def test_all_types_infer_register_ingest_file_success(client, all_types_parquet_file): # Get feature set all_types_fs = client.get_feature_set(name="all_types_parquet") @@ -660,14 +650,14 @@ class TestsBasedOnGrpc: @pytest.fixture(scope="module") def core_service_stub(self, core_url): if core_url.endswith(":443"): - core_channel = grpc.secure_channel( - core_url, grpc.ssl_channel_credentials() - ) + core_channel = grpc.secure_channel(core_url, grpc.ssl_channel_credentials()) else: core_channel = grpc.insecure_channel(core_url) try: - grpc.channel_ready_future(core_channel).result(timeout=self.GRPC_CONNECTION_TIMEOUT) + grpc.channel_ready_future(core_channel).result( + timeout=self.GRPC_CONNECTION_TIMEOUT + ) except grpc.FutureTimeoutError: raise ConnectionError( f"Connection timed out while attempting to connect to Feast " From c2969679d1a1b6177d7af45c52f818ecc1afe081 Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 10:09:29 +0800 Subject: [PATCH 07/17] Revert "Black formatting changes" This reverts commit d82200f25383fed9a81ebd3b973fb4797208e5fe. --- tests/e2e/redis/basic-ingest-redis-serving.py | 224 +++++++++--------- 1 file changed, 117 insertions(+), 107 deletions(-) diff --git a/tests/e2e/redis/basic-ingest-redis-serving.py b/tests/e2e/redis/basic-ingest-redis-serving.py index 444dc10c3d7..c893dca71ff 100644 --- a/tests/e2e/redis/basic-ingest-redis-serving.py +++ b/tests/e2e/redis/basic-ingest-redis-serving.py @@ -28,26 +28,27 @@ import uuid FLOAT_TOLERANCE = 0.00001 -PROJECT_NAME = "basic_" + uuid.uuid4().hex.upper()[0:6] +PROJECT_NAME = 'basic_' + uuid.uuid4().hex.upper()[0:6] DIR_PATH = os.path.dirname(os.path.realpath(__file__)) -@pytest.fixture(scope="module") +@pytest.fixture(scope='module') def core_url(pytestconfig): return pytestconfig.getoption("core_url") -@pytest.fixture(scope="module") +@pytest.fixture(scope='module') def serving_url(pytestconfig): return pytestconfig.getoption("serving_url") -@pytest.fixture(scope="module") +@pytest.fixture(scope='module') def allow_dirty(pytestconfig): - return True if pytestconfig.getoption("allow_dirty").lower() == "true" else False + return True if pytestconfig.getoption( + "allow_dirty").lower() == "true" else False -@pytest.fixture(scope="module") +@pytest.fixture(scope='module') def client(core_url, serving_url, allow_dirty): # Get client for core and serving client = Client(core_url=core_url, serving_url=serving_url) @@ -67,7 +68,8 @@ def client(core_url, serving_url, allow_dirty): def basic_dataframe(entities, features, ingest_time, n_size): offset = random.randint(1000, 100000) # ensure a unique key space is used df_dict = { - "datetime": [ingest_time.replace(tzinfo=pytz.utc) for _ in range(n_size)], + "datetime": [ingest_time.replace(tzinfo=pytz.utc) for _ in + range(n_size)], } for entity_name in entities: df_dict[entity_name] = list(range(1, n_size + 1)) @@ -83,37 +85,31 @@ def ingest_time(): @pytest.fixture(scope="module") def cust_trans_df(ingest_time): - return basic_dataframe( - entities=["customer_id"], - features=["daily_transactions", "total_transactions"], - ingest_time=ingest_time, - n_size=5, - ) + return basic_dataframe(entities=["customer_id"], + features=["daily_transactions", "total_transactions"], + ingest_time=ingest_time, + n_size=5) @pytest.fixture(scope="module") def driver_df(ingest_time): - return basic_dataframe( - entities=["driver_id"], - features=["rating", "cost"], - ingest_time=ingest_time, - n_size=5, - ) + return basic_dataframe(entities=["driver_id"], + features=["rating", "cost"], + ingest_time=ingest_time, + n_size=5) def test_version_returns_results(client): version_info = client.version() - assert not version_info["core"] is "not configured" - assert not version_info["serving"] is "not configured" + assert not version_info['core'] is 'not configured' + assert not version_info['serving'] is 'not configured' @pytest.mark.timeout(45) @pytest.mark.run(order=10) def test_basic_register_feature_set_success(client): # Register feature set without project - cust_trans_fs_expected = FeatureSet.from_yaml( - f"{DIR_PATH}/basic/cust_trans_fs.yaml" - ) + cust_trans_fs_expected = FeatureSet.from_yaml(f"{DIR_PATH}/basic/cust_trans_fs.yaml") driver_fs_expected = FeatureSet.from_yaml(f"{DIR_PATH}/basic/driver_fs.yaml") client.apply(cust_trans_fs_expected) client.apply(driver_fs_expected) @@ -123,14 +119,11 @@ def test_basic_register_feature_set_success(client): assert driver_fs_actual == driver_fs_expected # Register feature set with project - cust_trans_fs_expected = FeatureSet.from_yaml( - f"{DIR_PATH}/basic/cust_trans_fs.yaml" - ) + cust_trans_fs_expected = FeatureSet.from_yaml(f"{DIR_PATH}/basic/cust_trans_fs.yaml") client.set_project(PROJECT_NAME) client.apply(cust_trans_fs_expected) - cust_trans_fs_actual = client.get_feature_set( - "customer_transactions", project=PROJECT_NAME - ) + cust_trans_fs_actual = client.get_feature_set("customer_transactions", + project=PROJECT_NAME) assert cust_trans_fs_actual == cust_trans_fs_expected # Register feature set with labels @@ -179,21 +172,27 @@ def test_basic_retrieve_online_success(client, cust_trans_df): ) ], # Test retrieve with different variations of the string feature refs - feature_refs=["daily_transactions", "total_transactions",], + feature_refs=[ + "daily_transactions", + "total_transactions", + ] ) # type: GetOnlineFeaturesResponse if response is None: continue returned_daily_transactions = float( - response.field_values[0].fields["daily_transactions"].float_val + response.field_values[0] + .fields["daily_transactions"] + .float_val ) - sent_daily_transactions = float(cust_trans_df.iloc[0]["daily_transactions"]) + sent_daily_transactions = float( + cust_trans_df.iloc[0]["daily_transactions"]) if math.isclose( - sent_daily_transactions, - returned_daily_transactions, - abs_tol=FLOAT_TOLERANCE, + sent_daily_transactions, + returned_daily_transactions, + abs_tol=FLOAT_TOLERANCE, ): break @@ -218,7 +217,9 @@ def test_basic_retrieve_online_multiple_featureset(client, cust_trans_df, driver "customer_id": Value( int64_val=cust_trans_df.iloc[0]["customer_id"] ), - "driver_id": Value(int64_val=driver_df.iloc[0]["driver_id"]), + "driver_id": Value( + int64_val=driver_df.iloc[0]["driver_id"] + ) } ) ], @@ -230,7 +231,9 @@ def test_basic_retrieve_online_multiple_featureset(client, cust_trans_df, driver def check_response(ingest_df, response, feature_ref): returned_value = float( - response.field_values[0].fields[feature_ref].float_val + response.field_values[0] + .fields[feature_ref] + .float_val ) feature_ref_splits = feature_ref.split(":") if len(feature_ref_splits) == 1: @@ -238,13 +241,16 @@ def check_response(ingest_df, response, feature_ref): else: _, feature_name = feature_ref_splits - sent_value = float(ingest_df.iloc[0][feature_name]) + sent_value = float( + ingest_df.iloc[0][feature_name]) - return math.isclose(sent_value, returned_value, abs_tol=FLOAT_TOLERANCE,) + return math.isclose( + sent_value, + returned_value, + abs_tol=FLOAT_TOLERANCE, + ) - if all( - [check_response(df, response, ref) for ref, df in feature_ref_df_mapping] - ): + if all([check_response(df, response, ref) for ref, df in feature_ref_df_mapping]): break @@ -254,12 +260,9 @@ def test_basic_ingest_jobs(client): # list ingestion jobs given featureset cust_trans_fs = client.get_feature_set(name="customer_transactions") ingest_jobs = client.list_ingest_jobs( - feature_set_ref=FeatureSetRef.from_feature_set(cust_trans_fs) - ) + feature_set_ref=FeatureSetRef.from_feature_set(cust_trans_fs)) # filter ingestion jobs to only those that are running - ingest_jobs = [ - job for job in ingest_jobs if job.status == IngestionJobStatus.RUNNING - ] + ingest_jobs = [job for job in ingest_jobs if job.status == IngestionJobStatus.RUNNING] assert len(ingest_jobs) >= 1 for ingest_job in ingest_jobs: @@ -274,16 +277,18 @@ def test_basic_ingest_jobs(client): assert ingest_job.status == IngestionJobStatus.ABORTED -@pytest.fixture(scope="module") +@pytest.fixture(scope='module') def all_types_dataframe(): return pd.DataFrame( { - "datetime": [datetime.utcnow().replace(tzinfo=pytz.utc) for _ in range(3)], + "datetime": [datetime.utcnow().replace(tzinfo=pytz.utc) for _ in + range(3)], "user_id": [1001, 1002, 1003], "int32_feature": [np.int32(1), np.int32(2), np.int32(3)], "int64_feature": [np.int64(1), np.int64(2), np.int64(3)], "float_feature": [np.float(0.1), np.float(0.2), np.float(0.3)], - "double_feature": [np.float64(0.1), np.float64(0.2), np.float64(0.3)], + "double_feature": [np.float64(0.1), np.float64(0.2), + np.float64(0.3)], "string_feature": ["one", "two", "three"], "bytes_feature": [b"one", b"two", b"three"], "bool_feature": [True, False, False], @@ -345,7 +350,8 @@ def test_all_types_register_feature_set_success(client): Feature(name="float_list_feature", dtype=ValueType.FLOAT_LIST), Feature(name="int64_list_feature", dtype=ValueType.INT64_LIST), Feature(name="int32_list_feature", dtype=ValueType.INT32_LIST), - Feature(name="string_list_feature", dtype=ValueType.STRING_LIST), + Feature(name="string_list_feature", + dtype=ValueType.STRING_LIST), Feature(name="bytes_list_feature", dtype=ValueType.BYTES_LIST), ], max_age=Duration(seconds=3600), @@ -391,11 +397,8 @@ def test_all_types_retrieve_online_success(client, all_types_dataframe): response = client.get_online_features( entity_rows=[ GetOnlineFeaturesRequest.EntityRow( - fields={ - "user_id": Value( - int64_val=all_types_dataframe.iloc[0]["user_id"] - ) - } + fields={"user_id": Value( + int64_val=all_types_dataframe.iloc[0]["user_id"])} ) ], feature_refs=[ @@ -419,13 +422,15 @@ def test_all_types_retrieve_online_success(client, all_types_dataframe): continue returned_float_list = ( - response.field_values[0].fields["float_list_feature"].float_list_val.val + response.field_values[0] + .fields["float_list_feature"] + .float_list_val.val ) sent_float_list = all_types_dataframe.iloc[0]["float_list_feature"] if math.isclose( - returned_float_list[0], sent_float_list[0], abs_tol=FLOAT_TOLERANCE + returned_float_list[0], sent_float_list[0], abs_tol=FLOAT_TOLERANCE ): break @@ -436,12 +441,9 @@ def test_all_types_ingest_jobs(client, all_types_dataframe): # list ingestion jobs given featureset all_types_fs = client.get_feature_set(name="all_types") ingest_jobs = client.list_ingest_jobs( - feature_set_ref=FeatureSetRef.from_feature_set(all_types_fs) - ) + feature_set_ref=FeatureSetRef.from_feature_set(all_types_fs)) # filter ingestion jobs to only those that are running - ingest_jobs = [ - job for job in ingest_jobs if job.status == IngestionJobStatus.RUNNING - ] + ingest_jobs = [job for job in ingest_jobs if job.status == IngestionJobStatus.RUNNING] assert len(ingest_jobs) >= 1 for ingest_job in ingest_jobs: @@ -456,14 +458,15 @@ def test_all_types_ingest_jobs(client, all_types_dataframe): assert ingest_job.status == IngestionJobStatus.ABORTED -@pytest.fixture(scope="module") +@pytest.fixture(scope='module') def large_volume_dataframe(): ROW_COUNT = 100000 offset = random.randint(1000000, 10000000) # ensure a unique key space customer_data = pd.DataFrame( { "datetime": [ - datetime.utcnow().replace(tzinfo=pytz.utc) for _ in range(ROW_COUNT) + datetime.utcnow().replace(tzinfo=pytz.utc) for _ in + range(ROW_COUNT) ], "customer_id": [offset + inc for inc in range(ROW_COUNT)], "daily_transactions_large": [np.random.rand() for _ in range(ROW_COUNT)], @@ -477,8 +480,7 @@ def large_volume_dataframe(): @pytest.mark.run(order=30) def test_large_volume_register_feature_set_success(client): cust_trans_fs_expected = FeatureSet.from_yaml( - f"{DIR_PATH}/large_volume/cust_trans_large_fs.yaml" - ) + f"{DIR_PATH}/large_volume/cust_trans_large_fs.yaml") # Register feature set client.apply(cust_trans_fs_expected) @@ -486,7 +488,8 @@ def test_large_volume_register_feature_set_success(client): # Feast Core needs some time to fully commit the FeatureSet applied # when there is no existing job yet for the Featureset time.sleep(10) - cust_trans_fs_actual = client.get_feature_set(name="customer_transactions_large") + cust_trans_fs_actual = client.get_feature_set( + name="customer_transactions_large") assert cust_trans_fs_actual == cust_trans_fs_expected @@ -521,73 +524,80 @@ def test_large_volume_retrieve_online_success(client, large_volume_dataframe): GetOnlineFeaturesRequest.EntityRow( fields={ "customer_id": Value( - int64_val=large_volume_dataframe.iloc[0]["customer_id"] + int64_val=large_volume_dataframe.iloc[0][ + "customer_id"] ) } ) ], - feature_refs=["daily_transactions_large", "total_transactions_large",], + feature_refs=[ + "daily_transactions_large", + "total_transactions_large", + ], ) # type: GetOnlineFeaturesResponse if response is None: continue returned_daily_transactions = float( - response.field_values[0].fields["daily_transactions_large"].float_val + response.field_values[0] + .fields["daily_transactions_large"] + .float_val ) sent_daily_transactions = float( - large_volume_dataframe.iloc[0]["daily_transactions_large"] - ) + large_volume_dataframe.iloc[0]["daily_transactions_large"]) if math.isclose( - sent_daily_transactions, - returned_daily_transactions, - abs_tol=FLOAT_TOLERANCE, + sent_daily_transactions, + returned_daily_transactions, + abs_tol=FLOAT_TOLERANCE, ): break -@pytest.fixture(scope="module") +@pytest.fixture(scope='module') def all_types_parquet_file(): COUNT = 20000 df = pd.DataFrame( { "datetime": [datetime.utcnow() for _ in range(COUNT)], - "customer_id": [np.int32(random.randint(0, 10000)) for _ in range(COUNT)], - "int32_feature_parquet": [ - np.int32(random.randint(0, 10000)) for _ in range(COUNT) - ], - "int64_feature_parquet": [ - np.int64(random.randint(0, 10000)) for _ in range(COUNT) - ], + "customer_id": [np.int32(random.randint(0, 10000)) for _ in + range(COUNT)], + "int32_feature_parquet": [np.int32(random.randint(0, 10000)) for _ in + range(COUNT)], + "int64_feature_parquet": [np.int64(random.randint(0, 10000)) for _ in + range(COUNT)], "float_feature_parquet": [np.float(random.random()) for _ in range(COUNT)], - "double_feature_parquet": [ - np.float64(random.random()) for _ in range(COUNT) - ], - "string_feature_parquet": [ - "one" + str(random.random()) for _ in range(COUNT) - ], + "double_feature_parquet": [np.float64(random.random()) for _ in + range(COUNT)], + "string_feature_parquet": ["one" + str(random.random()) for _ in + range(COUNT)], "bytes_feature_parquet": [b"one" for _ in range(COUNT)], "int32_list_feature_parquet": [ np.array([1, 2, 3, random.randint(0, 10000)], dtype=np.int32) - for _ in range(COUNT) + for _ + in range(COUNT) ], "int64_list_feature_parquet": [ np.array([1, random.randint(0, 10000), 3, 4], dtype=np.int64) - for _ in range(COUNT) + for _ + in range(COUNT) ], "float_list_feature_parquet": [ - np.array([1.1, 1.2, 1.3, random.random()], dtype=np.float32) - for _ in range(COUNT) + np.array([1.1, 1.2, 1.3, random.random()], dtype=np.float32) for + _ + in range(COUNT) ], "double_list_feature_parquet": [ - np.array([1.1, 1.2, 1.3, random.random()], dtype=np.float64) - for _ in range(COUNT) + np.array([1.1, 1.2, 1.3, random.random()], dtype=np.float64) for + _ + in range(COUNT) ], "string_list_feature_parquet": [ - np.array(["one", "two" + str(random.random()), "three"]) - for _ in range(COUNT) + np.array(["one", "two" + str(random.random()), "three"]) for _ + in + range(COUNT) ], "bytes_list_feature_parquet": [ np.array([b"one", b"two", b"three"]) for _ in range(COUNT) @@ -598,7 +608,7 @@ def all_types_parquet_file(): # TODO: Boolean list is not being tested. # https://github.com/feast-dev/feast/issues/341 - file_path = os.path.join(tempfile.mkdtemp(), "all_types.parquet") + file_path = os.path.join(tempfile.mkdtemp(), 'all_types.parquet') df.to_parquet(file_path, allow_truncated_timestamps=True) return file_path @@ -608,8 +618,7 @@ def all_types_parquet_file(): def test_all_types_parquet_register_feature_set_success(client): # Load feature set from file all_types_parquet_expected = FeatureSet.from_yaml( - f"{DIR_PATH}/all_types_parquet/all_types_parquet.yaml" - ) + f"{DIR_PATH}/all_types_parquet/all_types_parquet.yaml") # Register feature set client.apply(all_types_parquet_expected) @@ -633,7 +642,8 @@ def test_all_types_parquet_register_feature_set_success(client): @pytest.mark.timeout(600) @pytest.mark.run(order=41) -def test_all_types_infer_register_ingest_file_success(client, all_types_parquet_file): +def test_all_types_infer_register_ingest_file_success(client, + all_types_parquet_file): # Get feature set all_types_fs = client.get_feature_set(name="all_types_parquet") @@ -650,14 +660,14 @@ class TestsBasedOnGrpc: @pytest.fixture(scope="module") def core_service_stub(self, core_url): if core_url.endswith(":443"): - core_channel = grpc.secure_channel(core_url, grpc.ssl_channel_credentials()) + core_channel = grpc.secure_channel( + core_url, grpc.ssl_channel_credentials() + ) else: core_channel = grpc.insecure_channel(core_url) try: - grpc.channel_ready_future(core_channel).result( - timeout=self.GRPC_CONNECTION_TIMEOUT - ) + grpc.channel_ready_future(core_channel).result(timeout=self.GRPC_CONNECTION_TIMEOUT) except grpc.FutureTimeoutError: raise ConnectionError( f"Connection timed out while attempting to connect to Feast " From de542292152eea88aecbcabb81183fe1609644f6 Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 12:57:29 +0800 Subject: [PATCH 08/17] Refactor FeatureSet labels filter logic --- .../java/feast/core/model/FeatureSet.java | 11 ++++++ .../java/feast/core/service/SpecService.java | 35 +++++++++++-------- 2 files changed, 32 insertions(+), 14 deletions(-) diff --git a/core/src/main/java/feast/core/model/FeatureSet.java b/core/src/main/java/feast/core/model/FeatureSet.java index f7b2dc7cd49..72a57b0cb72 100644 --- a/core/src/main/java/feast/core/model/FeatureSet.java +++ b/core/src/main/java/feast/core/model/FeatureSet.java @@ -121,6 +121,17 @@ private String getProjectName() { } } + public boolean hasAllLabels(Map labels) throws InvalidProtocolBufferException { + Map featureSetLabelsMap = this.toProto().getSpec().getLabelsMap(); + for (String key : labels.keySet()) { + if (!featureSetLabelsMap.containsKey(key) + || !featureSetLabelsMap.get(key).equals(labels.get(key))) { + return false; + } + } + return true; + } + public void setProject(Project project) { this.project = project; } diff --git a/core/src/main/java/feast/core/service/SpecService.java b/core/src/main/java/feast/core/service/SpecService.java index d8808a0e993..8009aa76cdb 100644 --- a/core/src/main/java/feast/core/service/SpecService.java +++ b/core/src/main/java/feast/core/service/SpecService.java @@ -48,6 +48,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -115,9 +116,9 @@ public GetFeatureSetResponse getFeatureSet(GetFeatureSetRequest request) } /** - * Return a list of feature sets matching the feature set name and project provided in the filter. - * All fields are requried. Use '*' for all arguments in order to return all feature sets in all - * projects. + * Return a list of feature sets matching the feature set name, project and labels provided in the + * filter. All fields are required. Use '*' in feature set name and project, and empty map in + * labels in order to return all feature sets in all projects. * *

Project name can be explicitly provided, or an asterisk can be provided to match all * projects. It is not possible to provide a combination of asterisks/wildcards and text. If the @@ -127,6 +128,9 @@ public GetFeatureSetResponse getFeatureSet(GetFeatureSetRequest request) * sets will be returned. Regex is not supported. Explicitly defining a feature set name is not * possible if a project name is not set explicitly * + *

The labels in the filter accepts a map. All feature sets which contain every provided label + * will be returned. + * * @param filter filter containing the desired featureSet name * @return ListFeatureSetsResponse with list of featureSets found matching the filter */ @@ -190,18 +194,21 @@ public ListFeatureSetsResponse listFeatureSets(ListFeatureSetsRequest.Filter fil ListFeatureSetsResponse.Builder response = ListFeatureSetsResponse.newBuilder(); if (featureSets.size() > 0) { + featureSets = + featureSets.stream() + .filter( + featureSet -> { + boolean validFeatureSet = false; + try { + validFeatureSet = featureSet.hasAllLabels(labels); + } catch (InvalidProtocolBufferException e) { + e.printStackTrace(); + } + return validFeatureSet; + }) + .collect(Collectors.toList()); for (FeatureSet featureSet : featureSets) { - if (labels != null && !labels.isEmpty()) { - Map tempLabels = featureSet.toProto().getSpec().getLabelsMap(); - for (Map.Entry entry : labels.entrySet()) { - if (tempLabels.containsKey(entry.getKey()) - && tempLabels.get(entry.getKey()).equals(entry.getValue())) { - response.addFeatureSets(featureSet.toProto()); - } - } - } else { - response.addFeatureSets(featureSet.toProto()); - } + response.addFeatureSets(featureSet.toProto()); } } From d3574347d38bf7c9255ef6739e0f62580444059e Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 12:57:53 +0800 Subject: [PATCH 09/17] Set default labels to empty dict --- sdk/python/feast/client.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/sdk/python/feast/client.py b/sdk/python/feast/client.py index 5ea39fa9884..2ec90e434db 100644 --- a/sdk/python/feast/client.py +++ b/sdk/python/feast/client.py @@ -425,7 +425,7 @@ def _apply_feature_set(self, feature_set: FeatureSet): feature_set._update_from_feature_set(applied_fs) def list_feature_sets( - self, project: str = None, name: str = None, labels: Dict[str, str] = None + self, project: str = None, name: str = None, labels: Dict[str, str] = dict() ) -> List[FeatureSet]: """ Retrieve a list of feature sets from Feast Core @@ -448,9 +448,6 @@ def list_feature_sets( if name is None: name = "*" - if labels is None: - labels = {} - filter = ListFeatureSetsRequest.Filter( project=project, feature_set_name=name, labels=labels ) From f4c858368a6602667dff8a95d0a8c440eb85e5fb Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 12:58:54 +0800 Subject: [PATCH 10/17] Refactor feature_set_list cli --- sdk/python/feast/cli.py | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/sdk/python/feast/cli.py b/sdk/python/feast/cli.py index a002ca74df4..d0c3adaf9b3 100644 --- a/sdk/python/feast/cli.py +++ b/sdk/python/feast/cli.py @@ -120,6 +120,18 @@ def feature_set(): pass +def _get_labels_dict(label_str: str): + labels_dict = {} + labels_kv = label_str.split(",") + if label_str == "": + return labels_dict + if len(labels_kv) % 2 == 1: + return None + for k, v in zip(labels_kv[0::2], labels_kv[1::2]): + labels_dict[k] = v + return labels_dict + + @feature_set.command(name="list") @click.option( "--project", @@ -131,7 +143,7 @@ def feature_set(): @click.option( "--name", "-n", - help="Name that feature set belongs to, may include * to match multiple feature sets eg. driver_*", + help="Filters feature sets by name. Wildcards may be included to match multiple feature sets", type=click.STRING, default="*", ) @@ -148,16 +160,9 @@ def feature_set_list(project: str, name: str, labels: str): """ feast_client = Client() # type: Client - labels_kv = labels.split(",") - labels_dict = {} - if labels == "": - pass - elif len(labels_kv) % 2 == 0: - for k, v in zip(labels_kv[0::2], labels_kv[1::2]): - labels_dict[k] = v - else: - print("Uneven key-value label pairs were entered") - return + labels_dict = _get_labels_dict(labels) + if labels_dict is None: + raise ValueError("Uneven key-value label pairs were entered") table = [] for fs in feast_client.list_feature_sets( From e10199a4d435f3f608570986d569e184e9630dfb Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 12:59:29 +0800 Subject: [PATCH 11/17] Refactor brittle test --- tests/e2e/redis/basic-ingest-redis-serving.py | 27 ++++++++++++++----- tests/e2e/redis/basic/driver_labels_fs.yaml | 14 ---------- 2 files changed, 21 insertions(+), 20 deletions(-) delete mode 100644 tests/e2e/redis/basic/driver_labels_fs.yaml diff --git a/tests/e2e/redis/basic-ingest-redis-serving.py b/tests/e2e/redis/basic-ingest-redis-serving.py index c893dca71ff..1424936f0c8 100644 --- a/tests/e2e/redis/basic-ingest-redis-serving.py +++ b/tests/e2e/redis/basic-ingest-redis-serving.py @@ -127,17 +127,32 @@ def test_basic_register_feature_set_success(client): assert cust_trans_fs_actual == cust_trans_fs_expected # Register feature set with labels - driver_fs_expected = FeatureSet.from_yaml(f"{DIR_PATH}/basic/driver_fs.yaml") - driver_fs_labeled_expected = FeatureSet.from_yaml( - f"{DIR_PATH}/basic/driver_labels_fs.yaml" + driver_unlabelled_fs = FeatureSet( + "driver_unlabelled", + features=[ + Feature("rating", ValueType.FLOAT), + Feature("cost", ValueType.FLOAT) + ], + entities=[Entity("entity_id", ValueType.INT64)], + max_age=Duration(seconds=100) + ) + driver_labeled_fs_expected = FeatureSet( + "driver_labeled", + features=[ + Feature("rating", ValueType.FLOAT), + Feature("cost", ValueType.FLOAT) + ], + entities=[Entity("entity_id", ValueType.INT64)], + max_age=Duration(seconds=100), + labels={"key1":"val1"} ) client.set_project(PROJECT_NAME) - client.apply(driver_fs_expected) - client.apply(driver_fs_labeled_expected) + client.apply(driver_unlabelled_fs) + client.apply(driver_labeled_fs_expected) driver_fs_actual = client.list_feature_sets( project=PROJECT_NAME, labels={"key1": "val1"} )[0] - assert driver_fs_actual == driver_fs_labeled_expected + assert driver_fs_actual == driver_labeled_fs_expected # reset client's project for other tests client.set_project() diff --git a/tests/e2e/redis/basic/driver_labels_fs.yaml b/tests/e2e/redis/basic/driver_labels_fs.yaml deleted file mode 100644 index 70aef796cc5..00000000000 --- a/tests/e2e/redis/basic/driver_labels_fs.yaml +++ /dev/null @@ -1,14 +0,0 @@ -kind: feature_set -spec: - name: driver_labeled - entities: - - name: driver_id - valueType: INT64 - features: - - name: rating - valueType: FLOAT - - name: cost - valueType: FLOAT - maxAge: 3600s - labels: - key1: val1 From ec8313758605e4947a059bc4f54c2460a8f0577e Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 14:10:36 +0800 Subject: [PATCH 12/17] Implement helper function and not convert to proto --- core/src/main/java/feast/core/model/FeatureSet.java | 8 ++++++-- .../src/main/java/feast/core/service/SpecService.java | 11 +---------- 2 files changed, 7 insertions(+), 12 deletions(-) diff --git a/core/src/main/java/feast/core/model/FeatureSet.java b/core/src/main/java/feast/core/model/FeatureSet.java index 72a57b0cb72..74b4f5eaa68 100644 --- a/core/src/main/java/feast/core/model/FeatureSet.java +++ b/core/src/main/java/feast/core/model/FeatureSet.java @@ -121,8 +121,8 @@ private String getProjectName() { } } - public boolean hasAllLabels(Map labels) throws InvalidProtocolBufferException { - Map featureSetLabelsMap = this.toProto().getSpec().getLabelsMap(); + public boolean hasAllLabels(Map labels) { + Map featureSetLabelsMap = this.getLabelsMap(); for (String key : labels.keySet()) { if (!featureSetLabelsMap.containsKey(key) || !featureSetLabelsMap.get(key).equals(labels.get(key))) { @@ -273,6 +273,10 @@ public FeatureSetProto.FeatureSet toProto() throws InvalidProtocolBufferExceptio return FeatureSetProto.FeatureSet.newBuilder().setMeta(meta).setSpec(spec).build(); } + public Map getLabelsMap() { + return TypeConversion.convertJsonStringToMap(this.labels); + } + @Override public int hashCode() { HashCodeBuilder hcb = new HashCodeBuilder(); diff --git a/core/src/main/java/feast/core/service/SpecService.java b/core/src/main/java/feast/core/service/SpecService.java index 8009aa76cdb..1351c8c7dec 100644 --- a/core/src/main/java/feast/core/service/SpecService.java +++ b/core/src/main/java/feast/core/service/SpecService.java @@ -196,16 +196,7 @@ public ListFeatureSetsResponse listFeatureSets(ListFeatureSetsRequest.Filter fil if (featureSets.size() > 0) { featureSets = featureSets.stream() - .filter( - featureSet -> { - boolean validFeatureSet = false; - try { - validFeatureSet = featureSet.hasAllLabels(labels); - } catch (InvalidProtocolBufferException e) { - e.printStackTrace(); - } - return validFeatureSet; - }) + .filter(featureSet -> featureSet.hasAllLabels(labels)) .collect(Collectors.toList()); for (FeatureSet featureSet : featureSets) { response.addFeatureSets(featureSet.toProto()); From 13e36cca18d7ba9be58d61b2bacb84b63c4aab2f Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 14:10:53 +0800 Subject: [PATCH 13/17] Update cli --- sdk/python/feast/cli.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/sdk/python/feast/cli.py b/sdk/python/feast/cli.py index d0c3adaf9b3..ae4afb0745b 100644 --- a/sdk/python/feast/cli.py +++ b/sdk/python/feast/cli.py @@ -126,7 +126,7 @@ def _get_labels_dict(label_str: str): if label_str == "": return labels_dict if len(labels_kv) % 2 == 1: - return None + raise ValueError("Uneven key-value label pairs were entered") for k, v in zip(labels_kv[0::2], labels_kv[1::2]): labels_dict[k] = v return labels_dict @@ -161,8 +161,6 @@ def feature_set_list(project: str, name: str, labels: str): feast_client = Client() # type: Client labels_dict = _get_labels_dict(labels) - if labels_dict is None: - raise ValueError("Uneven key-value label pairs were entered") table = [] for fs in feast_client.list_feature_sets( From 70c38110da807ec7edbb297b6c19800fead46c4d Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 16:43:34 +0800 Subject: [PATCH 14/17] Add docstrings and improve variable naming --- core/src/main/java/feast/core/model/FeatureSet.java | 12 +++++++++--- .../main/java/feast/core/service/SpecService.java | 4 ++-- sdk/python/feast/cli.py | 5 ++++- 3 files changed, 15 insertions(+), 6 deletions(-) diff --git a/core/src/main/java/feast/core/model/FeatureSet.java b/core/src/main/java/feast/core/model/FeatureSet.java index 74b4f5eaa68..341af3dcdfc 100644 --- a/core/src/main/java/feast/core/model/FeatureSet.java +++ b/core/src/main/java/feast/core/model/FeatureSet.java @@ -121,11 +121,17 @@ private String getProjectName() { } } - public boolean hasAllLabels(Map labels) { + /** + * Return a boolean to facilitate streaming elements on the basis of given predicate. + * + * @param labelsFilter labels contain key-value mapping for labels attached to the FeatureSet + * @return boolean True if FeatureSet contains all labels in the labelsFilter + */ + public boolean hasAllLabels(Map labelsFilter) { Map featureSetLabelsMap = this.getLabelsMap(); - for (String key : labels.keySet()) { + for (String key : labelsFilter.keySet()) { if (!featureSetLabelsMap.containsKey(key) - || !featureSetLabelsMap.get(key).equals(labels.get(key))) { + || !featureSetLabelsMap.get(key).equals(labelsFilter.get(key))) { return false; } } diff --git a/core/src/main/java/feast/core/service/SpecService.java b/core/src/main/java/feast/core/service/SpecService.java index 1351c8c7dec..ec295c2301d 100644 --- a/core/src/main/java/feast/core/service/SpecService.java +++ b/core/src/main/java/feast/core/service/SpecService.java @@ -138,7 +138,7 @@ public ListFeatureSetsResponse listFeatureSets(ListFeatureSetsRequest.Filter fil throws InvalidProtocolBufferException { String name = filter.getFeatureSetName(); String project = filter.getProject(); - Map labels = filter.getLabelsMap(); + Map labelsFilter = filter.getLabelsMap(); if (name.isEmpty()) { throw new IllegalArgumentException( @@ -196,7 +196,7 @@ public ListFeatureSetsResponse listFeatureSets(ListFeatureSetsRequest.Filter fil if (featureSets.size() > 0) { featureSets = featureSets.stream() - .filter(featureSet -> featureSet.hasAllLabels(labels)) + .filter(featureSet -> featureSet.hasAllLabels(labelsFilter)) .collect(Collectors.toList()); for (FeatureSet featureSet : featureSets) { response.addFeatureSets(featureSet.toProto()); diff --git a/sdk/python/feast/cli.py b/sdk/python/feast/cli.py index ae4afb0745b..34a3f600e15 100644 --- a/sdk/python/feast/cli.py +++ b/sdk/python/feast/cli.py @@ -121,6 +121,9 @@ def feature_set(): def _get_labels_dict(label_str: str): + """ + Converts CLI input labels string to dictionary format if provided string is valid. + """ labels_dict = {} labels_kv = label_str.split(",") if label_str == "": @@ -143,7 +146,7 @@ def _get_labels_dict(label_str: str): @click.option( "--name", "-n", - help="Filters feature sets by name. Wildcards may be included to match multiple feature sets", + help="Filters feature sets by name. Wildcards (*) may be included to match multiple feature sets", type=click.STRING, default="*", ) From 17679d4853f329caaebdb10d63cb35f33a6aa3a3 Mon Sep 17 00:00:00 2001 From: Terence Date: Thu, 11 Jun 2020 18:44:36 +0800 Subject: [PATCH 15/17] Use getter method instead --- core/src/main/java/feast/core/model/FeatureSet.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/java/feast/core/model/FeatureSet.java b/core/src/main/java/feast/core/model/FeatureSet.java index 341af3dcdfc..54bfcb08064 100644 --- a/core/src/main/java/feast/core/model/FeatureSet.java +++ b/core/src/main/java/feast/core/model/FeatureSet.java @@ -280,7 +280,7 @@ public FeatureSetProto.FeatureSet toProto() throws InvalidProtocolBufferExceptio } public Map getLabelsMap() { - return TypeConversion.convertJsonStringToMap(this.labels); + return TypeConversion.convertJsonStringToMap(this.getLabels()); } @Override From 03e62ee2278777b4e7269452fa2acd79d08c11d8 Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 17 Jun 2020 17:46:57 +0800 Subject: [PATCH 16/17] Fix java lint --- core/src/main/java/feast/core/service/SpecService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/java/feast/core/service/SpecService.java b/core/src/main/java/feast/core/service/SpecService.java index 035799b938e..a4f1202858d 100644 --- a/core/src/main/java/feast/core/service/SpecService.java +++ b/core/src/main/java/feast/core/service/SpecService.java @@ -47,8 +47,8 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; -import java.util.stream.Collectors; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.beans.factory.annotation.Autowired; From e61be4aa0a8db312fa86de8badb5c848c96d5a8b Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 17 Jun 2020 18:11:26 +0800 Subject: [PATCH 17/17] Add description for labels filter proto --- protos/feast/core/CoreService.proto | 2 ++ 1 file changed, 2 insertions(+) diff --git a/protos/feast/core/CoreService.proto b/protos/feast/core/CoreService.proto index 6a9bfd8ff28..a6ac313b2a5 100644 --- a/protos/feast/core/CoreService.proto +++ b/protos/feast/core/CoreService.proto @@ -141,6 +141,8 @@ message ListFeatureSetsRequest { // - my-feature-set-6 can be used to select a single feature set string feature_set_name = 1; + // User defined metadata for feature set. + // Feature sets with all matching labels will be returned. map labels = 4; } }