| 72 | super().__init__(urls, nshards, worker_seed_sat_this, deterministic) |
| 73 | |
| 74 | class SimpleDistributedWebDataset(DataPipeline): |
| 75 | def __init__(self, path, process_fn, seed, *, shuffle_buffer=1000): |
| 76 | # set shuffle_buffer = 1 to disable it, model-parallel will be different due to shuffle |
| 77 | try: |
| 78 | from sat.mpu import get_model_parallel_world_size |
| 79 | if get_model_parallel_world_size() > 1: |
| 80 | shuffle_buffer = 1 |
| 81 | except Exception: |
| 82 | pass |
| 83 | super().__init__( |
| 84 | ConfiguredResampledShards(path, seed), # Lots of shards are recommended, or not evenly |
| 85 | tarfile_to_samples(), |
| 86 | wds.shuffle(shuffle_buffer), |
| 87 | process_fn |
| 88 | ) |
| 89 | |
| 90 | def tar_file_iterator_with_meta(fileobj, meta_names, skip_meta=r"__[^/]*__($|/)", suffix=None,handler=reraise_exception, meta_stream=None): |
| 91 | """Iterate over tar file, yielding filename, content pairs for the given tar stream. |
no outgoing calls