Repository navigation
Add Spark materialization engine for parallel, distributed materialization of large datasets. #3167
Description
Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]
Current implementation of Spark offline store doesn't have Spark based materialization engine. This makes materialization slow, inefficient and makes Spark offline store not very useful since materialization is still happening in driver node and will be limited by its resources.
Describe the solution you'd like
A clear and concise description of what you want to happen.
Spark based materialization engine.
Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.
BytewaxMaterializationEngine - it relies on offline_job.to_remote_storage() but SparkRetrievalJob doesn't support to_remote_storage(). Also, would rather use one stack for job execution (preferably Spark) instead of two.
Additional context
Add any other context or screenshots about the feature request here.
spark_materialization_engine would make Feast highly scalable and leverage full Spark potential. Right now it it very limited.
Activity
adchia commentedon Sep 1, 2022
Hey! Thanks for the FR. Are you in the Slack? There have been recent discussions in the #feast-development Slack about this topic
niklasvm commentedon Sep 4, 2022
Hi @ckarwicki
I'm taking a look at potential designs for a
SparkBatchMaterializationEngine. I've got an open PR that implements theto_remote_storagemethod for aSparkRetrievalJobin the mean time. I currently see 3 solutions:(1) The default
LocalMaterializationEnginewrites the data to the local file system on the driver node and then loops over each parquet file in series. One possible solution is to parallelise this process however the success of this will still be limited by the size of the driver node.(2) Alternatively, data can be written to remote storage (like S3) and the
LambdaMaterializationEnginecan be used however this requires the use of AWS Lambda.(3) I don't know if it's possible to parallelise the code from the
LocalMaterializationEngineto run over the executor nodes.Any thoughts or ideas?
ckarwicki commentedon Sep 5, 2022
@niklasvm We shouldn't rely on parquet files. This should all be done in executors which will make materialization distributed. Initially we can parallelize
_materialize_one()after thatmaterialize(). We can start very simple. We can take code fromLocalMaterializationEngineand parallelize it inSparkBatchMaterializationEngine. Take outtable = offline_job.to_arrow()it is inefficient. Then take code https://github.com/feast-dev/feast/blob/master/sdk/python/feast/infra/materialization/local_engine.py#L158-L178 and put it in https://spark.apache.org/docs/3.1.1/api/python/reference/api/pyspark.sql.DataFrame.foreachPartition.html#pyspark.sql.DataFrame.foreachPartition. It is recommended to useforeachPartitionwhen writing to external stores.We basically have
SparkRetrievalJobafter a call toself.offline_store.pull_latest_from_table_or_query()so we execute that and we getsql.DataFramewhich is just df. We callforeachPartition()on that df and put that modified code fromLocalMaterializationEnginewhich will write rows fromsql.DataFrameto online store in executors.foreachPartitionis transformation so we will need to trigger action in driver, which is fine. With this approach materialization will be handled from executors in parallel, distributed way.niklasvm commentedon Sep 5, 2022
Thanks @ckarwicki for the explanation. I have actually gone ahead and implemeted something quite similar in #3184 but instead of using
foreachPartitionI've usedapplyInPandas. I'm not sure which is superior.ckarwicki commentedon Sep 5, 2022
@niklasvm Not sure if we should use
applyInPandasIt is memory intensive and requires full shuffle: https://spark.apache.org/docs/3.2.0/api/python/reference/api/pyspark.sql.GroupedData.applyInPandas.htmlIt works in three steps: Split-apply-combine, check this: https://docs.databricks.com/spark/latest/spark-sql/pandas-function-apis.html#grouped-map
We should use
foreachPartition()niklasvm commentedon Sep 5, 2022
That is very useful, thank you. I'll update the PR.
ckarwicki commentedon Sep 5, 2022
@niklasvm You should also use repartition() on df before calling
foreachPartition()where number of partitions should be number of simultaneous connections (writes) to online store. We can call it without arguments - it will create 200 partitions (number taken fromspark.sql.shuffle.partitions). It will be that many simultaneous connections (parallel writes) to Redis. Redis default is 10K connections. You can change it on Redis side if you need more.niklasvm commentedon Sep 6, 2022
I believe
repartitionwill cause a shuffle operation. Here's the DAG that usesrepartition:And without
repartition:Is the intention to only repartition to a number that redis can handle? Unfortunately
numPartitionsis a required parameter ofrepartition.Could we use coalesce()?
ckarwicki commentedon Sep 7, 2022
@niklasvm coalesce() should be used when partitioning down. It would be useful to call repartition() since it would allow control on the parallelism of materialization and simultaneous writes. They are saying it should be possible to call it without arguments:
niklasvm commentedon Sep 8, 2022
Thanks @ckarwicki. I have tested running this without a parameter and it fails.