GeezerVison is showing github.com — fix: Clean up snowflake to_spark_df() (#3607) · ExpediaGroup/feast@e8e643e
File treeExpand file treeCollapse file tree Open diff view settings docs/reference/offline-stores
Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: docs/reference/offline-stores/overview.md +3-3Lines changed: 3 additions & 3 deletions - Display the source diff
- Display the rich diff
| Original file line number | Diff line number | Diff line change |
|---|
@@ -46,11 +46,11 @@ Below is a matrix indicating which `RetrievalJob`s support what functionality. | 46 | 46 | | --------------------------------- | --- | --- | --- | --- | --- | --- | --- | | 47 | 47 | | export to dataframe | yes | yes | yes | yes | yes | yes | yes | | 48 | 48 | | export to arrow table | yes | yes | yes | yes | yes | yes | yes | | 49 | | -| export to arrow batches | no | no | yes | yes | no | no | no | | 50 | | -| export to SQL | no | yes | yes | yes | yes | no | yes | | | 49 | +| export to arrow batches | no | no | no | yes | no | no | no | | | 50 | +| export to SQL | no | yes | yes | yes | yes | no | yes | | 51 | 51 | | export to data lake (S3, GCS, etc.) | no | no | yes | no | yes | no | no | | 52 | 52 | | export to data warehouse | no | yes | yes | yes | yes | no | no | | 53 | | -| export as Spark dataframe | no | no | yes | no | no | yes | no | | | 53 | +| export as Spark dataframe | no | no | yes | no | no | yes | no | | 54 | 54 | | local execution of Python-based on-demand transforms | yes | yes | yes | yes | yes | no | yes | | 55 | 55 | | remote execution of Python-based on-demand transforms | no | no | no | no | no | no | no | | 56 | 56 | | persist results in the offline store | yes | yes | yes | yes | yes | yes | no | | |
Collapse file Copy file name to clipboardExpand all lines: docs/reference/offline-stores/snowflake.md +1-1Lines changed: 1 addition & 1 deletion - Display the source diff
- Display the rich diff
| Original file line number | Diff line number | Diff line change |
|---|
@@ -53,7 +53,7 @@ Below is a matrix indicating which functionality is supported by `SnowflakeRetri | 53 | 53 | | ----------------------------------------------------- | --------- | | 54 | 54 | | export to dataframe | yes | | 55 | 55 | | export to arrow table | yes | | 56 | | -| export to arrow batches | yes | | | 56 | +| export to arrow batches | yes | | 57 | 57 | | export to SQL | yes | | 58 | 58 | | export to data lake (S3, GCS, etc.) | yes | | 59 | 59 | | export to data warehouse | yes | | |
Collapse file Copy file name to clipboardExpand all lines: sdk/python/feast/errors.py -8Lines changed: 0 additions & 8 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -56,14 +56,6 @@ def __init__(self, name, project=None): | 56 | 56 | super().__init__(f"Feature view {name} does not exist") | 57 | 57 |
| 58 | 58 |
| 59 | | -class InvalidSparkSessionException(Exception): | 60 | | - def __init__(self, spark_arg): | 61 | | - super().__init__( | 62 | | - f" Need Spark Session to convert results to spark data frame\ | 63 | | - recieved {type(spark_arg)} instead. " | 64 | | - ) | 65 | | - | 66 | | - | 67 | 59 | class OnDemandFeatureViewNotFoundException(FeastObjectNotFoundException): | 68 | 60 | def __init__(self, name, project=None): | 69 | 61 | if project: | |
Collapse file Copy file name to clipboardExpand all lines: sdk/python/feast/infra/offline_stores/snowflake.py +11-21Lines changed: 11 additions & 21 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 28 | 28 |
| 29 | 29 | from feast import OnDemandFeatureView | 30 | 30 | from feast.data_source import DataSource | 31 | | -from feast.errors import ( | 32 | | - EntitySQLEmptyResults, | 33 | | - InvalidEntityType, | 34 | | - InvalidSparkSessionException, | 35 | | -) | | 31 | +from feast.errors import EntitySQLEmptyResults, InvalidEntityType | 36 | 32 | from feast.feature_logging import LoggingConfig, LoggingSource | 37 | 33 | from feast.feature_view import DUMMY_ENTITY_ID, DUMMY_ENTITY_VAL, FeatureView | 38 | 34 | from feast.infra.offline_stores import offline_utils | @@ -528,28 +524,22 @@ def to_spark_df(self, spark_session: "SparkSession") -> "DataFrame": | 528 | 524 | """ | 529 | 525 |
| 530 | 526 | try: | 531 | | - from pyspark.sql import DataFrame, SparkSession | | 527 | + from pyspark.sql import DataFrame | 532 | 528 | except ImportError as e: | 533 | 529 | from feast.errors import FeastExtrasDependencyImportError | 534 | 530 |
| 535 | 531 | raise FeastExtrasDependencyImportError("spark", str(e)) | 536 | 532 |
| 537 | | - if isinstance(spark_session, SparkSession): | 538 | | - arrow_batches = self.to_arrow_batches() | | 533 | + spark_session.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true") | 539 | 534 |
| 540 | | - if arrow_batches: | 541 | | - spark_df = reduce( | 542 | | - DataFrame.unionAll, | 543 | | - [ | 544 | | - spark_session.createDataFrame(batch.to_pandas()) | 545 | | - for batch in arrow_batches | 546 | | - ], | 547 | | - ) | 548 | | - return spark_df | 549 | | - else: | 550 | | - raise EntitySQLEmptyResults(self.to_sql()) | 551 | | - else: | 552 | | - raise InvalidSparkSessionException(spark_session) | | 535 | + # This can be improved by parallelizing the read of chunks | | 536 | + pandas_batches = self.to_pandas_batches() | | 537 | + | | 538 | + spark_df = reduce( | | 539 | + DataFrame.unionAll, | | 540 | + [spark_session.createDataFrame(batch) for batch in pandas_batches], | | 541 | + ) | | 542 | + return spark_df | 553 | 543 |
| 554 | 544 | def persist( | 555 | 545 | self, | |
|