GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — feat: Implement spark materialization engine by niklasvm · Pull Request #3184 · feast-dev/feast

Conversation

@niklasvm

@niklasvm niklasvm commented Sep 5, 2022 •
edited
Loading

Uh oh!

There was an error while loading. Please reload this page.

Copy link
Copy Markdown
Collaborator

What this PR does / why we need it:

Implement SparkMaterializationEngine which parallelizes writing to the online store across spark executors. This introduces a spark batch engine type.

How

  1. Data is queried from the spark offline store
  2. foreachPartition is called on the spark data frame. Each partition of data is processed on the worker nodes.

Usage

The SparkMaterializationEngine is intended to only work with the SparkOfflineStore and an online store that supports parallel writes (not sqlite).

e.g. feature_store.yaml:

project: my_project
registry: data/registry.db
provider: local
batch_engine:
    type: spark.engine
offline_store:
    type: spark.offline
    spark_conf:
        spark.master: "local[*]"
        spark.ui.enabled: "false"
        spark.eventLog.enabled: "false"
        spark.sql.catalogImplementation: "hive"
        spark.sql.parser.quotedRegexColumnNames: "true"
        spark.sql.session.timeZone: "UTC"
online_store:
    type: redis
        connection_string: "localhost:6379"
entity_key_serialization_version: 2

Some considerations

  • Should we distinguish between the spark offline store and engine the same way as snowflake i.e. spark.offline and spark.engine?
  • Can we introduce some larger test data sets that run part of integration tests?

Unit and integration tests are running successfully however this process should be tested on a larger set of data to ensure parallelization is working appropriately.

Which issue(s) this PR fixes:

Fixes #3167

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
All reactions
@niklasvm
implement spark materialization engine
26e0b3e 
Signed-off-by: niklasvm <[email protected]>
@feast-ci-bot feast-ci-bot added do-not-merge/work-in-progress size/L labels Sep 5, 2022
@niklasvm
remove redundant code
807e7ca 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
niklasvm marked this pull request as ready for review September 5, 2022 09:32
@feast-ci-bot feast-ci-bot removed the do-not-merge/work-in-progress label Sep 5, 2022
@codecov-commenter

codecov-commenter commented Sep 5, 2022 •
edited
Loading

Uh oh!

There was an error while loading. Please reload this page.

Copy link
Copy Markdown

Codecov Report

Base: 67.02% // Head: 58.28% // Decreases project coverage by -8.73% ⚠️

Coverage data is based on head (f828d2b) compared to base (11a0e1a).
Patch coverage: 51.61% of modified lines in pull request are covered.

Additional details and impacted files 
@@            Coverage Diff             @@
##           master    #3184      +/-   ##
==========================================
- Coverage   67.02%   58.28%   -8.74%     
==========================================
  Files         175      210      +35     
  Lines       15942    17689    +1747     
==========================================
- Hits        10685    10310     -375     
- Misses       5257     7379    +2122
Flag Coverage Δ
integrationtests ?
unittests 58.28% <51.61%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files Coverage Δ
sdk/python/feast/repo_config.py 76.04% <ø> (-6.47%) ⬇️
...ration/materialization/contrib/spark/test_spark.py 50.00% <50.00%> (ø)
...ffline_stores/contrib/spark_offline_store/spark.py 33.18% <100.00%> (ø)
...sts/integration/registration/test_universal_cli.py 20.20% <0.00%> (-79.80%) ⬇️
...ts/integration/offline_store/test_offline_write.py 26.08% <0.00%> (-73.92%) ⬇️
...fline_store/test_universal_historical_retrieval.py 28.75% <0.00%> (-71.25%) ⬇️
...ests/integration/e2e/test_python_feature_server.py 29.50% <0.00%> (-70.50%) ⬇️
...dk/python/tests/integration/e2e/test_validation.py 27.55% <0.00%> (-69.30%) ⬇️
...s/integration/registration/test_universal_types.py 32.25% <0.00%> (-67.75%) ⬇️
sdk/python/feast/infra/online_stores/redis.py 28.39% <0.00%> (-66.67%) ⬇️
... and 162 more

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

☔ View full report at Codecov.
📢 Do you have feedback about the report comment? Let us know in this issue.
All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
niklasvm added 2 commits September 5, 2022 11:35
@niklasvm
make function private
f8e70ea 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
refactor serializing into a class
b42352e 
Signed-off-by: niklasvm <[email protected]>
@niklasvm niklasvm mentioned this pull request Sep 5, 2022
Add Spark materialization engine for parallel, distributed materialization of large datasets. #3167
Closed
niklasvm added 2 commits September 5, 2022 21:43
@niklasvm
switch to using foreachPartition
f609cfb 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
remove batch_size parameter
79ea412 
Signed-off-by: niklasvm <[email protected]>
@kevjumba kevjumba added the ok-to-test label Sep 7, 2022
@kevjumba kevjumba self-assigned this Sep 7, 2022
niklasvm added 5 commits September 7, 2022 22:04
@niklasvm
add partitions parameter
937a0e3 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
linting
610614a 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
rename spark to spark.offline and spark.engine
84cc858 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
fix to test
8cc4928 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
forgot to stage
6a9663d 
Signed-off-by: niklasvm <[email protected]>
achals
achals reviewed Sep 7, 2022
View reviewed changes
Comment thread sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py Resolved

Uh oh!

There was an error while loading. Please reload this page.
Comment thread sdk/python/feast/infra/materialization/contrib/spark/spark_materialization_engine.py
Comment on lines +175 to +183
spark_df = offline_job.to_spark_df()
if self.repo_config.batch_engine.partitions != 0:
spark_df = spark_df.repartition(
self.repo_config.batch_engine.partitions
)

spark_df.foreachPartition(
lambda x: _process_by_partition(x, spark_serialized_artifacts)
)

@achals achals Sep 7, 2022

Copy link
Copy Markdown
Member

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💯

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
niklasvm reacted with thumbs up emoji
All reactions
  • 👍 1 reaction
niklasvm added 5 commits September 8, 2022 07:39
@niklasvm
revert spark.offline to spark to ensure backward compatibility
715bb72 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
Merge branch 'master' into spark_materialization_engine
de1a85a
@niklasvm
fix import
32b3111 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
remove code from testing a large data set
0c13af9 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
linting
542705f 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
niklasvm requested a review from achals September 8, 2022 07:00
@ckarwicki

ckarwicki commented Sep 13, 2022

Copy link
Copy Markdown
Contributor

@niklasvm Can you fix integration tests? We are also waiting on this PR and would like to use it. Did you get a chance to test it on a cluster? Feast would have to be provided to worker nodes since it deserializes config. We can have example how to do it.
All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
@niklasvm

niklasvm commented Sep 13, 2022

Copy link
Copy Markdown
Collaborator Author

@niklasvm Can you fix integration tests? We are also waiting on this PR and would like to use it. Did you get a chance to test it on a cluster? Feast would have to be provided to worker nodes since it deserializes config. We can have example how to do it.

@ckarwicki I didn't realise the integration test failed. It looks like the issue is related to the redis docker container not accepting connections. I will rerun the pipeline.

I have not tested this on a cluster, only in spark local mode. What type of cluster are you using?
All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
@niklasvm
test without repartition
262af10 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
niklasvm requested a review from kevjumba as a code owner September 13, 2022 04:56
niklasvm added 2 commits September 13, 2022 07:06
@niklasvm
test alternate connection string
8e59da2 
Signed-off-by: niklasvm <[email protected]>
@niklasvm
use redis online creator
f828d2b 
Signed-off-by: niklasvm <[email protected]>
adchia
adchia approved these changes Sep 13, 2022
View reviewed changes
Comment thread sdk/python/feast/infra/materialization/contrib/spark/spark_materialization_engine.py
# unserialize artifacts
feature_view, online_store, repo_config = spark_serialized_artifacts.unserialize()

if feature_view.batch_source.field_mapping is not None:

@adchia adchia Sep 13, 2022

Copy link
Copy Markdown
Collaborator

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

since lines 249 to 257 are also used in feature_store.write_to_online_store, maybe it makes sense to refactor this into a util method?

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
All reactions
Comment thread sdk/python/feast/infra/materialization/contrib/spark/spark_materialization_engine.py
)


class SparkMaterializationEngineConfig(FeastConfigBaseModel):

@adchia adchia Sep 13, 2022

Copy link
Copy Markdown
Collaborator

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

probably makes sense to throw an error somewhere if the offline store is not the SparkOfflineStore?

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
All reactions
@feast-ci-bot feast-ci-bot added the approved label Sep 13, 2022
@ckarwicki

ckarwicki commented Sep 13, 2022 •
edited
Loading

Uh oh!

There was an error while loading. Please reload this page.

Copy link
Copy Markdown
Contributor

@niklasvm We are running Bitnami Spark cluster on EKS.
All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
@hao-affirm

hao-affirm commented Sep 14, 2022

Copy link
Copy Markdown
Contributor

I'm also looking forward to use this!
pengyuan reacted with thumbs up emoji
All reactions
  • 👍 1 reaction

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
@niklasvm

niklasvm commented Sep 15, 2022

Copy link
Copy Markdown
Collaborator Author

@adchia what is left before this can be merged. I see there is one test failing however the failures are unrelated to this PR
All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
adchia
adchia approved these changes Sep 15, 2022
View reviewed changes

@adchia adchia left a comment

Copy link
Copy Markdown
Collaborator

There was a problem hiding this comment.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

/lgtm

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
niklasvm reacted with thumbs up emoji
All reactions
  • 👍 1 reaction
@feast-ci-bot feast-ci-bot assigned adchia Sep 15, 2022
@feast-ci-bot feast-ci-bot added the lgtm label Sep 15, 2022
@feast-ci-bot

feast-ci-bot commented Sep 15, 2022

Copy link
Copy Markdown
Collaborator

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: adchia, niklasvm

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Details Needs approval from an approver in each of these files:
  • OWNERS [adchia]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment
All reactions

Sorry, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.
@feast-ci-bot
feast-ci-bot merged commit a59c33a into feast-dev:master Sep 15, 2022
felixwang9817 pushed a commit that referenced this pull request Sep 20, 2022
@feast-ci-bot
chore(release): release 0.25.0
769c318 
# [0.25.0](v0.24.0...v0.25.0) (2022-09-20)

### Bug Fixes

* Broken Feature Service Link ([#3227](#3227)) ([e117082](e117082))
* Feature-server image is missing mysql dependency for mysql registry ([#3223](#3223)) ([ae37b20](ae37b20))
* Fix handling of TTL in Go server ([#3232](#3232)) ([f020630](f020630))
* Fix materialization when running on Spark cluster. ([#3166](#3166)) ([175fd25](175fd25))
* Fix push API to respect feature view's already inferred entity types ([#3172](#3172)) ([7c50ab5](7c50ab5))
* Fix release workflow ([#3144](#3144)) ([20a9dd9](20a9dd9))
* Fix Shopify timestamp bug and add warnings to help with debugging entity registration ([#3191](#3191)) ([de75971](de75971))
* Handle complex Spark data types in SparkSource ([#3154](#3154)) ([5ddb83b](5ddb83b))
* Local staging location provision ([#3195](#3195)) ([cdf0faf](cdf0faf))
* Remove bad snowflake offline store method ([#3204](#3204)) ([dfdd0ca](dfdd0ca))
* Remove opening file object when validating S3 parquet source ([#3217](#3217)) ([a906018](a906018))
* Snowflake config file search error ([#3193](#3193)) ([189afb9](189afb9))
* Update Snowflake Online docs ([#3206](#3206)) ([7bc1dff](7bc1dff))

### Features

* Add `to_remote_storage` functionality to `SparkOfflineStore` ([#3175](#3175)) ([2107ce2](2107ce2))
* Add ability to give boto extra args for registry config ([#3219](#3219)) ([fbc6a2c](fbc6a2c))
* Add health endpoint to py server ([#3202](#3202)) ([43222f2](43222f2))
* Add snowflake support for date & number with scale ([#3148](#3148)) ([50e8755](50e8755))
* Add tag kwarg to set Snowflake online store table path ([#3176](#3176)) ([39aeea3](39aeea3))
* Add workgroup to athena offline store config ([#3139](#3139)) ([a752211](a752211))
* Implement spark materialization engine ([#3184](#3184)) ([a59c33a](a59c33a))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@adchia adchia adchia approved these changes
@achals achals Awaiting requested review from achals
@kevjumba kevjumba Awaiting requested review from kevjumba

Uh oh!

There was an error while loading. Please reload this page.

Assignees

@adchia adchia

@kevjumba kevjumba

Labels

approved lgtm ok-to-test size/L

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

Add Spark materialization engine for parallel, distributed materialization of large datasets.

Uh oh!

There was an error while loading. Please reload this page.

8 participants

@niklasvm @codecov-commenter @ckarwicki @hao-affirm @feast-ci-bot @achals @adchia @kevjumba
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
You are visitor 01402503 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.