Cross posting from Discord since I saw the announcement ther

Question

Cross posting from Discord since I saw the announcement there a little too late Hello everyone I m relatively new to Kedro I m using it together with Dask for some data processing and I have some issues questions with regards to data locality I have a pipeline that has three nodes where the datasets are loaded like follows `dask ParquetDataSet from s3 gt MemoryDataSet gt dask ParquetDataSet to s3` I run this pipeline from my local workstation for testing purposes My Dask Cluster is then deployed on AWS EC2 Scheduler+Workers and they communicate privately I noticed that on the last node the `MemoryDataSet gt dask ParquetDataSet to s3` causes the data to be transferred to my local machine where the Kedro pipeline is being run and then transferred back to s3 Needless to say this introduces costs and lag and is not what I intended Can I tell the workers to write this data directly to s3 If not what is the intended way to do this I read through the documentation and there is some very good information on getting the Pipeline to run as either or on but this is not quite the deployment flow I had in mind Is it intended for the pipeline to be run on the same infrastructure where the workers are deployed

datajoely · Answer

I m not actually sure on this one < Deepyaman Datta> any ideas

Filip Panovski · Answer

It might be worth noting that the last node isn t quite the standard `dask ParquetDataSet` but a custom one that subclasses it All I m doing is passing a pyarrow schema to the ` save` method though ``` def save self data dd DataFrame gt None data to parquet self filepath schema=pa schema self pyarrow schema storage options=self fs args self save args ``` as opposed to the standard `dask ParquetDataSet save` ``` def save self data dd DataFrame gt None data to parquet self filepath storage options=self fs args self save args ``` so I don t think this should have any bearing on behaviour but just in case

Deepyaman Datta · Answer

All `dask ParquetDataSet` does is load as a Dask distributed dataframe it doesn t actually do execution utilizing a Dask cluster A common use case for this is when reading a partitioned Parquet eg written by Spark because Pandas doesn t handle that natively If you actually want to utilize a Dask cluster for execution you should create a `DaskRunner` <https kedro readthedocs io en latest deployment dask html>

Filip Panovski · Answer

Ah I did see those docs but I must have misunderstood them a bit I ll give it a try Thanks

Filip Panovski · Answer

I m running into some problems with the above approach unfortunately I ve got a runner set up after following the docs and am starting the pipeline as `kedro run runner=product classification runner DaskRunner pipeline type transformation` but my Tasks are now failing with the following error ```2022 11 04 10 26 10 440 distributed worker ERROR Could not deserialize task run node 8a52e7a0be61dae2e732e8bcf7b955cc Traceback most recent call last File opt conda lib python3 10 site packages distributed worker py line 2845 in loads function result = cache loads bytes object File opt conda lib python3 10 site packages distributed collections py line 24 in getitem value = super getitem key File opt conda lib python3 10 collections init py line 1106 in getitem raise KeyError key KeyError b x80 x04 x95F x00 x00 x00 x00 x00 x00 x00 x8c product classification runner dask runner x94 x8c x14DaskRunner run node x94 x93 x94 During handling of the above exception another exception occurred Traceback most recent call last File opt conda lib python3 10 site packages distributed worker py line 2212 in execute function args kwargs = await self maybe deserialize task ts File opt conda lib python3 10 site packages distributed worker py line 2185 in maybe deserialize task function args kwargs = deserialize ts run spec File opt conda lib python3 10 site packages distributed worker py line 2856 in deserialize function = loads function function File opt conda lib python3 10 site packages distributed worker py line 2847 in loads function result = pickle loads bytes object File opt conda lib python3 10 site packages distributed protocol pickle py line 73 in loads return pickle loads x ModuleNotFoundError No module named product classification ``` I m guessing this is some kind of serialization error where it s trying to serialize and distribute my code to the workers and failing to do so Do I have to package the code and install it on the Workers manually for the DaskRunner approach to work I can t find much info unfortunately though I m not 100 sure that this is a Kedro issue at this point Here s my project structure in case it helps <https gist github com filpano de3f92542eb225fdfd46a7bd7a43145c>

Deepyaman Datta · Answer

Yeah exactly the code needs to be available to all the workers but I m not exactly sure what the solution here is I did a bit of quick looking around and it seems like `cloudpickle` should automatically include dependencies during serialization `pickle` would not but some threads from a few years ago say Dask uses `cloudpickle` by default However <https distributed dask org en stable serialization html> says that Dask uses a mixture of `dask` and `pickle` by default but maybe you can specify `cloudpickle` I m unfortunately not a Dask expert much as I would love to be so take these thoughts with plenty of salt

Filip Panovski · Answer

Thank you very much that s already more info than I was able to find on the topic I ll take a look I m also unfortunately not a Dask expert