Added Ray remote backend and Dask distributed preprocessing #1090
Conversation
| # TODO ray: select this more intelligently, | ||
| # must be greather than or equal to number of Horovod workers | ||
| return dict( | ||
| parallelism=int(ray.cluster_resources()['CPU']) |
There was a problem hiding this comment.
@clarkzinzow does this make sense as the default repartition value? One partition per CPU? Not sure if there's a more reasonable heuristic for this. The one restriction we have is that for Petastorm, we must have at least one row group per Horovod worker, and the safest way to guarantee this at the moment is to repartition the dataframe.
There was a problem hiding this comment.
That's the typical heuristic, yes, under the soft constraint of those chunks/partitions fitting nicely into each worker's memory.
| features = get_combined_features(config) | ||
| for feature in features: | ||
| name = feature[NAME] | ||
| proc_column = feature[PROC_COLUMN] | ||
| reshape = training_set_metadata[name].get('reshape') | ||
| if reshape is not None: | ||
| dataset[proc_column] = self.map_objects(dataset[proc_column], lambda x: x.reshape(-1)) |
There was a problem hiding this comment.
Curious about this, what is it a work around for? (probably worth adding a comment about it too)
There was a problem hiding this comment.
Basically, PyArrow cannot serialize the data as a multi-dimensional tensors, so we need to store the shape so when we read it back, we can reshape it into its correct form. Will add a comment to this effect.
| t = getattr(sample, feature_name) | ||
| reshape_dim = self.reshape_features.get(feature_name) | ||
| if reshape_dim is not None: | ||
| t = tf.reshape(t, reshape_dim) |
There was a problem hiding this comment.
I guess my previous question had to do with this, right? Tensors with multiple ranks get squashed and then put back in shape, right?
There was a problem hiding this comment.
Yes, exactly, I can add a comment here.
No description provided.