Skip to content

feat: Add column reordering to write_to_offline_store #2876

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
feast-ci-bot merged 11 commits into from
Jun 30, 2022
Merged

feat: Add column reordering to write_to_offline_store #2876

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
0a152ee
Add feature extraction logic to batch writer
felixwang9817 Jun 23, 2022
0ce6433
Enable StreamProcessor to write to both online and offline stores
felixwang9817 Jun 27, 2022
a47b39b
Fix incorrect columns error message
felixwang9817 Jun 28, 2022
791322b
Reorder columns in _write_to_offline_store
felixwang9817 Jun 28, 2022
9970032
Make _write_to_offline_store a public method
felixwang9817 Jun 28, 2022
d1ce38f
Import FeatureStore correctly
felixwang9817 Jun 29, 2022
5ff276c
Remove defaults for `processing_time` and `query_timeout`
felixwang9817 Jun 29, 2022
838363a
Clean up `test_offline_write.py`
felixwang9817 Jun 29, 2022
b090a1d
Do not do any custom logic for double underscore columns
felixwang9817 Jun 29, 2022
b75883a
Lint
felixwang9817 Jun 29, 2022
e54ea6e
Switch entity values for all tests using push sources to not affect o…
felixwang9817 Jun 30, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Switch entity values for all tests using push sources to not affect o… 
…ther tests

Signed-off-by: Felix Wang <wangfelix98@gmail.com>
  • Loading branch information
@felixwang9817
felixwang9817 committed Jun 30, 2022
commit e54ea6eb14463f28bbf885327f7c217710b3f672
11 changes: 7 additions & 4 deletions sdk/python/tests/integration/e2e/test_python_feature_server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,13 +63,16 @@ def test_get_online_features(python_fs_client):
@pytest.mark.integration
@pytest.mark.universal_online_stores
def test_push(python_fs_client):
initial_temp = get_temperatures(python_fs_client, location_ids=[1])[0]
# TODO(felixwang9817): Note that we choose an entity value of 102 here since it is not included
# in the existing range of entity values (1-49). This allows us to push data for this test
# without affecting other tests. This decision is tech debt, and should be resolved by finding a
# better way to isolate data sources across tests.
json_data = json.dumps(
{
"push_source_name": "location_stats_push_source",
"df": {
"location_id": [1],
"temperature": [initial_temp * 100],
"location_id": [102],
"temperature": [4],
"event_timestamp": [str(datetime.utcnow())],
"created": [str(datetime.utcnow())],
},
Expand All @@ -79,7 +82,7 @@ def test_push(python_fs_client):

# Check new pushed temperature is fetched
assert response.status_code == 200
assert get_temperatures(python_fs_client, location_ids=[1]) == [initial_temp * 100]
assert get_temperatures(python_fs_client, location_ids=[102]) == [4]


def get_temperatures(client, location_ids: List[int]):
Expand Down
8 changes: 6 additions & 2 deletions sdk/python/tests/integration/offline_store/test_push_offline_retrieval.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,17 +26,21 @@ def test_push_features_and_read_from_offline_store(environment, universal_data_s
now = pd.Timestamp(datetime.datetime.utcnow()).round("ms")

store.apply([driver(), customer(), location(), *feature_views.values()])
entity_df = pd.DataFrame.from_dict({"location_id": [1], "event_timestamp": [now]})
entity_df = pd.DataFrame.from_dict({"location_id": [100], "event_timestamp": [now]})

before_df = store.get_historical_features(
entity_df=entity_df,
features=["pushable_location_stats:temperature"],
full_feature_names=False,
).to_df()

# TODO(felixwang9817): Note that we choose an entity value of 100 here since it is not included
# in the existing range of entity values (1-49). This allows us to push data for this test
# without affecting other tests. This decision is tech debt, and should be resolved by finding a
# better way to isolate data sources across tests.
data = {
"event_timestamp": [now],
"location_id": [1],
"location_id": [100],
"temperature": [4],
"created": [now],
}
Expand Down
11 changes: 8 additions & 3 deletions sdk/python/tests/integration/online_store/test_push_online_retrieval.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,13 @@ def test_push_features_and_read(environment, universal_data_sources):
feature_views = construct_universal_feature_views(data_sources)

store.apply([driver(), customer(), location(), *feature_views.values()])

# TODO(felixwang9817): Note that we choose an entity value of 101 here since it is not included
# in the existing range of entity values (1-49). This allows us to push data for this test
# without affecting other tests. This decision is tech debt, and should be resolved by finding a
# better way to isolate data sources across tests.
data = {
"location_id": [1],
"location_id": [101],
"temperature": [4],
"event_timestamp": [pd.Timestamp(datetime.datetime.utcnow()).round("ms")],
"created": [pd.Timestamp(datetime.datetime.utcnow()).round("ms")],
Expand All @@ -34,8 +39,8 @@ def test_push_features_and_read(environment, universal_data_sources):

online_resp = store.get_online_features(
features=["pushable_location_stats:temperature"],
entity_rows=[{"location_id": 1}],
entity_rows=[{"location_id": 101}],
)
online_resp_dict = online_resp.to_dict()
assert online_resp_dict["location_id"] == [1]
assert online_resp_dict["location_id"] == [101]
assert online_resp_dict["temperature"] == [4]