[data] Support large-scale HuggingFace datasets

[raulchen]

Current ray.data.from_huggingface needs to materialize all data first. We need to make it more scalable.
Potential solutions:

1. support IterableDataset
2. List parquet files and use ray.data.read_parquet

[c21]

Just to add more context:

Currently from_huggingface() would load the Hugging Face dataset in-memory in a single node. This is working for small size of dataset, but not large size. For example, user hit OOM issue when trying to use https://huggingface.co/datasets/togethercomputer/RedPajama-Data-1T . We should change the API to make the reading in a streaming way across multiple tasks.

Hugging Face dataset streaming - https://huggingface.co/docs/datasets/stream
split_dataset_by_node - https://huggingface.co/docs/datasets/package_reference/main_classes#datasets.distributed.split_dataset_by_node

Use case
Read large scale Hugging Face dataset efficiently for fine tuning LLM

[pcmoritz]

Given the performance of #38432, I don't think we should mark this as done yet :)

[anyscalesam]

@c21 please review with @pcmoritz on whether performance gap has been addressed

[npuichigo]

@pcmoritz I cannot find any conext regarding the performance issue. Could you help to explain that.

Besides, what's the recommend way to consume large huggingface dataset now? I'm not sure the way I do is correct

1. use to_parquet to save huggingface datasets as parquet format.
2. use ray for parallel read.

[lhoestq]

If it can help, you can stream a dataset using an IterableDataset and stream it in a distributed setup using

from datasets.distributed import split_dataset_by_node

ds = split_dataset_by_node(ds, rank=rank, world_size=world_size)

The documentation is here