Repository: huggingface/datasets
Pull request: Map-style Dataset to IterableDataset
Added `ds.to_iterable()` to get an iterable dataset from a map-style arrow dataset.

It also has a `num_shards` argument to split the dataset before converting to an iterable dataset. Sharding is important to enable efficient shuffling and parallel loading of iterable datasets.

TODO:
- [x] tests
- [x] docs

Fix https://github.com/huggingface/datasets/issues/5265

Problem statement:
Get an IterableDataset from a map-style Dataset
This is useful to leverage iterable datasets specific features like:
- fast approximate shuffling
- lazy map, filter etc.

Iterating over the resulting iterable dataset should be at least as fast at iterating over the map-style dataset.

Here are some ideas regarding the API:
```python
# 1. 
# - consistency with load_dataset(..., streaming=True)
# - gives intuition that map/filter/etc. are done on-the-fly
ids = ds.stream()

# 2.
# - more explicit on the output type
# - but maybe sounds like a conversion tool rather than a step in a processing pipeline
ids = ds.as_iterable_dataset()
```

Hints from issue discussion:
I think `stream` could be misleading since the data is not being streamed from remote endpoints (one could think that's the case when they see `load_dataset` followed by `stream`). Hence, I prefer the second option.

PS: When we resolve https://github.com/huggingface/datasets/issues/4542, we could add `as_tf_dataset` to the API for consistency and deprecate `to_tf_dataset`.

Commit message: style

Patch to review:
diff --git a/docs/source/_toctree.yml b/docs/source/_toctree.yml
index 0d2fe5c7ac5..adfbc7637b2 100644
--- a/docs/source/_toctree.yml
+++ b/docs/source/_toctree.yml
@@ -95,6 +95,8 @@
     title: Datasets 🤝 Arrow
   - local: about_cache
     title: The cache
+  - local: about_mapstyle_vs_iterable
+    title: Dataset or IterableDataset
   - local: about_dataset_features
     title: Dataset features
   - local: about_dataset_load
diff --git a/docs/source/about_mapstyle_vs_iterable.mdx b/docs/source/about_mapstyle_vs_iterable.mdx
new file mode 100644
index 00000000000..44fca70472d
--- /dev/null
+++ b/docs/source/about_mapstyle_vs_iterable.mdx
@@ -0,0 +1,219 @@
+# Differences between Dataset and IterableDataset
+
+There are two types of dataset objects, a [`Dataset`] and an [`IterableDataset`].
+Whichever type of dataset you choose to use or create depends on the size of the dataset.
+In general, an [`IterableDataset`] is ideal for big datasets (think hundreds of GBs!) due to its lazy behavior and speed advantages, while a [`Dataset`] is great for everything else.
+This page will compare the differences between a [`Dataset`] and an [`IterableDataset`] to help you pick the right dataset object for you.
+
+## Downloading and streaming
+
+When you have a regular [`Dataset`], you can access it using `my_dataset[0]`. This provides "random access" to the rows.
+Such datasets are also called "map-style" datasets.
+For example you can download ImageNet-1k like this and access any row:
+
+```python
+from datasets import load_dataset
+
+imagenet = load_dataset("imagenet-1k", split="train")  # downloads the full dataset
+print(imagenet[0])
+```
+
+But one caveat is that you must have the entire dataset stored on your disk or in memory, which blocks you from accessing datasets bigger than the disk.
+Because it can become inconvenient for big datasets, there exists another type of dataset, the [`IterableDataset`].
+When you have an `IterableDataset`, you can access it using a `for` loop to load the data progressively as you iterate over the dataset.
+This way, only a small fraction of examples is loaded in memory, and you don't write anything on disk.
+
+For example, you can stream the ImageNet-1k dataset without waiting for it to download completely:
+
+```python
+from datasets import load_dataset
+
+imagenet = load_dataset("imagenet-1k", split="train", streaming=True)  # will start loading the data when iterated over
+for example in imagenet:
+    print(example)
+    break
+```
+
+Streaming can read online data without writing any file to disk.
+For example, you can stream datasets made out of multiple shards, each of which is hundreds of gigabytes like [C4](https://huggingface.co/datasets/c4), [OSCAR](https://huggingface.co/datasets/oscar) or [LAION-2B](https://huggingface.co/datasets/laion/laion2B-en).
+Learn more about how to stream a dataset in the [Dataset Streaming Guide](./stream).
+
+This is not the only difference though, because the "lazy" behavior of an `IterableDataset` is also present when it comes to dataset creation and processing.
+
+## Creating map-style datasets and iterable datasets
+
+You can create a [`Dataset`] using lists or dictionaries, and the data is entirely converted to Arrow so you can easily access any row:
+```python
+my_dataset = Dataset.from_dict({"col_1": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]})
+print(my_dataset[0])
+```
+
+To create an `IterableDataset` on the other hand, you must provide a "lazy" way to load the data.
+In Python, we generally use generator functions. These functions `yield` one example at a time, which means you can't access a row by slicing it like a regular `Dataset`:
+```python
+def my_generator(n):
+    for i in range(n):
+        yield {"col_1": i}
+
+my_iterable_dataset = IterableDataset.from_generator(my_generator, gen_kwargs={"n": 10})
+for example in my_iterable_dataset:
+    print(example)
+    break
+```
+
+## Loading local files entirely and progressively
+
+It is possible to convert local or remote data files to an Arrow [`Dataset`] using [`load_dataset`]:
+```python
+data_files = {"train": ["path/to/data.csv"]}
+my_dataset = load_dataset("csv", data_files=data_files, split="train")
+print(my_dataset[0])
+```
+
+However, this requires a conversion step from CSV to Arrow format, which takes time and disk space if your dataset is big.
+
+To save disk space and skip the conversion step, you can define an `IterableDataset` by streaming from the local files directly.
+This way, the data is read progressively from the local files as you iterate over the dataset:
+
+```python
+data_files = {"train": ["path/to/data.csv"]}
+my_iterable_dataset = load_dataset("csv", data_files=data_files, split="train", streaming=True)
+for example in my_iterable_dataset:  # this reads the CSV file progressively as you iterate over the dataset
+    print(example)
+    break
+```
+
+Many file formats are supported, like CSV, JSONL, and Parquet, as well as image and audio files.
+You can find more information in the corresponding guides for loading [tabular](./tabular_load), [text](./nlp_load), [vision](./image_load), and [audio](./audio_load]) datasets.
+
+## Eager data processing and lazy data processing
+
+When you process a [`Dataset`] object using [`Dataset.map`], the entire dataset is processed immediately and returned.
+This is similar to how `pandas` works for example.
+
+```python
+my_dataset = my_dataset.map(process_fn)  # process_fn is applied on all the examples of the dataset
+print(my_dataset[0])
+```
+
+On the other hand, due to the "lazy" nature of an `IterableDataset`, calling [`IterableDataset.map`] does not apply your `map` function over the full dataset.
+Instead, your `map` function is applied on-the-fly.
+
+Because of that, you can chain multiple processing steps and they will all run at once when you start iterating over the dataset:
+
+```python
+my_iterable_dataset = my_iterable_dataset.map(process_fn_1)
+my_iterable_dataset = my_iterable_dataset.filter(filter_fn)
+my_iterable_dataset = my_iterable_dataset.map(process_fn_2)
+
+# process_fn_1, filter_fn and process_fn_2 are applied on-the-fly when iterating over the dataset
+for example in my_iterable_dataset:  
+    print(example)
+    break
+```
+
+## Exact and fast approximate shuffling
+
+When you shuffle a [`Dataset`] using [`Dataset.shuffle`], you apply an exact shuffling of the dataset.
+It works by taking a list of indices `[0, 1, 2, ... len(my_dataset) - 1]` and shuffling this list.
+Then, accessing `my_dataset[0]` returns the row and index defined by the first element of the indices mapping that has been shuffled:
+```python
+my_dataset = my_dataset.shuffle(seed=42)
+print(my_dataset[0])
+```
+
+Since we don't have random access to the rows in the case of an `IterableDataset`, we can't use a shuffled list of indices and access a row at an arbitrary position.
+This prevents the use of exact shuffling.
+Instead, a fast approximate shuffling is used in [`IterableDataset.shuffle`].
+It uses a shuffle buffer to sample random examples iteratively from the dataset.
+Since the dataset is still read iteratively, it provides excellent speed performance:
+```python
+my_iterable_dataset = my_iterable_dataset.shuffle(seed=42, buffer_size=100)
+for example in my_iterable_dataset:
+    print(example)
+    break
+```
+
+But using a shuffle buffer is not enough to provide a satisfactory shuffling for machine learning model training. So [`IterableDataset.shuffle`] also shuffles the dataset shards if your dataset is made of multiple files or sources:
+
+```python
+# Stream from the internet
+my_iterable_dataset = load_dataset("c4", "en", split="train", streaming=True)
+my_iterable_dataset.n_shards  # 1024
+
+# Stream from local files
+data_files = {"train": [f"path/to/data_{i}.csv" for i in range(1024)]}
+my_iterable_dataset = load_dataset("csv", data_files=data_files, split="train", streaming=True)
+my_iterable_dataset.n_shards  # 1024
+
+# From a generator function
+def my_generator(n, sources):
+    for source in sources:
+        for example_id_for_current_source in range(n):
+            yield {"example_id": f"{source}_{example_id_for_current_source}"}
+
+gen_kwargs = {"n": 10, "sources": [f"path/to/data_{i}" for i in range(1024)]}
+my_iterable_dataset = IterableDataset.from_generator(my_generator, gen_kwargs=gen_kwargs)
+my_iterable_dataset.n_shards  # 1024
+```
+
+## Speed differences
+
+Regular [`Dataset`] objects are based on Arrow which provides fast random access to the rows.
+Thanks to memory mapping and the fact that Arrow is an in-memory format, reading data from disk doesn't do expensive system calls and deserialization.
+It provides even faster data loading when iterating using a `for` loop by iterating on contiguous Arrow record batches.
+
+However as soon as your [`Dataset`] has an indices mapping (via [`Dataset.shuffle`] for example), the speed can become 10x slower.
+This is because there is an extra step to get the row index to read using the indices mapping, and most importantly, you aren't reading contiguous chunks of data anymore.
+To restore the speed, you'd need to rewrite the entire dataset on your disk again using [`Dataset.flatten_indices`], which removes the indices mapping.
+This may take a lot of time depending of the size of your dataset though:
+
+```python
+my_dataset[0]  # fast
+my_dataset = my_dataset.shuffle(seed=42)
+my_dataset[0]  # up to 10x slower
+my_dataset = my_dataset.flatten_indices()  # rewrite the shuffled dataset on disk as contiguous chunks of data
+my_dataset[0]  # fast again
+```
+
+
+In this case, we recommend switching to an [`IterableDataset`] and leveraging its fast approximate shuffling method [`IterableDataset.shuffle`].
+It only shuffles the shards order and adds a shuffle buffer to your dataset, which keeps the speed of your dataset optimal.
+You can also reshuffle the dataset easily:
+
+```python
+for example in enumerate(my_iterable_dataset):  # fast
+    pass
+
+shuffled_iterable_dataset = my_iterable_dataset.shuffle(seed=42, buffer_size=100)
+
+for example in enumerate(shuffled_iterable_dataset):  # as fast as before
+    pass
+
+shuffled_iterable_dataset = my_iterable_dataset.shuffle(seed=1337, buffer_size=100)  # reshuffling using another seed is instantaneous
+
+for example in enumerate(shuffled_iterable_dataset):  # still as fast as before
+    pass
+```
+
+If you're using your dataset on multiple epochs, the effective seed to shuffle the shards order in the shuffle buffer is `seed + epoch`.
+It makes it easy to reshuffle a dataset betwen epochs:
+```python
+for epoch in range(n_epochs):
+    my_iterable_dataset.set_epoch(epoch)
+    for example in my_iterable_dataset:  # fast + reshuffled at each epoch using `effective_seed = seed + epoch`
+        pass
+```
+
+## Switch from map-style to iterable
+
+If you want to benefit from the "lazy" behavior of an [`IterableDataset`] or their speed advantages, you can switch your map-style [`Dataset`] to an [`IterableDataset`]:
+```python
+my_iterable_dataset = my_dataset.to_iterable_dataset()
+```
+
+If you want to shuffle your dataset or [use it with a PyTorch DataLoader](./use_with_pytorch#stream-data), we recommend generating a shared [`IterableDataset`]:
+```python
+my_iterable_dataset = my_dataset.to_iterable_dataset(num_shards=1024)
+my_iterable_dataset.n_shards  # 1024
+```
diff --git a/docs/source/process.mdx b/docs/source/process.mdx
index 481b204be80..e34f86b1c9b 100644
--- a/docs/source/process.mdx
+++ b/docs/source/process.mdx
@@ -28,7 +28,8 @@ All processing methods in this guide return a new [`Dataset`] object. Modificati
 
 ## Sort, shuffle, select, split, and shard
 
-There are several functions for rearranging the structure of a dataset. These functions are useful for selecting only the rows you want, creating train and test splits, and sharding very large datasets into smaller chunks.
+There are several functions for rearranging the structure of a dataset.
+These functions are useful for selecting only the rows you want, creating train and test splits, and sharding very large datasets into smaller chunks.
 
 ### Sort
 
@@ -44,6 +45,9 @@ Use [`~Dataset.sort`] to sort column values according to their numerical values.
 [1, 1, 1, 1, 1, 1, 1, 1, 1, 1]
 ```
 
+Under the hood, this creates a list of indices that is sorted according to values of the column.
+This indices mapping is then used to access the right rows in the underlying Arrow table.
+
 ### Shuffle
 
 The [`~Dataset.shuffle`] function randomly rearranges the column values. You can specify the `generator` parameter in this function to use a different `numpy.random.Generator` if you want more control over the algorithm used to shuffle the dataset.
@@ -54,6 +58,17 @@ The [`~Dataset.shuffle`] function randomly rearranges the column values. You can
 [1, 1, 1, 0, 1, 1, 1, 1, 1, 0]
 ```
 
+Shuffling takes the list of indices `[0:len(my_dataset)]` and shuffles it to create an indices mapping.
+However as soon as your [`Dataset`] has an indices mapping, the speed can become 10x slower.
+This is because there is an extra step to get the row index to read using the indices mapping, and most importantly, you aren't reading contiguous chunks of data anymore.
+To restore the speed, you'd need to rewrite the entire dataset on your disk again using [`Dataset.flatten_indices`], which removes the indices mapping.
+Alternatively, you can switch to an [`IterableDataset`] and leverage its fast approximate shuffling [`IterableDataset.shuffle`]:
+
+```py
+>>> iterable_dataset = dataset.to_iterable_dataset(num_shards=128)
+>>> shuffled_iterable_dataset = iterable_dataset.shuffle(seed=42, buffer_size=1000)
+```
+
 ### Select and Filter
 
 There are two options for filtering rows in a dataset: [`~Dataset.select`] and [`~Dataset.filter`].
@@ -92,6 +107,8 @@ There are two options for filtering rows in a dataset: [`~Dataset.select`] and [
 1834.0
 ```
 
+Unless the list of indices to keep is contiguous, those methods also create an indices mapping under the .
+
 ### Split
 
 The [`~Dataset.train_test_split`] function creates train and test splits if your dataset doesn't already have them. This allows you to adjust the relative proportions or an absolute number of samples in each split. In the example below, use the `test_size` parameter to create a test split that is 10% of the original dataset:
diff --git a/docs/source/use_with_pytorch.mdx b/docs/source/use_with_pytorch.mdx
index 03cb9126fd2..cd5884e60cf 100644
--- a/docs/source/use_with_pytorch.mdx
+++ b/docs/source/use_with_pytorch.mdx
@@ -200,7 +200,10 @@ You must use a `BatchSampler` if you want the transform to be given full batches
 
 ### Stream data
 
-Loading a dataset in streaming mode allows one to iterate over the dataset without downloading it on disk.
+Stream a dataset by loading it as an [`IterableDataset`]. This allows you to progressively iterate over a dataset without downloading it on disk as well as any local dataset files you have stored.
+Learn more about which type of dataset is best for your use case in the (choosing between a regular dataset or an iterable dataset)[./dataset_vs_iterable_dataset] guide.
+
+
 An iterable dataset from `datasets` inherits from `torch.utils.data.IterableDataset` so you can pass it to a `torch.utils.data.DataLoader`:
 
 ```py
@@ -209,17 +212,17 @@ An iterable dataset from `datasets` inherits from `torch.utils.data.IterableData
 >>> from torch.utils.data import DataLoader
 >>> data = np.random.rand(10_000)
 >>> Dataset.from_dict({"data": data}).push_to_hub("<username>/my_dataset")  # Upload to the Hugging Face Hub
->>> ds = load_dataset("<username>/my_dataset", streaming=True, split="train")
->>> dataloader = DataLoader(ds, batch_size=32)
+>>> my_iterable_dataset = load_dataset("<username>/my_dataset", streaming=True, split="train")
+>>> dataloader = DataLoader(my_iterable_dataset, batch_size=32)
 ```
 
 If the dataset is split in several shards (i.e. if the dataset consists of multiple data files), then you can stream in parallel using `num_workers`:
 
 ```py
->>> ds = load_dataset("c4", "en", streaming=True, split="train")
->>> ds.n_shards
+>>> my_iterable_dataset = load_dataset("c4", "en", streaming=True, split="train")
+>>> my_iterable_dataset.n_shards
 1024
->>> dataloader = DataLoader(ds, batch_size=32, num_workers=4)
+>>> dataloader = DataLoader(my_iterable_dataset, batch_size=32, num_workers=4)
 ```
 
 In this case each worker is given a subset of the list of shards to stream from.
diff --git a/src/datasets/arrow_dataset.py b/src/datasets/arrow_dataset.py
index 438d6de8bea..5aa9b9a9a1b 100644
--- a/src/datasets/arrow_dataset.py
+++ b/src/datasets/arrow_dataset.py
@@ -121,6 +121,7 @@
     import sqlalchemy
 
     from .dataset_dict import DatasetDict
+    from .iterable_dataset import IterableDataset
 
 logger = logging.get_logger(__name__)
 
@@ -3848,6 +3849,34 @@ def shuffle(
         Currently shuffling uses numpy random generators.
         You can either supply a NumPy BitGenerator to use, or a seed to initiate NumPy's default random generator (PCG64).
 
+        Shuffling takes the list of indices `[0:len(my_dataset)]` and shuffles it to create an indices mapping.
+        However as soon as your [`Dataset`] has an indices mapping, the speed can become 10x slower.
+        This is because there is an extra step to get the row index to read using the indices mapping, and most importantly, you aren't reading contiguous chunks of data anymore.
+        To restore the speed, you'd need to rewrite the entire dataset on your disk again using [`Dataset.flatten_indices`], which removes the indices mapping.
+        This may take a lot of time depending of the size of your dataset though:
+
+        ```python
+        my_dataset[0]  # fast
+        my_dataset = my_dataset.shuffle(seed=42)
+        my_dataset[0]  # up to 10x slower
+        my_dataset = my_dataset.flatten_indices()  # rewrite the shuffled dataset on disk as contiguous chunks of data
+        my_dataset[0]  # fast again
+        ```
+
+        In this case, we recommend switching to an [`IterableDataset`] and leveraging its fast approximate shuffling method [`IterableDataset.shuffle`].
+        It only shuffles the shards order and adds a shuffle buffer to your dataset, which keeps the speed of your dataset optimal:
+
+        ```python
+        my_iterable_dataset = my_dataset.to_iterable_dataset(num_shards=128)
+        for example in enumerate(my_iterable_dataset):  # fast
+            pass
+
+        shuffled_iterable_dataset = my_iterable_dataset.shuffle(seed=42, buffer_size=100)
+
+        for example in enumerate(shuffled_iterable_dataset):  # as fast as before
+            pass
+        ```
+
         Args:
             seed (`int`, *optional*):
                 A seed to initialize the default BitGenerator if `generator=None`.
@@ -4643,6 +4672,120 @@ def extra_nbytes_visitor(array, feature):
             dataset_nbytes = dataset_nbytes * len(self._indices) / len(self.data)
         return dataset_nbytes
 
+    @staticmethod
+    def _iter_shards(shards: List["Dataset"]):
+        for shard in shards:
+            yield from shard
+
+    def to_iterable_dataset(self, num_shards: Optional[int] = 1) -> "IterableDataset":
+        """Get an [`datasets.IterableDataset`] from a map-style [`datasets.Dataset`].
+        This is equivalent to loading a dataset in streaming mode with [`datasets.load_dataset`], but much faster since the data is streamed from local files.
+
+        Contrary to map-style datasets, iterable datasets are lazy and can only be iterated over (e.g. using a for loop).
+        Since they are read sequentially in training loops, iterable datasets are much faster than map-style datasets.
+        All the transformations applied to iterable datasets like filtering or processing are done on-the-fly when you start iterating over the dataset.
+
+        Still, it is possible to shuffle an iterable dataset using [`datasets.IterableDataset.shuffle`].
+        This is a fast approximate shuffling that works best if you have multiple shards and if you specify a buffer size that is big enough.
+
+        To get the best speed performance, make sure your dataset doesn't have an indices mapping.
+        If this is the case, the data are not read contiguously, which can be slow sometimes.
+        You can use `ds = ds.flatten_indices()` to write your dataset in contiguous chunks of data and have optimal speed before switching to an iterable dataset.
+
+        Args:
+            num_shards (`int`, default to `1`):
+                Number of shards to define when instantiating the iterable dataset. This is especially useful for big datasets to be able to shuffle properly,
+                and also to enable fast parallel loading using a PyTorch DataLoader or in distributed setups for example.
+                Shards are defined using [`datasets.Dataset.shard`]: it simply slices the data without writing anything on disk.
+
+        Returns:
+            [`datasets.IterableDataset`]
+
+        Example:
+
+        Basic usage:
+        ```python
+        >>> ids = ds.to_iterable_dataset()
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With lazy filtering and processing:
+        ```python
+        >>> ids = ds.to_iterable_dataset()
+        >>> ids = ids.filter(filter_fn).map(process_fn)  # will filter and process on-the-fly when you start iterating over the iterable dataset
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With sharding to enable efficient shuffling:
+        ```python
+        >>> ids = ds.to_iterable_dataset(num_shards=64)  # the dataset is split into 64 shards to be iterated over
+        >>> ids = ids.shuffle(buffer_size=10_000)  # will shuffle the shards order and use a shuffle buffer for fast approximate shuffling when you start iterating
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With a PyTorch DataLoader:
+        ```python
+        >>> import torch
+        >>> ids = ds.to_iterable_dataset(num_shards=64)
+        >>> ids = ids.filter(filter_fn).map(process_fn)
+        >>> dataloader = torch.utils.data.DataLoader(ids, num_workers=4)  # will assign 64 / 4 = 16 shards to each worker to load, filter and process when you start iterating
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With a PyTorch DataLoader and shuffling:
+        ```python
+        >>> import torch
+        >>> ids = ds.to_iterable_dataset(num_shards=64)
+        >>> ids = ids.shuffle(buffer_size=10_000)  # will shuffle the shards order and use a shuffle buffer when you start iterating
+        >>> dataloader = torch.utils.data.DataLoader(ids, num_workers=4)  # will assign 64 / 4 = 16 shards from the shuffled list of shards to each worker when you start iterating
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        In a distributed setup like PyTorch DDP with a PyTorch DataLoader and shuffling
+        ```python
+        >>> from datasets.distributed import split_dataset_by_node
+        >>> ids = ds.to_iterable_dataset(num_shards=512)
+        >>> ids = ids.shuffle(buffer_size=10_000)  # will shuffle the shards order and use a shuffle buffer when you start iterating
+        >>> ids = split_dataset_by_node(ds, world_size=8, rank=0)  # will keep only 512 / 8 = 64 shards from the shuffled lists of shards when you start iterating
+        >>> dataloader = torch.utils.data.DataLoader(ids, num_workers=4)  # will assign 64 / 4 = 16 shards from this node's list of shards to each worker when you start iterating
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With shuffling and multiple epochs:
+        ```python
+        >>> ids = ds.to_iterable_dataset(num_shards=64)
+        >>> ids = ids.shuffle(buffer_size=10_000, seed=42)  # will shuffle the shards order and use a shuffle buffer when you start iterating
+        >>> for epoch in range(n_epochs):
+        ...     ids.set_epoch(epoch)  # will use effective_seed = seed + epoch to shuffle the shards and for the shuffle buffer when you start iterating
+        ...     for example in ids:
+        ...         pass
+        ```
+        Feel free to also use [`IterableDataset.set_epoch`] when using a PyTorch DataLoader or in distributed setups.
+        """
+        from .iterable_dataset import IterableDataset
+
+        if num_shards > len(self):
+            raise ValueError(f"Unable to shard a dataset of size {len(self)} into {num_shards} shards.")
+        if self._indices is not None:
+            logger.info(
+                "Converting an Arrow dataset to iterable but it has an indices mapping that can make it slower. "
+                "You can use `ds = ds.flatten_indices()` to write your dataset in contiguous chunks of data and have optimal speed."
+            )
+        shards = (
+            [copy.deepcopy(self)]
+            if num_shards == 1
+            else [
+                self.shard(num_shards=num_shards, index=shard_idx, contiguous=True) for shard_idx in range(num_shards)
+            ]
+        )
+        return IterableDataset.from_generator(Dataset._iter_shards, gen_kwargs={"shards": shards})
+
     def _push_parquet_shards_to_hub(
         self,
         repo_id: str,
diff --git a/src/datasets/builder.py b/src/datasets/builder.py
index 80d05092705..c5660eab849 100644
--- a/src/datasets/builder.py
+++ b/src/datasets/builder.py
@@ -493,7 +493,7 @@ def _create_builder_config(
         )
         is_custom = (config_id not in self.builder_configs) and config_id != "default"
         if is_custom:
-            logger.warning(f"Using custom data configuration {config_id}")
+            logger.info(f"Using custom data configuration {config_id}")
         else:
             if (
                 builder_config.name in self.builder_configs
diff --git a/tests/test_arrow_dataset.py b/tests/test_arrow_dataset.py
index 332f4d87ffa..6b2cccdc85f 100644
--- a/tests/test_arrow_dataset.py
+++ b/tests/test_arrow_dataset.py
@@ -37,6 +37,7 @@
 )
 from datasets.filesystems import extract_path_from_uri
 from datasets.info import DatasetInfo
+from datasets.iterable_dataset import IterableDataset
 from datasets.splits import NamedSplit
 from datasets.table import ConcatenationTable, InMemoryTable, MemoryMappedTable
 from datasets.tasks import (
@@ -4190,6 +4191,18 @@ def test_dataset_estimate_nbytes():
     assert 0.9 * ds._estimate_nbytes() < 100 * 100, "must be smaller than one chunk"
 
 
+def test_dataset_to_iterable_dataset(dataset):
+    iterable_dataset = dataset.to_iterable_dataset()
+    assert isinstance(iterable_dataset, IterableDataset)
+    assert list(iterable_dataset) == list(dataset)
+    iterable_dataset = dataset.to_iterable_dataset(num_shards=3)
+    assert isinstance(iterable_dataset, IterableDataset)
+    assert list(iterable_dataset) == list(dataset)
+    assert iterable_dataset.n_shards == 3
+    with pytest.raises(ValueError):
+        dataset.to_iterable_dataset(num_shards=len(dataset) + 1)
+
+
 @pytest.mark.parametrize("return_lazy_dict", [True, False, "mix"])
 def test_map_cases(return_lazy_dict):
     def f(x):


Code review comments:
diff_hunk: @@ -4643,6 +4672,120 @@ def extra_nbytes_visitor(array, feature):
             dataset_nbytes = dataset_nbytes * len(self._indices) / len(self.data)
         return dataset_nbytes
 
+    @staticmethod
+    def _iter_shards(shards: List["Dataset"]):
+        for shard in shards:
+            yield from shard
+
+    def to_iterable_dataset(self, num_shards: Optional[int] = 1) -> "IterableDataset":
+        """Get an [`datasets.IterableDataset`] from a map-style [`datasets.Dataset`].
+        This is equivalent to loading a dataset in streaming mode with [`datasets.load_dataset`], but much faster since the data is streamed from local files.
+
+        Contrary to map-style datasets, iterable datasets are lazy and can only be iterated over (e.g. using a for loop).
+        Since they are read sequentially in training loops, iterable datasets are much faster than map-style datasets.
+        All the transformations applied to iterable datasets like filtering or processing are done on-the-fly when you start iterating over the dataset.
+
+        Still, it is possible to shuffle an iterable dataset using [`datasets.IterableDataset.shuffle`].
+        This is a fast approximate shuffling that works best if you have multiple shards and if you specify a buffer size that is big enough.
+
+        To get the best speed performance, make sure your dataset doesn't have an indices mapping.
+        If this is the case, the data are not read contiguously, which can be slow sometimes.
+        You can use `ds = ds.flatten_indices()` to write your dataset in contiguous chunks of data and have optimal speed before switching to an iterable dataset.
+
+        Args:
+            num_shards (`int`, default to `1`):
+                Number of shards to define when instantiating the iterable dataset. This is especially useful for big datasets to be able to shuffle properly,
+                and also to enable fast parallel loading using a PyTorch DataLoader or in distributed setups for example.
+                Shards are defined using [`datasets.Dataset.shard`]: it simply slices the data without writing anything on disk.
+
+        Returns:
+            [`datasets.IterableDataset`]
+
+        Example:
+
+        Basic usage:
+        ```python
+        >>> ids = ds.to_iterable_dataset()
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With lazy filtering and processing:
+        ```python
+        >>> ids = ds.to_iterable_dataset()
+        >>> ids = ids.filter(filter_fn).map(process_fn)  # will filter and process on-the-fly when you start iterating over the iterable dataset
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With sharding to enable efficient shuffling:
+        ```python
+        >>> ids = ds.to_iterable_dataset(num_shards=64)  # the dataset is split into 64 shards to be iterated over
+        >>> ids = ids.shuffle(buffer_size=10_000)  # will shuffle the shards order and use a shuffle buffer for fast approximate shuffling when you start iterating
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With a PyTorch DataLoader:
+        ```python
+        >>> import torch
+        >>> ids = ds.to_iterable_dataset(num_shards=64)
+        >>> ids = ids.filter(filter_fn).map(process_fn)
+        >>> dataloader = torch.utils.data.DataLoader(ids, num_workers=4)  # will assign 64 / 4 = 16 shards to each worker to load, filter and process when you start iterating
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With a PyTorch DataLoader and shuffling:
+        ```python
+        >>> import torch
+        >>> ids = ds.to_iterable_dataset(num_shards=64)
+        >>> ids = ids.shuffle(buffer_size=10_000)  # will shuffle the shards order and use a shuffle buffer when you start iterating
+        >>> dataloader = torch.utils.data.DataLoader(ids, num_workers=4)  # will assign 64 / 4 = 16 shards from the shuffled list of shards to each worker when you start iterating
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        In a distributed setup like PyTorch DDP with a PyTorch DataLoader and shuffling
+        ```python
+        >>> from datasets.distributed import split_dataset_by_node
+        >>> ids = ds.to_iterable_dataset(num_shards=512)
+        >>> ids = ids.shuffle(buffer_size=10_000)  # will shuffle the shards order and use a shuffle buffer when you start iterating
+        >>> ids = split_dataset_by_node(ds, world_size=8, rank=0)  # will keep only 512 / 8 = 64 shards from the shuffled lists of shards when you start iterating
+        >>> dataloader = torch.utils.data.DataLoader(ids, num_workers=4)  # will assign 64 / 4 = 16 shards from this node's list of shards to each worker when you start iterating
+        >>> for example in ids:
+        ...     pass
+        ```
+
+        With shuffling and multiple epochs:
+        ```python
+        >>> ids = ds.to_iterable_dataset(num_shards=64)
+        >>> ids = ids.shuffle(buffer_size=10_000, seed=42)  # will shuffle the shards order and use a shuffle buffer when you start iterating
+        >>> for epoch in range(n_epochs):
+        ...     ids.set_epoch(epoch)  # will use effective_seed = seed + epoch to shuffle the shards and for the shuffle buffer when you start iterating
+        ...     for example in ids:
+        ...         pass
+        ```
+        Feel free to also use [`IterableDataset.set_epoch`] when using a PyTorch DataLoader or in distributed setups.
+        """
+        from .iterable_dataset import IterableDataset
+
+        if num_shards > len(self):
+            raise ValueError(f"Unable to shard a dataset of size {len(self)} into {num_shards} shards.")
path: src/datasets/arrow_dataset.py
text: @user1:
```suggestion
            raise ValueError(f"Unable to shard a dataset of size {len(self)} into {num_shards} shards (the number of shards exceeds the number of samples).")
```
or rows?