Skip to content

chore: Refactoring to fix mypy linting #3947

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

Closed
bushwhackr wants to merge 13 commits into from
Closed

chore: Refactoring to fix mypy linting #3947

Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
68d18d2
allowing follow-imports + fixing on_demand_feature_view.py
bushwhackr Feb 11, 2024
73bba2c
creating base type `SnowflakeStoreConfig` to fix circular imports
bushwhackr Feb 11, 2024
d80acd8
fixing circular imports
bushwhackr Feb 11, 2024
3ecff66
converting create_data_source to take in arbitrary arguments
bushwhackr Feb 11, 2024
39fb45b
lint fix on base function `create_online_store`
bushwhackr Feb 11, 2024
83818e0
lint fix on ingest_stream_feature_view
bushwhackr Feb 11, 2024
2fa14b7
lint fixes in configuration
bushwhackr Feb 11, 2024
a96c266
isort linting
bushwhackr Feb 11, 2024
f0f7fec
mypy checking of all files in sdk/python folder
bushwhackr Feb 11, 2024
28010b7
checking only feast directory for mypy errors
bushwhackr Feb 11, 2024
5dbe13d
allow extras for FeastConfigBaseModel
bushwhackr Feb 13, 2024
004949d
fixing tests
bushwhackr Feb 13, 2024
2696cc3
adding extra field for online store config
bushwhackr Feb 13, 2024
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
lint fix on ingest_stream_feature_view 
Signed-off-by: Chester Ong <chester.ong.ch@gmail.com>
  • Loading branch information
@bushwhackr
bushwhackr committed Feb 11, 2024
commit 83818e065fe3dbe71f14a7379c7a865186d05ddc
10 changes: 3 additions & 7 deletions 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 @@ -68,13 +68,10 @@ def __init__(
# data_source type has been checked to be an instance of KafkaSource.
self.data_source: KafkaSource = self.data_source # type: ignore

def ingest_stream_feature_view(
self, to: PushMode = PushMode.ONLINE
) -> StreamingQuery:
def ingest_stream_feature_view(self, to: PushMode = PushMode.ONLINE):
ingested_stream_df = self._ingest_stream_data()
transformed_df = self._construct_transformation_plan(ingested_stream_df)
online_store_query = self._write_stream_data(transformed_df, to)
return online_store_query
self._write_stream_data(transformed_df, to)

def _ingest_stream_data(self) -> StreamTable:
"""Only supports json and avro formats currently."""
Expand Down Expand Up @@ -129,7 +126,7 @@ def _ingest_stream_data(self) -> StreamTable:
def _construct_transformation_plan(self, df: StreamTable) -> StreamTable:
return self.sfv.udf.__call__(df) if self.sfv.udf else df

def _write_stream_data(self, df: StreamTable, to: PushMode) -> StreamingQuery:
def _write_stream_data(self, df: StreamTable, to: PushMode):
# Validation occurs at the fs.write_to_online_store() phase against the stream feature view schema.
def batch_write(row: DataFrame, batch_id: int):
rows: pd.DataFrame = row.toPandas()
Expand Down Expand Up @@ -168,4 +165,3 @@ def batch_write(row: DataFrame, batch_id: int):
)

query.awaitTermination(timeout=self.query_timeout)
return query