Right now, the number of shuffle mappers is equal to the number of input files, with each mapper reading a single input file. However:

1. For very large files, we may want multiple mappers reading segments of a file in parallel for the sake of memory utilization and/or maximizing read throughput.
2. For many very small files, we may want multiple files to be read by a single mapper for the sake of reducing task overhead and/or possibly maximizing read throughput.

(1) has already been brought up by a user as a concrete need. We should therefore support this by allowing num_mappers != num_files, and letting num_mappers be set by the user with a best-effort default as a fallback.

Implementation Notes

(2) (i.e. num_files > num_mappers) won't involve much more than the typical np.array_split() pattern, e.g.

for files in np.array_split(filenames, num_mappers):
    reducer_parts = shuffle_map.options(num_returns=num_reducers).remote(files, ...)

(1) (i.e. num_files < num_mappers), however, will require us to read a file starting at a certain offset; given that the Pandas read_parquet() API doesn't expose that functionality, the most likely course of action would be using the pyarrow Parquet reading facilities directly. Here is an example of such usage, although we will most likely want to use a ParquetDataset with split_row_groups=True and rely on the row group granularity for the MVP.

Add user guide to docs. This should include:

• Walkthrough of installation, setup, and example.
• Cataloguing of all configuration details, e.g. what to set num_reducers to, why max_concurrent_epochs should almost always be 2, etc.

Add unit tests for the following:

1. Shuffler (shuffle.py)
2. BatchQueue (batch_queue.py)
3. ShufflingDataset (dataset.py)
4. TorchShufflingDataset (torch_dataset.py)

Implementation Notes

• ShufflingDataset and TorchShufflingDataset testing could initially be as simple as running the example in the main block of dataset.py and torch_dataset.py, respectively, although these don't do much to assert correctness of the yielded batches.
• See these tests for easily portable examples for testing BatchQueue.

Add a contributing README to repo.

We need to build a connector to TF dataset iterator.

Impl idea from @clarkzinzow:
We’d take the base shuffling dataset, create a ShufflingTFDataset that converts each batch dataframe to feature and target tensors, and then pass that ShufflingTFDataset as the generator to tf.data.Dataset.from_generator, which can then be used as your typical TensorFlow dataset:

ds = tf.data.Dataset.from_generator(ShufflingTFDataset(filenames, num_epochs, num_trainers, batch_size, dataframe_to_tensor_spec))
for batch_idx, (features, targets) in enumerate(ds):
    print(f"Processing batch {batch_idx}!")

I can’t see any obvious issues with doing this except for mapping TensorFlow’s distributed dataset + data-parallel training paradigms to our current rank-based shuffling dataset, where we kick off the shuffle from the rank-0 training worker and give each worker an independent queue of batches. The latter should be doable from via getting the replica ID from tf.distribute.get_replica_context() during iteration and using that to access the correct queue, but the former paradigm may need to be tweaked.