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
Next Next commit
Make _write_to_offline_store a public method 
Signed-off-by: Felix Wang <wangfelix98@gmail.com>
  • Loading branch information
@felixwang9817
felixwang9817 committed Jun 30, 2022
commit 997003269244b0f14e90991b12b2ba1d26379ad3
4 changes: 2 additions & 2 deletions sdk/python/feast/feature_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1384,7 +1384,7 @@ def push(
fv.name, df, allow_registry_cache=allow_registry_cache
)
if to == PushMode.OFFLINE or to == PushMode.ONLINE_AND_OFFLINE:
self._write_to_offline_store(
self.write_to_offline_store(
fv.name, df, allow_registry_cache=allow_registry_cache
)

Expand Down Expand Up @@ -1416,7 +1416,7 @@ def write_to_online_store(
provider.ingest_df(feature_view, entities, df)

@log_exceptions_and_usage
def _write_to_offline_store(
def write_to_offline_store(
self,
feature_view_name: str,
df: pd.DataFrame,
Expand Down
2 changes: 1 addition & 1 deletion sdk/python/feast/infra/contrib/spark_kafka_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ def batch_write(row: DataFrame, batch_id: int):
if to == PushMode.ONLINE or to == PushMode.ONLINE_AND_OFFLINE:
self.fs.write_to_online_store(self.sfv.name, rows)
if to == PushMode.OFFLINE or to == PushMode.ONLINE_AND_OFFLINE:
self.fs._write_to_offline_store(self.sfv.name, rows)
self.fs.write_to_offline_store(self.sfv.name, rows)

query = (
df.writeStream.outputMode("update")
Expand Down