Open bschifferer opened 2 years ago
The only thing I know that serialized Python lambdas
is cloudpickle
, so I guess we could serialize the Workflow to cloudpickle
and then deserialize it on the workers? Not too sure how to do that, but it's the only approach I can think of.
@bschifferer, as a short-term workaround - can you try this instead to explicitly set the dtype in the LambdaOp?
col_cat_int8 = col_cat_int8 >> nvt.ops.Categorify() >> nvt.ops.LambdaOp(lambda x: x.astype('int8'), dtype="int8")
workflow.transform(dataset).to_parquet(output_path, out_files_per_proc=10)
This should at least cause your workflow to execute on multiple workers.
I think the root cause of the issue here is that the 'capture_dtypes' functionality doesn't work on a distributed cluster environment - since the dtypes are only captured here https://github.com/NVIDIA-Merlin/NVTabular/blob/bc71785aab96368fb51f60d876fdfecba709e494/nvtabular/workflow/workflow.py#L534-L535 on the worker process, and the dtypes for each node in the graph aren't communicated back to the actual workflow.
The LambdaOp issue is a red-herring - we're already using cloudpickle under the hood (either to save the workflow itself, or distribute work with dask https://distributed.dask.org/en/stable/serialization.html#defaults . The issue was in raising the exception, called the LambdaOp.label functionality which failed =(. I've fixed the LambdaOp.label call here https://github.com/NVIDIA-Merlin/NVTabular/pull/1634- and added a basic test using LambdaOp's with dask there too
Yes that worked around works for me - thanks
Describe the bug I run a multi=GPU NVTabular workflow with LambdaOps. It executes the fit functionality, but when the pipeline does tramsform and to parquet, I get following error. I I run on single GPU, it does work.
Steps/Code to reproduce bug I use the criteo dask cluster initalization
E.g.