IterableDataset returns duplicated data using PyTorch DDP
NONE
If you use huggingface trainer, you will find the trainer has wrapped a `IterableDatasetShard` to avoid duplication.
See:
https://github.com/huggingface/transformers/blob/dfd818420dcbad68e05a502495cf666d338b2bfb/src/transformers/trainer.py#L835

2022-12-14T16:10:06Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1351702833
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5QkVkx
https://api.github.com/repos/huggingface/datasets/issues/comments/1351702833/reactions
2022-12-14T16:10:06Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1351702833
https://avatars.githubusercontent.com/u/9592150?v=4
https://api.github.com/users/Ethan-yt/events{/privacy}
https://api.github.com/users/Ethan-yt/followers
https://api.github.com/users/Ethan-yt/following{/other_user}
https://api.github.com/users/Ethan-yt/gists{/gist_id}
https://github.com/Ethan-yt
Ethan-yt
MDQ6VXNlcjk1OTIxNTA=
https://api.github.com/users/Ethan-yt/orgs
https://api.github.com/users/Ethan-yt/received_events
https://api.github.com/users/Ethan-yt/repos
https://api.github.com/users/Ethan-yt/starred{/owner}{/repo}
https://api.github.com/users/Ethan-yt/subscriptions
User
https://api.github.com/users/Ethan-yt
public
NONE
If you want to support it by datasets natively, maybe we also need to change the code in `transformers` ?
2022-12-14T16:13:15Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1351708705
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5QkXAh
https://api.github.com/repos/huggingface/datasets/issues/comments/1351708705/reactions
2022-12-14T16:13:15Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1351708705
https://avatars.githubusercontent.com/u/9592150?v=4
https://api.github.com/users/Ethan-yt/events{/privacy}
https://api.github.com/users/Ethan-yt/followers
https://api.github.com/users/Ethan-yt/following{/other_user}
https://api.github.com/users/Ethan-yt/gists{/gist_id}
https://github.com/Ethan-yt
Ethan-yt
MDQ6VXNlcjk1OTIxNTA=
https://api.github.com/users/Ethan-yt/orgs
https://api.github.com/users/Ethan-yt/received_events
https://api.github.com/users/Ethan-yt/repos
https://api.github.com/users/Ethan-yt/starred{/owner}{/repo}
https://api.github.com/users/Ethan-yt/subscriptions
User
https://api.github.com/users/Ethan-yt
public
MEMBER
Opened https://github.com/huggingface/transformers/issues/20770 to discuss this :)
2022-12-14T16:56:11Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1351767433
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5QklWJ
https://api.github.com/repos/huggingface/datasets/issues/comments/1351767433/reactions
2022-12-14T16:56:11Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1351767433
https://avatars.githubusercontent.com/u/42851186?v=4
https://api.github.com/users/lhoestq/events{/privacy}
https://api.github.com/users/lhoestq/followers
https://api.github.com/users/lhoestq/following{/other_user}
https://api.github.com/users/lhoestq/gists{/gist_id}
https://github.com/lhoestq
lhoestq
MDQ6VXNlcjQyODUxMTg2
https://api.github.com/users/lhoestq/orgs
https://api.github.com/users/lhoestq/received_events
https://api.github.com/users/lhoestq/repos
https://api.github.com/users/lhoestq/starred{/owner}{/repo}
https://api.github.com/users/lhoestq/subscriptions
User
https://api.github.com/users/lhoestq
public
MEMBER
Maybe something like this then ?
```python
from datasets.distributed import split_dataset_by_node
ds = split_dataset_by_node(ds, rank=rank, world_size=world_size)
```

For map-style datasets the implementation is trivial (it can simply use `.shard()`).

For iterable datasets we would need to implement a new ExamplesIterable that would only iterate on a subset of the (possibly shuffled and re-shuffled after each epoch) list of shards, based on the rank and world size.
2022-12-14T18:19:55Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1351904257
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5QlGwB
https://api.github.com/repos/huggingface/datasets/issues/comments/1351904257/reactions
2022-12-14T18:20:05Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1351904257
https://avatars.githubusercontent.com/u/42851186?v=4
https://api.github.com/users/lhoestq/events{/privacy}
https://api.github.com/users/lhoestq/followers
https://api.github.com/users/lhoestq/following{/other_user}
https://api.github.com/users/lhoestq/gists{/gist_id}
https://github.com/lhoestq
lhoestq
MDQ6VXNlcjQyODUxMTg2
https://api.github.com/users/lhoestq/orgs
https://api.github.com/users/lhoestq/received_events
https://api.github.com/users/lhoestq/repos
https://api.github.com/users/lhoestq/starred{/owner}{/repo}
https://api.github.com/users/lhoestq/subscriptions
User
https://api.github.com/users/lhoestq
public
MEMBER
My plan is to skip examples by default to not end up with duplicates.

And if a dataset has a number of shards that is a factor of the world size, then I'd make it more optimized by distributing the shards evenly across nodes instead.
2022-12-15T17:13:41Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1353424178
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5Qq50y
https://api.github.com/repos/huggingface/datasets/issues/comments/1353424178/reactions
2022-12-15T17:13:41Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1353424178
https://avatars.githubusercontent.com/u/42851186?v=4
https://api.github.com/users/lhoestq/events{/privacy}
https://api.github.com/users/lhoestq/followers
https://api.github.com/users/lhoestq/following{/other_user}
https://api.github.com/users/lhoestq/gists{/gist_id}
https://github.com/lhoestq
lhoestq
MDQ6VXNlcjQyODUxMTg2
https://api.github.com/users/lhoestq/orgs
https://api.github.com/users/lhoestq/received_events
https://api.github.com/users/lhoestq/repos
https://api.github.com/users/lhoestq/starred{/owner}{/repo}
https://api.github.com/users/lhoestq/subscriptions
User
https://api.github.com/users/lhoestq
public
MEMBER
Opened a PR here: https://github.com/huggingface/datasets/pull/5369

feel free to play with it and share your feedbacks :)
2022-12-16T17:47:20Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1355302471
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5QyEZH
https://api.github.com/repos/huggingface/datasets/issues/comments/1355302471/reactions
2022-12-16T17:47:20Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1355302471
https://avatars.githubusercontent.com/u/42851186?v=4
https://api.github.com/users/lhoestq/events{/privacy}
https://api.github.com/users/lhoestq/followers
https://api.github.com/users/lhoestq/following{/other_user}
https://api.github.com/users/lhoestq/gists{/gist_id}
https://github.com/lhoestq
lhoestq
MDQ6VXNlcjQyODUxMTg2
https://api.github.com/users/lhoestq/orgs
https://api.github.com/users/lhoestq/received_events
https://api.github.com/users/lhoestq/repos
https://api.github.com/users/lhoestq/starred{/owner}{/repo}
https://api.github.com/users/lhoestq/subscriptions
User
https://api.github.com/users/lhoestq
public
NONE
@lhoestq I add shuffle after split_dataset_by_node, duplicated data still exist. 
For example, we have a directory named `mock_pretraining_data`, which has three files, `part-00000`, `part-00002`,`part-00002`. 
Text in `part-00000` is like this: 
{"id": 0}
{"id": 1}
{"id": 2}
{"id": 3}
{"id": 4}
{"id": 5}
{"id": 6}
{"id": 7}
{"id": 8}
{"id": 9}

and `part-00001`
{"id": 10}
{"id": 11}
{"id": 12}
{"id": 13}
{"id": 14}
{"id": 15}
{"id": 16}
{"id": 17}
{"id": 18}
{"id": 19}

and `part-00002`
{"id": 20}
{"id": 21}
{"id": 22}
{"id": 23}
{"id": 24}
{"id": 25}
{"id": 26}
{"id": 27}
{"id": 28}
{"id": 29}

And code in `test_dist.py` like this,
```python
import torch
from torch.utils.data import Dataset, DataLoader
from datasets import load_dataset
import os
from transformers import AutoTokenizer, NezhaForPreTraining
from transformers import AdamW, get_linear_schedule_with_warmup
import torch.nn.functional as F
import torch.nn as nn
import torch.distributed as dist
from datasets.distributed import split_dataset_by_node
from torch.nn.parallel import DistributedDataParallel as DDP

os.environ["CUDA_VISIBLE_DEVICES"] = '5,6,7'

dist.init_process_group("nccl")
local_rank = int(os.environ['LOCAL_RANK'])
world_size = torch.distributed.get_world_size()
device = torch.device('cuda', local_rank)
data_dir = './'

def load_trainset(train_path):
    dataset = load_dataset('json', data_dir=os.path.join(data_dir, train_path), split='train', streaming=True)
    return dataset

def collate_fn(examples):
    input_ids = []
    for example in examples:
        input_ids.append(example['id'])
    return torch.LongTensor(input_ids).to(device)


dataset = load_trainset('mock_pretraining_data')
dataset = split_dataset_by_node(dataset, rank=local_rank, world_size=world_size).shuffle(buffer_size=512)
# train_sampler = torch.utils.data.distributed.DistributedSampler(dataset)
batch_size = 3
print('batch_size: {}'.format(batch_size))
train_dataloader = DataLoader(dataset, batch_size=batch_size, collate_fn=collate_fn)

for x in train_dataloader:
    print({'rank': local_rank, 'id': x})
```
run  `python -m torch.distributed.launch --nproc_per_node=3 test_dist.py`
The output is
```
{'rank': 1, 'id': tensor([12, 15, 14], device='cuda:1')}
{'rank': 1, 'id': tensor([16, 10, 18], device='cuda:1')}
{'rank': 1, 'id': tensor([17, 13, 19], device='cuda:1')}
{'rank': 1, 'id': tensor([11], device='cuda:1')}
{'rank': 0, 'id': tensor([0, 2, 9], device='cuda:0')}
{'rank': 0, 'id': tensor([4, 8, 1], device='cuda:0')}
{'rank': 0, 'id': tensor([5, 3, 6], device='cuda:0')}
{'rank': 0, 'id': tensor([7], device='cuda:0')}
{'rank': 2, 'id': tensor([13, 15, 14], device='cuda:2')}
{'rank': 2, 'id': tensor([19, 17, 18], device='cuda:2')}
{'rank': 2, 'id': tensor([12, 16, 11], device='cuda:2')}
{'rank': 2, 'id': tensor([10], device='cuda:2')}
```
`part-00001` is loaded twice, `part-00002` isn't loaded.

If I run `python -m torch.distributed.launch --nproc_per_node=2 test_dist.py`
The output is weirder，many numbers appear twice
```
{'rank': 1, 'id': tensor([26,  8, 13], device='cuda:1')}
{'rank': 1, 'id': tensor([22, 19, 20], device='cuda:1')}
{'rank': 1, 'id': tensor([12, 28, 11], device='cuda:1')}
{'rank': 1, 'id': tensor([24,  2, 14], device='cuda:1')}
{'rank': 1, 'id': tensor([ 6, 27,  3], device='cuda:1')}
{'rank': 0, 'id': tensor([ 8, 25,  1], device='cuda:0')}
{'rank': 0, 'id': tensor([20,  4, 12], device='cuda:0')}
{'rank': 0, 'id': tensor([14, 29,  5], device='cuda:0')}
{'rank': 0, 'id': tensor([ 7, 18, 23], device='cuda:0')}
{'rank': 0, 'id': tensor([19, 17, 11], device='cuda:0')}
``` 
2023-03-22T09:55:02Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1479241561
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5YK29Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1479241561/reactions
2023-03-22T10:00:54Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1479241561
https://avatars.githubusercontent.com/u/11674304?v=4
https://api.github.com/users/gauss-clb/events{/privacy}
https://api.github.com/users/gauss-clb/followers
https://api.github.com/users/gauss-clb/following{/other_user}
https://api.github.com/users/gauss-clb/gists{/gist_id}
https://github.com/gauss-clb
gauss-clb
MDQ6VXNlcjExNjc0MzA0
https://api.github.com/users/gauss-clb/orgs
https://api.github.com/users/gauss-clb/received_events
https://api.github.com/users/gauss-clb/repos
https://api.github.com/users/gauss-clb/starred{/owner}{/repo}
https://api.github.com/users/gauss-clb/subscriptions
User
https://api.github.com/users/gauss-clb
public
MEMBER
Hi ! Thanks for reporting, you need to pass `seed=` to `shuffle()` or the processes won't use the same seed to shuffle the shards order before assigning each shard to a node.

The issue is that the workers are not using the same seed to shuffle the shards before splitting the shards list by node.
2023-04-03T09:36:34Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1493998041
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5ZDJnZ
https://api.github.com/repos/huggingface/datasets/issues/comments/1493998041/reactions
2023-04-03T10:21:52Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1493998041
https://avatars.githubusercontent.com/u/42851186?v=4
https://api.github.com/users/lhoestq/events{/privacy}
https://api.github.com/users/lhoestq/followers
https://api.github.com/users/lhoestq/following{/other_user}
https://api.github.com/users/lhoestq/gists{/gist_id}
https://github.com/lhoestq
lhoestq
MDQ6VXNlcjQyODUxMTg2
https://api.github.com/users/lhoestq/orgs
https://api.github.com/users/lhoestq/received_events
https://api.github.com/users/lhoestq/repos
https://api.github.com/users/lhoestq/starred{/owner}{/repo}
https://api.github.com/users/lhoestq/subscriptions
User
https://api.github.com/users/lhoestq
public
MEMBER
Opened https://github.com/huggingface/datasets/issues/5696
2023-04-03T09:40:19Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1494003395
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5ZDK7D
https://api.github.com/repos/huggingface/datasets/issues/comments/1494003395/reactions
2023-04-03T09:40:19Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1494003395
https://avatars.githubusercontent.com/u/42851186?v=4
https://api.github.com/users/lhoestq/events{/privacy}
https://api.github.com/users/lhoestq/followers
https://api.github.com/users/lhoestq/following{/other_user}
https://api.github.com/users/lhoestq/gists{/gist_id}
https://github.com/lhoestq
lhoestq
MDQ6VXNlcjQyODUxMTg2
https://api.github.com/users/lhoestq/orgs
https://api.github.com/users/lhoestq/received_events
https://api.github.com/users/lhoestq/repos
https://api.github.com/users/lhoestq/starred{/owner}{/repo}
https://api.github.com/users/lhoestq/subscriptions
User
https://api.github.com/users/lhoestq
public
NONE
I have the same issue
```
ds['train'] = load_dataset(streaming=True)
ds['train'] = split_dataset_by_node(ds['train'], rank=int(os.environ["RANK"]), world_size=int(os.environ["WORLD_SIZE"]))
vectorized_datasets = ds.map(
    prepare_dataset,
    remove_columns=raw_datasets_features,
).with_format("torch")

vectorized_datasets["train"] = vectorized_datasets["train"].shuffle(
    buffer_size=500,
    seed=42,
)

def prepare_dataset(batch):
  ....
  print(f"sentence: {batch['sentence']}, target_text: {batch['target_text']}")
  return batch
```
When using split_dataset_by_node(), the data being read is indeed different for each GPU ID.

```
trainer = Trainer(
        model=model,
        data_collator=data_collator,
        args=training_args,
        compute_metrics=compute_metrics,
        train_dataset=vectorized_datasets["train"] if training_args.do_train else None,
        eval_dataset=vectorized_datasets["eval"] if training_args.do_eval else None,
        tokenizer=processor,
        callbacks=[ShuffleCallback()],
    )
...
train_result = trainer.train(resume_from_checkpoint=checkpoint)
```
However, when I execute trainer.train(), the data being read is different from what I expected.
Because I print the batch value in prepare_dataset() , I observe that the data is the same for each GPU ID.

How should I handle this issue?



2023-06-15T09:42:38Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1592716858
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5e7u46
https://api.github.com/repos/huggingface/datasets/issues/comments/1592716858/reactions
2023-06-15T09:42:38Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1592716858
https://avatars.githubusercontent.com/u/27708347?v=4
https://api.github.com/users/johnchienbronci/events{/privacy}
https://api.github.com/users/johnchienbronci/followers
https://api.github.com/users/johnchienbronci/following{/other_user}
https://api.github.com/users/johnchienbronci/gists{/gist_id}
https://github.com/johnchienbronci
johnchienbronci
MDQ6VXNlcjI3NzA4MzQ3
https://api.github.com/users/johnchienbronci/orgs
https://api.github.com/users/johnchienbronci/received_events
https://api.github.com/users/johnchienbronci/repos
https://api.github.com/users/johnchienbronci/starred{/owner}{/repo}
https://api.github.com/users/johnchienbronci/subscriptions
User
https://api.github.com/users/johnchienbronci
public
MEMBER
There are two ways an iterable dataset can be split by node:
1. if the number of shards is a factor of number of GPUs: in that case the shards are evenly distributed per GPU
2. otherwise, each GPU iterate on the data and at the end keeps 1 sample out of n(GPUs) - skipping the others.

In case 2. it's therefore possible to have the same examples passed to `prepare_dataset` for each GPU.

This doesn't sound optimized though, because it runs the preprocessing on samples that won't be used in the end.

Could you open a new issue so that we can discuss about this and find a solution ?
2023-06-15T09:51:13Z
https://github.com/huggingface/datasets/issues/5360#issuecomment-1592729051
https://api.github.com/repos/huggingface/datasets/issues/5360
IC_kwDODunzps5e7x3b
https://api.github.com/repos/huggingface/datasets/issues/comments/1592729051/reactions
2023-06-15T09:51:13Z
https://api.github.com/repos/huggingface/datasets/issues/comments/1592729051
https://avatars.githubusercontent.com/u/42851186?v=4
https://api.github.com/users/lhoestq/events{/privacy}
https://api.github.com/users/lhoestq/followers
https://api.github.com/users/lhoestq/following{/other_user}
https://api.github.com/users/lhoestq/gists{/gist_id}
https://github.com/lhoestq
lhoestq
MDQ6VXNlcjQyODUxMTg2
https://api.github.com/users/lhoestq/orgs
https://api.github.com/users/lhoestq/received_events
https://api.github.com/users/lhoestq/repos
https://api.github.com/users/lhoestq/starred{/owner}{/repo}
https://api.github.com/users/lhoestq/subscriptions
User
https://api.github.com/users/lhoestq
public
As mentioned in https://github.com/huggingface/datasets/issues/3423, when using PyTorch DDP the dataset ends up with duplicated data. We already check for the PyTorch `worker_info` for single node, but we should also check for `torch.distributed.get_world_size()` and `torch.distributed.get_rank()`