dataset_utils.py 27.7 KB
Newer Older
xingjinliang's avatar
xingjinliang committed
1
# Copyright (c) 2024, NVIDIA CORPORATION. All rights reserved.
Mohammad Shoeybi's avatar
Mohammad Shoeybi committed
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# coding=utf-8
# Copyright 2018 The Google AI Language Team Authors, and NVIDIA.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
16
17


18
19
20
21
# Most of the code here has been copied from:
#   https://github.com/google-research/albert/blob/master/create_pretraining_data.py
# with some modifications.

mohammad's avatar
mohammad committed
22
import math
23
import os
24
import time
25
import collections
Neel Kant's avatar
Neel Kant committed
26

27
import numpy as np
28
29
import torch

xingjinliang's avatar
xingjinliang committed
30
from megatron.training import (
31
32
33
    get_args,
    print_rank_0
)
34
from megatron.core import mpu
xingjinliang's avatar
xingjinliang committed
35
36
from megatron.core.datasets.indexed_dataset import IndexedDataset

37

38
DSET_TYPE_BERT = 'standard_bert'
39
DSET_TYPE_ICT = 'ict'
40
DSET_TYPE_T5  = 't5'
liangjing's avatar
v1  
liangjing committed
41
DSET_TYPE_MULTIMODAL = 'multimodal'
42

liangjing's avatar
v1  
liangjing committed
43
DSET_TYPES = [DSET_TYPE_BERT, DSET_TYPE_ICT, DSET_TYPE_T5, DSET_TYPE_MULTIMODAL]
44

45

mohammad's avatar
mohammad committed
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def get_datasets_weights_and_num_samples(data_prefix,
                                         train_valid_test_num_samples):

    # The data prefix should be in the format of:
    #   weight-1, data-prefix-1, weight-2, data-prefix-2, ..
    assert len(data_prefix) % 2 == 0
    num_datasets = len(data_prefix) // 2
    weights = [0]*num_datasets
    prefixes = [0]*num_datasets
    for i in range(num_datasets):
        weights[i] = float(data_prefix[2*i])
        prefixes[i] = (data_prefix[2*i+1]).strip()
    # Normalize weights
    weight_sum = 0.0
    for weight in weights:
        weight_sum += weight
    assert weight_sum > 0.0
    weights = [weight / weight_sum for weight in weights]

    # Add 0.5% (the 1.005 factor) so in case the bleding dataset does
    # not uniformly distribute the number of samples, we still have
    # samples left to feed to the network.
68
69
70
71
72
73
74
    if isinstance(train_valid_test_num_samples, list):
        datasets_train_valid_test_num_samples = []
        for weight in weights:
            datasets_train_valid_test_num_samples.append(
                [int(math.ceil(val * weight * 1.005))
                for val in train_valid_test_num_samples])
    else:
75
76
        # Used when separate dataset files are provided for train,
        # valid and test
77
78
79
        datasets_train_valid_test_num_samples = [
            int(math.ceil(train_valid_test_num_samples * weight * 1.005))
            for weight in weights]
mohammad's avatar
mohammad committed
80
81
82
83

    return prefixes, weights, datasets_train_valid_test_num_samples


84
def get_a_and_b_segments(sample, np_rng):
85
86
87
88
89
90
91
92
93
94
95
    """Divide sample into a and b segments."""

    # Number of sentences in the sample.
    n_sentences = len(sample)
    # Make sure we always have two sentences.
    assert n_sentences > 1, 'make sure each sample has at least two sentences.'

    # First part:
    # `a_end` is how many sentences go into the `A`.
    a_end = 1
    if n_sentences >= 3:
96
97
        # Note that randin in numpy is exclusive.
        a_end = np_rng.randint(1, n_sentences)
98
99
100
101
102
103
104
105
106
107
108
    tokens_a = []
    for j in range(a_end):
        tokens_a.extend(sample[j])

    # Second part:
    tokens_b = []
    for j in range(a_end, n_sentences):
        tokens_b.extend(sample[j])

    # Random next:
    is_next_random = False
109
    if np_rng.random() < 0.5:
110
111
112
113
114
115
        is_next_random = True
        tokens_a, tokens_b = tokens_b, tokens_a

    return tokens_a, tokens_b, is_next_random


116
def truncate_segments(tokens_a, tokens_b, len_a, len_b, max_num_tokens, np_rng):
117
    """Truncates a pair of sequences to a maximum sequence length."""
118
    #print(len_a, len_b, max_num_tokens)
119
    assert len_a > 0
120
121
122
    if len_a + len_b <= max_num_tokens:
        return False
    while len_a + len_b > max_num_tokens:
123
124
125
126
127
128
        if len_a > len_b:
            len_a -= 1
            tokens = tokens_a
        else:
            len_b -= 1
            tokens = tokens_b
129
        if np_rng.random() < 0.5:
130
131
132
            del tokens[0]
        else:
            tokens.pop()
133
    return True
134

Neel Kant's avatar
Neel Kant committed
135

136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def create_tokens_and_tokentypes(tokens_a, tokens_b, cls_id, sep_id):
    """Merge segments A and B, add [CLS] and [SEP] and build tokentypes."""

    tokens = []
    tokentypes = []
    # [CLS].
    tokens.append(cls_id)
    tokentypes.append(0)
    # Segment A.
    for token in tokens_a:
        tokens.append(token)
        tokentypes.append(0)
    # [SEP].
    tokens.append(sep_id)
    tokentypes.append(0)
    # Segment B.
    for token in tokens_b:
        tokens.append(token)
        tokentypes.append(1)
155
156
157
158
    if tokens_b:
        # [SEP].
        tokens.append(sep_id)
        tokentypes.append(1)
159

160
161
162
163
164
165
166
167
    return tokens, tokentypes


MaskedLmInstance = collections.namedtuple("MaskedLmInstance",
                                          ["index", "label"])


def is_start_piece(piece):
168
169
170
171
172
173
    """Check if the current word piece is the starting piece (BERT)."""
    # When a word has been split into
    # WordPieces, the first token does not have any marker and any subsequence
    # tokens are prefixed with ##. So whenever we see the ## token, we
    # append it to the previous set of word indexes.
    return not piece.startswith("##")
174
175
176
177
178
179
180


def create_masked_lm_predictions(tokens,
                                 vocab_id_list, vocab_id_to_token_dict,
                                 masked_lm_prob,
                                 cls_id, sep_id, mask_id,
                                 max_predictions_per_seq,
181
                                 np_rng,
182
183
184
                                 max_ngrams=3,
                                 do_whole_word_mask=True,
                                 favor_longer_ngram=False,
185
186
187
                                 do_permutation=False,
                                 geometric_dist=False,
                                 masking_style="bert"):
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
    """Creates the predictions for the masked LM objective.
    Note: Tokens here are vocab ids and not text tokens."""

    cand_indexes = []
    # Note(mingdachen): We create a list for recording if the piece is
    # the starting piece of current token, where 1 means true, so that
    # on-the-fly whole word masking is possible.
    token_boundary = [0] * len(tokens)

    for (i, token) in enumerate(tokens):
        if token == cls_id or token == sep_id:
            token_boundary[i] = 1
            continue
        # Whole Word Masking means that if we mask all of the wordpieces
        # corresponding to an original word.
        #
        # Note that Whole Word Masking does *not* change the training code
        # at all -- we still predict each WordPiece independently, softmaxed
        # over the entire vocabulary.
        if (do_whole_word_mask and len(cand_indexes) >= 1 and
                not is_start_piece(vocab_id_to_token_dict[token])):
            cand_indexes[-1].append(i)
210
        else:
211
212
213
            cand_indexes.append([i])
            if is_start_piece(vocab_id_to_token_dict[token]):
                token_boundary[i] = 1
214

215
    output_tokens = list(tokens)
216

217
218
    masked_lm_positions = []
    masked_lm_labels = []
219

220
221
222
    if masked_lm_prob == 0:
        return (output_tokens, masked_lm_positions,
                masked_lm_labels, token_boundary)
223

224
225
226
227
    num_to_predict = min(max_predictions_per_seq,
                         max(1, int(round(len(tokens) * masked_lm_prob))))

    ngrams = np.arange(1, max_ngrams + 1, dtype=np.int64)
228
229
230
231
232
233
234
    if not geometric_dist:
        # Note(mingdachen):
        # By default, we set the probilities to favor shorter ngram sequences.
        pvals = 1. / np.arange(1, max_ngrams + 1)
        pvals /= pvals.sum(keepdims=True)
        if favor_longer_ngram:
            pvals = pvals[::-1]
235

236
237
238
239
240
241
    ngram_indexes = []
    for idx in range(len(cand_indexes)):
        ngram_index = []
        for n in ngrams:
            ngram_index.append(cand_indexes[idx:idx + n])
        ngram_indexes.append(ngram_index)
242

243
    np_rng.shuffle(ngram_indexes)
244

245
    (masked_lms, masked_spans) = ([], [])
246
247
248
249
250
251
252
253
254
255
256
257
258
    covered_indexes = set()
    for cand_index_set in ngram_indexes:
        if len(masked_lms) >= num_to_predict:
            break
        if not cand_index_set:
            continue
        # Note(mingdachen):
        # Skip current piece if they are covered in lm masking or previous ngrams.
        for index_set in cand_index_set[0]:
            for index in index_set:
                if index in covered_indexes:
                    continue

259
260
261
262
263
264
265
266
267
268
        if not geometric_dist:
            n = np_rng.choice(ngrams[:len(cand_index_set)],
                              p=pvals[:len(cand_index_set)] /
                              pvals[:len(cand_index_set)].sum(keepdims=True))
        else:
            # Sampling "n" from the geometric distribution and clipping it to
            # the max_ngrams. Using p=0.2 default from the SpanBERT paper
            # https://arxiv.org/pdf/1907.10529.pdf (Sec 3.1)
            n = min(np_rng.geometric(0.2), max_ngrams)

269
270
        index_set = sum(cand_index_set[n - 1], [])
        n -= 1
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
        # Note(mingdachen):
        # Repeatedly looking for a candidate that does not exceed the
        # maximum number of predictions by trying shorter ngrams.
        while len(masked_lms) + len(index_set) > num_to_predict:
            if n == 0:
                break
            index_set = sum(cand_index_set[n - 1], [])
            n -= 1
        # If adding a whole-word mask would exceed the maximum number of
        # predictions, then just skip this candidate.
        if len(masked_lms) + len(index_set) > num_to_predict:
            continue
        is_any_index_covered = False
        for index in index_set:
            if index in covered_indexes:
                is_any_index_covered = True
                break
        if is_any_index_covered:
            continue
        for index in index_set:
            covered_indexes.add(index)
            masked_token = None
293
294
295
296
297
298
299
300
301
302
303
304
            if masking_style == "bert":
                # 80% of the time, replace with [MASK]
                if np_rng.random() < 0.8:
                    masked_token = mask_id
                else:
                    # 10% of the time, keep original
                    if np_rng.random() < 0.5:
                        masked_token = tokens[index]
                    # 10% of the time, replace with random word
                    else:
                        masked_token = vocab_id_list[np_rng.randint(0, len(vocab_id_list))]
            elif masking_style == "t5":
305
306
                masked_token = mask_id
            else:
307
                raise ValueError("invalid value of masking style")
308
309
310
311

            output_tokens[index] = masked_token
            masked_lms.append(MaskedLmInstance(index=index, label=tokens[index]))

312
313
314
315
316
        masked_spans.append(MaskedLmInstance(
            index=index_set,
            label=[tokens[index] for index in index_set]))

    assert len(masked_lms) <= num_to_predict
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
    np_rng.shuffle(ngram_indexes)

    select_indexes = set()
    if do_permutation:
        for cand_index_set in ngram_indexes:
            if len(select_indexes) >= num_to_predict:
                break
            if not cand_index_set:
                continue
            # Note(mingdachen):
            # Skip current piece if they are covered in lm masking or previous ngrams.
            for index_set in cand_index_set[0]:
                for index in index_set:
                    if index in covered_indexes or index in select_indexes:
                        continue

            n = np.random.choice(ngrams[:len(cand_index_set)],
                                 p=pvals[:len(cand_index_set)] /
                                 pvals[:len(cand_index_set)].sum(keepdims=True))
            index_set = sum(cand_index_set[n - 1], [])
            n -= 1

            while len(select_indexes) + len(index_set) > num_to_predict:
                if n == 0:
                    break
                index_set = sum(cand_index_set[n - 1], [])
                n -= 1
            # If adding a whole-word mask would exceed the maximum number of
            # predictions, then just skip this candidate.
            if len(select_indexes) + len(index_set) > num_to_predict:
                continue
            is_any_index_covered = False
            for index in index_set:
                if index in covered_indexes or index in select_indexes:
                    is_any_index_covered = True
                    break
            if is_any_index_covered:
                continue
            for index in index_set:
                select_indexes.add(index)
        assert len(select_indexes) <= num_to_predict

        select_indexes = sorted(select_indexes)
        permute_indexes = list(select_indexes)
        np_rng.shuffle(permute_indexes)
        orig_token = list(output_tokens)

        for src_i, tgt_i in zip(select_indexes, permute_indexes):
            output_tokens[src_i] = orig_token[tgt_i]
            masked_lms.append(MaskedLmInstance(index=src_i, label=orig_token[src_i]))

    masked_lms = sorted(masked_lms, key=lambda x: x.index)
369
370
    # Sort the spans by the index of the first span
    masked_spans = sorted(masked_spans, key=lambda x: x.index[0])
371
372
373
374

    for p in masked_lms:
        masked_lm_positions.append(p.index)
        masked_lm_labels.append(p.label)
375
    return (output_tokens, masked_lm_positions, masked_lm_labels, token_boundary, masked_spans)
376
377
378
379
380
381
382
383
384
385
386


def pad_and_convert_to_numpy(tokens, tokentypes, masked_positions,
                             masked_labels, pad_id, max_seq_length):
    """Pad sequences and convert them to numpy."""

    # Some checks.
    num_tokens = len(tokens)
    padding_length = max_seq_length - num_tokens
    assert padding_length >= 0
    assert len(tokentypes) == num_tokens
387
    assert len(masked_positions) == len(masked_labels)
388
389

    # Tokens and token types.
390
    filler = [pad_id] * padding_length
391
392
393
394
    tokens_np = np.array(tokens + filler, dtype=np.int64)
    tokentypes_np = np.array(tokentypes + filler, dtype=np.int64)

    # Padding mask.
395
    padding_mask_np = np.array([1] * num_tokens + [0] * padding_length,
Mohammad Shoeybi's avatar
Mohammad Shoeybi committed
396
                               dtype=np.int64)
397
398
399
400
401
402
403
404
405
406
407

    # Lables and loss mask.
    labels = [-1] * max_seq_length
    loss_mask = [0] * max_seq_length
    for i in range(len(masked_positions)):
        assert masked_positions[i] < num_tokens
        labels[masked_positions[i]] = masked_labels[i]
        loss_mask[masked_positions[i]] = 1
    labels_np = np.array(labels, dtype=np.int64)
    loss_mask_np = np.array(loss_mask, dtype=np.int64)

408
    return tokens_np, tokentypes_np, labels_np, padding_mask_np, loss_mask_np
409
410


xingjinliang's avatar
xingjinliang committed
411
def build_train_valid_test_datasets_with_prefixes(train_valid_test_num_samples,
liangjing's avatar
v1  
liangjing committed
412
413
414
415
416
417
418
419
420
421
422
423
424
                                                  max_seq_length,
                                                  seed,
                                                  train_data_prefix=None,
                                                  valid_data_prefix=None,
                                                  test_data_prefix=None,
                                                  binary_head=False,
                                                  max_seq_length_dec=None,
                                                  dataset_type='standard_bert'):
    print_rank_0("Separate data paths provided for train, valid & test.")

    train_dataset, valid_dataset, test_dataset = None, None, None
    # Single dataset.
    if train_data_prefix is not None:
xingjinliang's avatar
xingjinliang committed
425
        train_dataset = build_dataset("train", train_data_prefix,
liangjing's avatar
v1  
liangjing committed
426
                                      train_valid_test_num_samples[0],
xingjinliang's avatar
xingjinliang committed
427
                                      max_seq_length, seed,
liangjing's avatar
v1  
liangjing committed
428
429
430
431
                                      binary_head, max_seq_length_dec,
                                      dataset_type=dataset_type)

    if valid_data_prefix is not None:
xingjinliang's avatar
xingjinliang committed
432
        valid_dataset = build_dataset("valid", valid_data_prefix,
liangjing's avatar
v1  
liangjing committed
433
434
435
436
437
438
                                      train_valid_test_num_samples[1],
                                      max_seq_length, seed, False,
                                      binary_head, max_seq_length_dec,
                                      dataset_type=dataset_type)

    if test_data_prefix is not None:
xingjinliang's avatar
xingjinliang committed
439
        test_dataset = build_dataset("test", test_data_prefix,
liangjing's avatar
v1  
liangjing committed
440
441
442
443
444
445
446
447
                                     train_valid_test_num_samples[2],
                                     max_seq_length, seed, False,
                                     binary_head, max_seq_length_dec,
                                     dataset_type=dataset_type)

    return (train_dataset, valid_dataset, test_dataset)


xingjinliang's avatar
xingjinliang committed
448
def build_train_valid_test_datasets(data_prefix, splits_string,
449
                                    train_valid_test_num_samples,
liangjing's avatar
v1  
liangjing committed
450
                                    max_seq_length, seed,
xingjinliang's avatar
xingjinliang committed
451
                                    binary_head=False,
452
                                    max_seq_length_dec=None,
453
454
                                    dataset_type='standard_bert'):

mohammad's avatar
mohammad committed
455
456
    if len(data_prefix) == 1:
        return _build_train_valid_test_datasets(data_prefix[0],
xingjinliang's avatar
xingjinliang committed
457
                                                splits_string,
mohammad's avatar
mohammad committed
458
                                                train_valid_test_num_samples,
liangjing's avatar
v1  
liangjing committed
459
                                                max_seq_length, seed,
460
                                                binary_head,
461
                                                max_seq_length_dec,
mohammad's avatar
mohammad committed
462
463
                                                dataset_type=dataset_type)

xingjinliang's avatar
xingjinliang committed
464
465
466
467
    raise NotImplementedError("Blending currently unsupported for non-GPT dataset instances")


def _build_train_valid_test_datasets(data_prefix, splits_string,
mohammad's avatar
mohammad committed
468
                                     train_valid_test_num_samples,
liangjing's avatar
v1  
liangjing committed
469
                                     max_seq_length, seed,
xingjinliang's avatar
xingjinliang committed
470
                                     binary_head,
471
                                     max_seq_length_dec,
mohammad's avatar
mohammad committed
472
                                     dataset_type='standard_bert'):
473

474
475
    # Indexed dataset.
    indexed_dataset = get_indexed_dataset_(data_prefix,
xingjinliang's avatar
xingjinliang committed
476
                                           dataset_type)
477
478
479
480

    # Get start and end indices of train/valid/train into doc-idx
    # Note that doc-idx is desinged to be num-docs + 1 so we can
    # easily iterate over it.
xingjinliang's avatar
xingjinliang committed
481
    total_num_of_documents = indexed_dataset.document_indices.shape[0] - 1
482
483
484
485
486
487
488
489
490
491
    splits = get_train_valid_test_split_(splits_string, total_num_of_documents)

    # Print stats about the splits.
    print_rank_0(' > dataset split:')

    def print_split_stats(name, index):
        print_rank_0('    {}:'.format(name))
        print_rank_0('     document indices in [{}, {}) total of {} '
                     'documents'.format(splits[index], splits[index + 1],
                                        splits[index + 1] - splits[index]))
xingjinliang's avatar
xingjinliang committed
492
493
        start_index = indexed_dataset.document_indices[splits[index]]
        end_index = indexed_dataset.document_indices[splits[index + 1]]
494
495
496
497
498
499
500
        print_rank_0('     sentence indices in [{}, {}) total of {} '
                     'sentences'.format(start_index, end_index,
                                        end_index - start_index))
    print_split_stats('train', 0)
    print_split_stats('validation', 1)
    print_split_stats('test', 2)

liangjing's avatar
v1  
liangjing committed
501
    def build_split_dataset(index, name):
502
503
504
        dataset = None
        if splits[index + 1] > splits[index]:
            # Get the pointer to the original doc-idx so we can set it later.
xingjinliang's avatar
xingjinliang committed
505
            doc_idx_ptr = indexed_dataset.get_document_indices()
506
507
508
509
510
            # Slice the doc-idx
            start_index = splits[index]
            # Add +1 so we can index into the dataset to get the upper bound.
            end_index = splits[index + 1] + 1
            # New doc_idx view.
xingjinliang's avatar
xingjinliang committed
511
            indexed_dataset.set_document_indices(doc_idx_ptr[start_index:end_index])
liangjing's avatar
v1  
liangjing committed
512
513

            dataset = build_dataset(
xingjinliang's avatar
xingjinliang committed
514
                name, data_prefix,
liangjing's avatar
v1  
liangjing committed
515
                train_valid_test_num_samples[index], max_seq_length,
xingjinliang's avatar
xingjinliang committed
516
                seed, binary_head, max_seq_length_dec,
liangjing's avatar
v1  
liangjing committed
517
                dataset_type, indexed_dataset)
518
519

            # Set the original pointer so dataset remains the main dataset.
xingjinliang's avatar
xingjinliang committed
520
            indexed_dataset.set_document_indices(doc_idx_ptr)
521
            # Checks.
xingjinliang's avatar
xingjinliang committed
522
523
            assert indexed_dataset.document_indices[0] == 0
            assert indexed_dataset.document_indices.shape[0] == \
524
525
                (total_num_of_documents + 1)
        return dataset
liangjing's avatar
v1  
liangjing committed
526
527
528
529
    
    train_dataset = build_split_dataset(0, 'train')
    valid_dataset = build_split_dataset(1, 'valid')
    test_dataset = build_split_dataset(2, 'test')
530

531
    return (train_dataset, valid_dataset, test_dataset)
532
533


xingjinliang's avatar
xingjinliang committed
534
535
def build_dataset(name, data_prefix, max_num_samples,
                  max_seq_length, seed, binary_head,
liangjing's avatar
v1  
liangjing committed
536
537
538
                  max_seq_length_dec, dataset_type='standard_bert',
                  indexed_dataset=None):

xingjinliang's avatar
xingjinliang committed
539
540
541
542
543
    from megatron.legacy.data.ict_dataset import ICTDataset
    from megatron.legacy.data.multimodal_dataset import MultiModalDataset

    if dataset_type == DSET_TYPE_BERT or dataset_type == DSET_TYPE_T5:
        raise ValueError("The Megatron-LM BERT and T5 datasets are deprecated.")
liangjing's avatar
v1  
liangjing committed
544
545
546
547
548
549

    if dataset_type not in DSET_TYPES:
        raise ValueError("Invalid dataset_type: ", dataset_type)

    if indexed_dataset is None:
        indexed_dataset = get_indexed_dataset_(data_prefix,
xingjinliang's avatar
xingjinliang committed
550
                                               dataset_type)
liangjing's avatar
v1  
liangjing committed
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565

    kwargs = dict(
        name=name,
        data_prefix=data_prefix,
        num_epochs=None,
        max_num_samples=max_num_samples,
        max_seq_length=max_seq_length,
        seed=seed,
    )

    if dataset_type == DSET_TYPE_ICT:
        args = get_args()

        title_dataset = get_indexed_dataset_(
            args.titles_data_path,
xingjinliang's avatar
xingjinliang committed
566
            dataset_type)
liangjing's avatar
v1  
liangjing committed
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593

        dataset = ICTDataset(
            block_dataset=indexed_dataset,
            title_dataset=title_dataset,
            query_in_block_prob=args.query_in_block_prob,
            use_one_sent_docs=args.use_one_sent_docs,
            binary_head=binary_head,
            **kwargs
        )
    elif dataset_type == DSET_TYPE_MULTIMODAL:
        args = get_args()
        dataset = MultiModalDataset(
            name=name,
            data_prefix=data_prefix,
            indexed_dataset=indexed_dataset,
            num_samples=max_num_samples,
            seq_length=max_seq_length,
            seed=seed,
            img_h=args.img_h,
            img_w=args.img_w,
        )
    else:
        raise NotImplementedError("Dataset type not fully implemented.")

    return dataset


xingjinliang's avatar
xingjinliang committed
594
def get_indexed_dataset_(data_prefix, dataset_type):
595
596
597
598

    print_rank_0(' > building dataset index ...')

    start_time = time.time()
liangjing's avatar
v1  
liangjing committed
599
    multimodal = dataset_type == DSET_TYPE_MULTIMODAL
xingjinliang's avatar
xingjinliang committed
600
601
    indexed_dataset = IndexedDataset(data_prefix, multimodal)
    assert indexed_dataset.sequence_lengths.shape[0] == indexed_dataset.document_indices[-1]
602
603
604
605
606
    print_rank_0(' > finished creating indexed dataset in {:4f} '
                 'seconds'.format(time.time() - start_time))

    print_rank_0(' > indexed dataset stats:')
    print_rank_0('    number of documents: {}'.format(
xingjinliang's avatar
xingjinliang committed
607
        indexed_dataset.document_indices.shape[0] - 1))
608
    print_rank_0('    number of sentences: {}'.format(
xingjinliang's avatar
xingjinliang committed
609
        indexed_dataset.sequence_lengths.shape[0]))
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640

    return indexed_dataset


def get_train_valid_test_split_(splits_string, size):
    """ Get dataset splits from comma or '/' separated string list."""

    splits = []
    if splits_string.find(',') != -1:
        splits = [float(s) for s in splits_string.split(',')]
    elif splits_string.find('/') != -1:
        splits = [float(s) for s in splits_string.split('/')]
    else:
        splits = [float(splits_string)]
    while len(splits) < 3:
        splits.append(0.)
    splits = splits[:3]
    splits_sum = sum(splits)
    assert splits_sum > 0.0
    splits = [split / splits_sum for split in splits]
    splits_index = [0]
    for index, split in enumerate(splits):
        splits_index.append(splits_index[index] +
                            int(round(split * float(size))))
    diff = splits_index[-1] - size
    for index in range(1, len(splits_index)):
        splits_index[index] -= diff
    assert len(splits_index) == 4
    assert splits_index[-1] == size
    return splits_index

641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
def get_samples_mapping(indexed_dataset,
                        data_prefix,
                        num_epochs,
                        max_num_samples,
                        max_seq_length,
                        short_seq_prob,
                        seed,
                        name,
                        binary_head):
    """Get a list that maps a sample index to a starting sentence index, end sentence index, and length"""

    if not num_epochs:
        if not max_num_samples:
            raise ValueError("Need to specify either max_num_samples "
                             "or num_epochs")
        num_epochs = np.iinfo(np.int32).max - 1
    if not max_num_samples:
        max_num_samples = np.iinfo(np.int64).max - 1

    # Filename of the index mapping
    indexmap_filename = data_prefix
    indexmap_filename += '_{}_indexmap'.format(name)
    if num_epochs != (np.iinfo(np.int32).max - 1):
        indexmap_filename += '_{}ep'.format(num_epochs)
    if max_num_samples != (np.iinfo(np.int64).max - 1):
        indexmap_filename += '_{}mns'.format(max_num_samples)
    indexmap_filename += '_{}msl'.format(max_seq_length)
    indexmap_filename += '_{:0.2f}ssp'.format(short_seq_prob)
    indexmap_filename += '_{}s'.format(seed)
    indexmap_filename += '.npy'

    # Build the indexed mapping if not exist.
    if torch.distributed.get_rank() == 0 and \
       not os.path.isfile(indexmap_filename):
        print(' > WARNING: could not find index map file {}, building '
              'the indices on rank 0 ...'.format(indexmap_filename))

        # Make sure the types match the helpers input types.
xingjinliang's avatar
xingjinliang committed
679
680
        assert indexed_dataset.document_indices.dtype == np.int64
        assert indexed_dataset.sequence_lengths.dtype == np.int32
681
682
683
684

        # Build samples mapping
        verbose = torch.distributed.get_rank() == 0
        start_time = time.time()
mshoeybi's avatar
mshoeybi committed
685
        print_rank_0(' > building samples index mapping for {} ...'.format(
686
687
            name))
        # First compile and then import.
xingjinliang's avatar
xingjinliang committed
688
        from megatron.core.datasets import helpers
689
        samples_mapping = helpers.build_mapping(
xingjinliang's avatar
xingjinliang committed
690
691
            indexed_dataset.document_indices,
            indexed_dataset.sequence_lengths,
692
693
694
695
696
697
698
            num_epochs,
            max_num_samples,
            max_seq_length,
            short_seq_prob,
            seed,
            verbose,
            2 if binary_head else 1)
mshoeybi's avatar
mshoeybi committed
699
        print_rank_0(' > done building samples index maping')
700
701
702
703
704
705
706
707
708
709
        np.save(indexmap_filename, samples_mapping, allow_pickle=True)
        print_rank_0(' > saved the index mapping in {}'.format(
            indexmap_filename))
        # Make sure all the ranks have built the mapping
        print_rank_0(' > elasped time to build and save samples mapping '
                     '(seconds): {:4f}'.format(
                         time.time() - start_time))
    # This should be a barrier but nccl barrier assumes
    # device_index=rank which is not the case for model
    # parallel case
xingjinliang's avatar
xingjinliang committed
710
    counts = torch.tensor([1], dtype=torch.long, device='cuda')
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
    torch.distributed.all_reduce(counts, group=mpu.get_data_parallel_group())
    torch.distributed.all_reduce(counts, group=mpu.get_pipeline_model_parallel_group())
    assert counts[0].item() == (
        torch.distributed.get_world_size() //
        torch.distributed.get_world_size(group=mpu.get_tensor_model_parallel_group()))

    # Load indexed dataset.
    print_rank_0(' > loading indexed mapping from {}'.format(
        indexmap_filename))
    start_time = time.time()
    samples_mapping = np.load(indexmap_filename, allow_pickle=True, mmap_mode='r')
    print_rank_0('    loaded indexed file in {:3.3f} seconds'.format(
        time.time() - start_time))
    print_rank_0('    total number of samples: {}'.format(
        samples_mapping.shape[0]))
726

727
    return samples_mapping