dataset.py 28.3 KB
Newer Older
D
dongdaxiang 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
#   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
#
# 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.
T
tianshuo78520a 已提交
14
"""This is definition of dataset class, which is high performance IO."""
D
dongdaxiang 已提交
15 16 17 18

from paddle.fluid.proto import data_feed_pb2
from google.protobuf import text_format
from . import core
D
dongdaxiang 已提交
19
__all__ = ['DatasetFactory', 'InMemoryDataset', 'QueueDataset']
D
dongdaxiang 已提交
20 21 22


class DatasetFactory(object):
23 24
    """
    DatasetFactory is a factory which create dataset by its name,
H
hutuxian 已提交
25
    you can create "QueueDataset" or "InMemoryDataset", or "FileInstantDataset",
26 27 28
    the default is "QueueDataset".

    Example:
29 30 31 32 33
        .. code-block:: python

          import paddle.fluid as fluid
          dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")

34
    """
D
dongdaxiang 已提交
35

D
dongdaxiang 已提交
36
    def __init__(self):
37
        """ Init. """
D
dongdaxiang 已提交
38 39
        pass

40
    def create_dataset(self, datafeed_class="QueueDataset"):
41
        """
H
hutuxian 已提交
42
        Create "QueueDataset" or "InMemoryDataset", or "FileInstantDataset",
43
        the default is "QueueDataset".
D
dongdaxiang 已提交
44

45 46 47 48
        Args:
            datafeed_class(str): datafeed class name, QueueDataset or InMemoryDataset.
                                 Default is QueueDataset.

D
dongdaxiang 已提交
49
        Examples:
50 51 52 53 54
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()

55
        """
D
dongdaxiang 已提交
56 57
        try:
            dataset = globals()[datafeed_class]()
58
            return dataset
D
dongdaxiang 已提交
59 60 61 62 63 64
        except:
            raise ValueError("datafeed class %s does not exist" %
                             datafeed_class)


class DatasetBase(object):
65
    """ Base dataset class. """
D
dongdaxiang 已提交
66

D
dongdaxiang 已提交
67
    def __init__(self):
68
        """ Init. """
D
dongdaxiang 已提交
69 70 71 72
        # define class name here
        # to decide whether we need create in memory instance
        self.proto_desc = data_feed_pb2.DataFeedDesc()
        self.proto_desc.pipe_command = "cat"
X
xujiaqi01 已提交
73
        self.dataset = core.Dataset("MultiSlotDataset")
74
        self.thread_num = 1
J
jiaqi 已提交
75
        self.filelist = []
D
dongdaxiang 已提交
76 77 78 79 80 81

    def set_pipe_command(self, pipe_command):
        """
        Set pipe command of current dataset
        A pipe command is a UNIX pipeline command that can be used only

82 83 84 85 86 87
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_pipe_command("python my_script.py")
88 89

        Args:
90
            pipe_command(str): pipe command
91

D
dongdaxiang 已提交
92 93 94
        """
        self.proto_desc.pipe_command = pipe_command

95 96 97 98 99 100 101 102
    def set_fea_eval(self, record_candidate_size, fea_eval=True):
        """
        set fea eval mode for slots shuffle to debug the importance level of
        slots(features), fea_eval need to be set True for slots shuffle.
        
        Args:
            record_candidate_size(int): size of instances candidate to shuffle 
                                        one slot
T
tianshuo78520a 已提交
103
            fea_eval(bool): whether enable fea eval mode to enable slots shuffle.
104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139
                            default is True.
            
        Examples:
            .. code-block:: python

            import paddle.fluid as fluid
            dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
            dataset.set_fea_eval(1000000, True)

        """
        if fea_eval:
            self.dataset.set_fea_eval(fea_eval, record_candidate_size)
        self.fea_eval = fea_eval

    def slots_shuffle(self, slots):
        """
        Slots Shuffle 
        Slots Shuffle is a shuffle method in slots level, which is usually used 
        in sparse feature with large scale of instances. To compare the metric, i.e.
        auc while doing slots shuffle on one or several slots with baseline to 
        evaluate the importance level of slots(features).
        
        Args:
            slots(list[string]): the set of slots(string) to do slots shuffle.

        Examples:
            import paddle.fluid as fluid
            dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
            dataset.set_merge_by_lineid()
            #suppose there is a slot 0
            dataset.slots_shuffle(['0'])
        """
        if self.fea_eval:
            slots_set = set(slots)
            self.dataset.slots_shuffle(slots_set)

D
dongdaxiang 已提交
140 141 142 143
    def set_batch_size(self, batch_size):
        """
        Set batch size. Will be effective during training

144 145 146 147 148 149
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_batch_size(128)
D
dongdaxiang 已提交
150 151

        Args:
152
            batch_size(int): batch size
D
dongdaxiang 已提交
153 154 155 156

        """
        self.proto_desc.batch_size = batch_size

157
    def set_thread(self, thread_num):
158 159 160
        """
        Set thread num, it is the num of readers.

161 162 163 164 165 166
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
               dataset.set_thread(12)
167 168

        Args:
169
            thread_num(int): thread num
170
        """
171
        self.dataset.set_thread_num(thread_num)
172
        self.thread_num = thread_num
173 174

    def set_filelist(self, filelist):
175 176 177
        """
        Set file list in current worker.

178 179 180 181 182 183
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_filelist(['a.txt', 'b.txt'])
184 185

        Args:
186
            filelist(list): file list
187
        """
188
        self.dataset.set_filelist(filelist)
J
jiaqi 已提交
189
        self.filelist = filelist
190

D
dongdaxiang 已提交
191
    def set_use_var(self, var_list):
192 193 194
        """
        Set Variables which you will use.

195 196 197 198 199 200
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_use_var([data, label])
201 202

        Args:
203
            var_list(list): variable list
204
        """
205
        multi_slot = self.proto_desc.multi_slot_desc
D
dongdaxiang 已提交
206
        for var in var_list:
207
            slot_var = multi_slot.slots.add()
D
dongdaxiang 已提交
208 209 210 211
            slot_var.is_used = True
            slot_var.name = var.name
            if var.lod_level == 0:
                slot_var.is_dense = True
212
                slot_var.shape.extend(var.shape)
213
            if var.dtype == core.VarDesc.VarType.FP32:
D
dongdaxiang 已提交
214
                slot_var.type = "float"
215
            elif var.dtype == core.VarDesc.VarType.INT64:
D
dongdaxiang 已提交
216 217 218 219 220 221
                slot_var.type = "uint64"
            else:
                raise ValueError(
                    "Currently, fluid.dataset only supports dtype=float32 and dtype=int64"
                )

222
    def set_hdfs_config(self, fs_name, fs_ugi):
223 224 225
        """
        Set hdfs config: fs name ad ugi

226 227 228 229 230 231
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_hdfs_config("my_fs_name", "my_fs_ugi")
232 233

        Args:
234 235
            fs_name(str): fs name
            fs_ugi(str): fs ugi
236
        """
237 238
        self.dataset.set_hdfs_config(fs_name, fs_ugi)

239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254
    def set_download_cmd(self, download_cmd):
        """
        Set customized download cmd: download_cmd

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              dataset.set_download_cmd("./read_from_afs")

        Args:
            download_cmd(str): customized download command
        """
        self.dataset.set_download_cmd(download_cmd)

255
    def _prepare_to_run(self):
256 257 258 259
        """
        Set data_feed_desc before load or shuffle,
        user no need to call this function.
        """
J
jiaqi 已提交
260 261 262
        if self.thread_num > len(self.filelist):
            self.thread_num = len(self.filelist)
        self.dataset.set_thread_num(self.thread_num)
263
        self.dataset.set_data_feed_desc(self.desc())
J
jiaqi 已提交
264 265 266 267
        self.dataset.create_readers()

    def _finish_to_run(self):
        self.dataset.destroy_readers()
268

D
dongdaxiang 已提交
269 270 271 272
    def desc(self):
        """
        Returns a protobuf message for this DataFeedDesc

273 274 275 276 277 278
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset()
              print(dataset.desc())
D
dongdaxiang 已提交
279 280 281 282 283 284

        Returns:
            A string message
        """
        return text_format.MessageToString(self.proto_desc)

285 286 287 288 289 290
    def _dynamic_adjust_before_train(self, thread_num):
        pass

    def _dynamic_adjust_after_train(self):
        pass

D
dongdaxiang 已提交
291 292

class InMemoryDataset(DatasetBase):
293 294
    """
    InMemoryDataset, it will load data into memory
D
dongdaxiang 已提交
295 296
    and shuffle data before training.
    This class should be created by DatasetFactory
297 298

    Example:
299
        dataset = paddle.fluid.DatasetFactory().create_dataset("InMemoryDataset")
300
    """
D
dongdaxiang 已提交
301

D
dongdaxiang 已提交
302
    def __init__(self):
303
        """ Init. """
304 305
        super(InMemoryDataset, self).__init__()
        self.proto_desc.name = "MultiSlotInMemoryDataFeed"
306
        self.fleet_send_batch_size = None
307
        self.is_user_set_queue_num = False
J
jiaqi 已提交
308
        self.queue_num = None
309 310
        self.parse_ins_id = False
        self.parse_content = False
311
        self.merge_by_lineid = False
312
        self.fleet_send_sleep_seconds = None
J
jiaqi 已提交
313 314 315 316 317 318

    def _prepare_to_run(self):
        """
        Set data_feed_desc before load or shuffle,
        user no need to call this function.
        """
319
        if self.thread_num <= 0:
320
            self.thread_num = 1
J
jiaqi 已提交
321 322 323 324
        self.dataset.set_thread_num(self.thread_num)
        if self.queue_num is None:
            self.queue_num = self.thread_num
        self.dataset.set_queue_num(self.queue_num)
325 326
        self.dataset.set_parse_ins_id(self.parse_ins_id)
        self.dataset.set_parse_content(self.parse_content)
J
jiaqi 已提交
327 328 329 330
        self.dataset.set_data_feed_desc(self.desc())
        self.dataset.create_channel()
        self.dataset.create_readers()

331 332
    def _dynamic_adjust_before_train(self, thread_num):
        if not self.is_user_set_queue_num:
H
hutuxian 已提交
333
            self.dataset.dynamic_adjust_channel_num(thread_num, False)
334 335 336 337
        self.dataset.dynamic_adjust_readers_num(thread_num)

    def _dynamic_adjust_after_train(self):
        if not self.is_user_set_queue_num:
H
hutuxian 已提交
338
            self.dataset.dynamic_adjust_channel_num(self.thread_num, False)
339 340
        self.dataset.dynamic_adjust_readers_num(self.thread_num)

J
jiaqi 已提交
341 342 343 344 345
    def set_queue_num(self, queue_num):
        """
        Set Dataset output queue num, training threads get data from queues

        Args:
346
            queue_num(int): dataset output queue num
J
jiaqi 已提交
347 348 349 350 351 352 353 354 355

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_queue_num(12)

        """
356
        self.is_user_set_queue_num = True
J
jiaqi 已提交
357 358
        self.queue_num = queue_num

359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392
    def set_parse_ins_id(self, parse_ins_id):
        """
        Set id Dataset need to parse insid

        Args:
            parse_ins_id(bool): if parse ins_id or not

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_parse_ins_id(True)

        """
        self.parse_ins_id = parse_ins_id

    def set_parse_content(self, parse_content):
        """
        Set if Dataset need to parse content

        Args:
            parse_content(bool): if parse content or not

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_parse_content(True)

        """
        self.parse_content = parse_content

393
    def set_fleet_send_batch_size(self, fleet_send_batch_size=1024):
J
jiaqi 已提交
394
        """
395
        Set fleet send batch size, default is 1024
J
jiaqi 已提交
396 397 398 399 400 401 402 403 404 405 406 407 408

        Args:
            fleet_send_batch_size(int): fleet send batch size

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_fleet_send_batch_size(800)

        """
        self.fleet_send_batch_size = fleet_send_batch_size
409

410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426
    def set_fleet_send_sleep_seconds(self, fleet_send_sleep_seconds=0):
        """
        Set fleet send sleep time, default is 0

        Args:
            fleet_send_sleep_seconds(int): fleet send sleep time

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_fleet_send_sleep_seconds(2)

        """
        self.fleet_send_sleep_seconds = fleet_send_sleep_seconds

427
    def set_merge_by_lineid(self, merge_size=2):
428 429 430 431 432
        """
        Set merge by line id, instances of same line id will be merged after
        shuffle, you should parse line id in data generator.

        Args:
433
            merge_size(int): ins size to merge. default is 2.
434 435 436 437 438 439 440 441 442

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              dataset.set_merge_by_lineid()

        """
443
        self.dataset.set_merge_by_lineid(merge_size)
444
        self.merge_by_lineid = True
445
        self.parse_ins_id = True
446

447 448 449 450 451 452 453 454 455 456
    def set_generate_unique_feasigns(self, generate_uni_feasigns, shard_num):
        self.dataset.set_generate_unique_feasigns(generate_uni_feasigns)
        self.gen_uni_feasigns = generate_uni_feasigns
        self.local_shard_num = shard_num

    def generate_local_tables_unlock(self, table_id, fea_dim, read_thread_num,
                                     consume_thread_num, shard_num):
        self.dataset.generate_local_tables_unlock(
            table_id, fea_dim, read_thread_num, consume_thread_num, shard_num)

457
    def load_into_memory(self):
458 459 460
        """
        Load data into memory

461 462 463 464 465 466 467 468
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
469
        """
470
        self._prepare_to_run()
471
        self.dataset.load_into_memory()
D
dongdaxiang 已提交
472

473
    def preload_into_memory(self, thread_num=None):
J
jiaqi 已提交
474 475 476
        """
        Load data into memory in async mode

477 478 479
        Args:
            thread_num(int): preload thread num

J
jiaqi 已提交
480 481 482 483 484 485 486 487 488 489 490
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.preload_into_memory()
              dataset.wait_preload_done()
        """
        self._prepare_to_run()
491 492 493 494
        if thread_num is None:
            thread_num = self.thread_num
        self.dataset.set_preload_thread_num(thread_num)
        self.dataset.create_preload_readers()
J
jiaqi 已提交
495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511
        self.dataset.preload_into_memory()

    def wait_preload_done(self):
        """
        Wait preload_into_memory done

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.preload_into_memory()
              dataset.wait_preload_done()
        """
        self.dataset.wait_preload_done()
512
        self.dataset.destroy_preload_readers()
J
jiaqi 已提交
513

D
dongdaxiang 已提交
514
    def local_shuffle(self):
515 516 517
        """
        Local shuffle

518 519 520 521 522 523 524 525 526
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              dataset.local_shuffle()
527
        """
528
        self.dataset.local_shuffle()
D
dongdaxiang 已提交
529

530
    def global_shuffle(self, fleet=None, thread_num=12):
531 532
        """
        Global shuffle.
533 534 535
        Global shuffle can be used only in distributed mode. i.e. multiple
        processes on single machine or multiple machines training together.
        If you run in distributed mode, you should pass fleet instead of None.
536

537
        Examples:
538 539 540 541 542 543 544 545 546
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              dataset.global_shuffle(fleet)
547 548

        Args:
549
            fleet(Fleet): fleet singleton. Default None.
550
            thread_num(int): shuffle thread num. Default is 12.
551

552
        """
553 554
        trainer_num = 1
        if fleet is not None:
X
xujiaqi01 已提交
555
            fleet._role_maker.barrier_worker()
556
            trainer_num = fleet.worker_num()
557
        if self.fleet_send_batch_size is None:
558 559 560
            self.fleet_send_batch_size = 1024
        if self.fleet_send_sleep_seconds is None:
            self.fleet_send_sleep_seconds = 0
561
        self.dataset.register_client2client_msg_handler()
562
        self.dataset.set_trainer_num(trainer_num)
J
jiaqi 已提交
563
        self.dataset.set_fleet_send_batch_size(self.fleet_send_batch_size)
564
        self.dataset.set_fleet_send_sleep_seconds(self.fleet_send_sleep_seconds)
565
        if fleet is not None:
X
xujiaqi01 已提交
566
            fleet._role_maker.barrier_worker()
567
        self.dataset.global_shuffle(thread_num)
568
        if fleet is not None:
X
xujiaqi01 已提交
569
            fleet._role_maker.barrier_worker()
570 571 572
        if self.merge_by_lineid:
            self.dataset.merge_by_lineid()
        if fleet is not None:
X
xujiaqi01 已提交
573
            fleet._role_maker.barrier_worker()
D
dongdaxiang 已提交
574

575 576 577 578
    def release_memory(self):
        """
        Release InMemoryDataset memory data, when data will not be used again.

579 580 581 582 583 584 585 586 587 588 589 590 591 592 593
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              dataset.global_shuffle(fleet)
              exe = fluid.Executor(fluid.CPUPlace())
              exe.run(fluid.default_startup_program())
              exe.train_from_dataset(fluid.default_main_program(), dataset)
              dataset.release_memory()

594 595
        """
        self.dataset.release_memory()
D
dongdaxiang 已提交
596

597 598 599 600 601 602 603 604 605 606 607 608 609 610
    def get_memory_data_size(self, fleet=None):
        """
        Get memory data size, user can call this function to know the num
        of ins in all workers after load into memory.

        Note:
            This function may cause bad performance, because it has barrier

        Args:
            fleet(Fleet): Fleet Object.

        Returns:
            The size of memory data.

611 612 613 614 615 616 617 618 619 620
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              print dataset.get_memory_data_size(fleet)
621 622 623 624 625 626 627

        """
        import numpy as np
        local_data_size = self.dataset.get_memory_data_size()
        local_data_size = np.array([local_data_size])
        if fleet is not None:
            global_data_size = local_data_size * 0
X
xujiaqi01 已提交
628 629
            fleet._role_maker.all_reduce_worker(local_data_size,
                                                global_data_size)
630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647
            return global_data_size[0]
        return local_data_size[0]

    def get_shuffle_data_size(self, fleet=None):
        """
        Get shuffle data size, user can call this function to know the num
        of ins in all workers after local/global shuffle.

        Note:
            This function may cause bad performance to local shuffle,
            because it has barrier. It does not affect global shuffle.

        Args:
            fleet(Fleet): Fleet Object.

        Returns:
            The size of shuffle data.

648 649 650 651 652 653 654 655 656 657 658
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
              dataset.global_shuffle(fleet)
              print dataset.get_shuffle_data_size(fleet)
659 660 661 662 663 664 665

        """
        import numpy as np
        local_data_size = self.dataset.get_shuffle_data_size()
        local_data_size = np.array([local_data_size])
        if fleet is not None:
            global_data_size = local_data_size * 0
X
xujiaqi01 已提交
666 667
            fleet._role_maker.all_reduce_worker(local_data_size,
                                                global_data_size)
668 669 670
            return global_data_size[0]
        return local_data_size[0]

X
xjqbest 已提交
671

D
dongdaxiang 已提交
672
class QueueDataset(DatasetBase):
673 674 675
    """
    QueueDataset, it will process data streamly.

676 677 678 679 680 681
    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
          dataset = fluid.DatasetFactory().create_dataset("QueueDataset")

682
    """
D
dongdaxiang 已提交
683

D
dongdaxiang 已提交
684
    def __init__(self):
685
        """
D
dongdaxiang 已提交
686 687
        Initialize QueueDataset
        This class should be created by DatasetFactory
688
        """
689
        super(QueueDataset, self).__init__()
D
dongdaxiang 已提交
690
        self.proto_desc.name = "MultiSlotDataFeed"
X
xujiaqi01 已提交
691

692 693 694 695 696 697 698 699 700 701 702 703 704 705
    def _prepare_to_run(self):
        """
        Set data_feed_desc/thread num/filelist before run,
        user no need to call this function.
        """
        if self.thread_num > len(self.filelist):
            self.thread_num = len(self.filelist)
        if self.thread_num == 0:
            self.thread_num = 1
        self.dataset.set_thread_num(self.thread_num)
        self.dataset.set_filelist(self.filelist)
        self.dataset.set_data_feed_desc(self.desc())
        self.dataset.create_readers()

X
xujiaqi01 已提交
706
    def local_shuffle(self):
707
        """
708
        Local shuffle data.
D
dongdaxiang 已提交
709

D
dongdaxiang 已提交
710 711
        Local shuffle is not supported in QueueDataset
        NotImplementedError will be raised
712 713 714 715 716 717 718 719

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("QueueDataset")
              dataset.local_shuffle()

720 721 722
        Raises:
            NotImplementedError: QueueDataset does not support local shuffle

723
        """
D
dongdaxiang 已提交
724 725 726
        raise NotImplementedError(
            "QueueDataset does not support local shuffle, "
            "please use InMemoryDataset for local_shuffle")
X
xujiaqi01 已提交
727

728
    def global_shuffle(self, fleet=None):
729
        """
730 731
        Global shuffle data.

D
dongdaxiang 已提交
732 733
        Global shuffle is not supported in QueueDataset
        NotImplementedError will be raised
734

735 736 737
        Args:
            fleet(Fleet): fleet singleton. Default None.

738 739 740 741 742 743 744 745
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              from paddle.fluid.incubate.fleet.parameter_server.pslib import fleet
              dataset = fluid.DatasetFactory().create_dataset("QueueDataset")
              dataset.global_shuffle(fleet)

746 747 748
        Raises:
            NotImplementedError: QueueDataset does not support global shuffle

749
        """
D
dongdaxiang 已提交
750 751 752
        raise NotImplementedError(
            "QueueDataset does not support global shuffle, "
            "please use InMemoryDataset for global_shuffle")
H
hutuxian 已提交
753 754 755 756 757


class FileInstantDataset(DatasetBase):
    """
    FileInstantDataset, it will process data streamly.
758 759 760 761 762 763

    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
          dataset = fluid.DatasetFactory.create_dataset("FileInstantDataset")
H
hutuxian 已提交
764 765 766 767
    """

    def __init__(self):
        """
768 769
        Initialize FileInstantDataset
        This class should be created by DatasetFactory
H
hutuxian 已提交
770 771 772 773 774 775
        """
        super(FileInstantDataset, self).__init__()
        self.proto_desc.name = "MultiSlotFileInstantDataFeed"

    def local_shuffle(self):
        """
776 777
        Local shuffle
        FileInstantDataset does not support local shuffle
H
hutuxian 已提交
778 779 780 781 782 783 784 785
        """
        raise NotImplementedError(
            "FileInstantDataset does not support local shuffle, "
            "please use InMemoryDataset for local_shuffle")

    def global_shuffle(self, fleet=None):
        """
        Global shuffle
786
        FileInstantDataset does not support global shuffle
H
hutuxian 已提交
787 788 789 790
        """
        raise NotImplementedError(
            "FileInstantDataset does not support global shuffle, "
            "please use InMemoryDataset for global_shuffle")
H
hutuxian 已提交
791 792 793 794 795 796 797 798 799 800


class BoxPSDataset(InMemoryDataset):
    """
    BoxPSDataset: derived from InMemoryDataset.

    Examples:
        .. code-block:: python

          import paddle.fluid as fluid
H
hutuxian 已提交
801
          dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
H
hutuxian 已提交
802 803 804 805
    """

    def __init__(self):
        """
806 807
        Initialize BoxPSDataset
        This class should be created by DatasetFactory
H
hutuxian 已提交
808 809 810 811
        """
        super(BoxPSDataset, self).__init__()
        self.boxps = core.BoxPS(self.dataset)

H
hutuxian 已提交
812 813 814 815 816 817 818 819 820
    def set_date(self, date):
        """
        Workaround for date
        """
        year = int(date[:4])
        month = int(date[4:6])
        day = int(date[6:])
        self.boxps.set_date(year, month, day)

H
hutuxian 已提交
821 822
    def begin_pass(self):
        """
823
        Begin Pass
H
hutuxian 已提交
824 825 826 827 828 829 830 831 832
        Notify BoxPS to load sparse parameters of next pass to GPU Memory 

        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              dataset.begin_pass()
        """
H
hutuxian 已提交
833 834
        self.boxps.begin_pass()

835
    def end_pass(self, need_save_delta):
H
hutuxian 已提交
836
        """
837
        End Pass
H
hutuxian 已提交
838 839 840 841 842 843
        Notify BoxPS that current pass ended 
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
844
              dataset.end_pass(True)
H
hutuxian 已提交
845
        """
846
        self.boxps.end_pass(need_save_delta)
H
hutuxian 已提交
847 848 849

    def wait_preload_done(self):
        """
T
tianshuo78520a 已提交
850
        Wait async preload done
851
        Wait Until Feed Pass Done
H
hutuxian 已提交
852 853 854 855 856 857 858 859 860 861
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.preload_into_memory()
              dataset.wait_preload_done()
        """
H
hutuxian 已提交
862 863 864 865
        self.boxps.wait_feed_pass_done()

    def load_into_memory(self):
        """
H
hutuxian 已提交
866 867 868 869 870 871 872 873 874 875
        Load next pass into memory and notify boxps to fetch its emb from SSD
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.load_into_memory()
	    """
H
hutuxian 已提交
876 877 878 879 880
        self._prepare_to_run()
        self.boxps.load_into_memory()

    def preload_into_memory(self):
        """
H
hutuxian 已提交
881 882 883 884 885 886 887 888 889 890
        Begin async preload next pass while current pass may be training
        Examples:
            .. code-block:: python

              import paddle.fluid as fluid
              dataset = fluid.DatasetFactory().create_dataset("BoxPSDataset")
              filelist = ["a.txt", "b.txt"]
              dataset.set_filelist(filelist)
              dataset.preload_into_memory()
        """
H
hutuxian 已提交
891 892
        self._prepare_to_run()
        self.boxps.preload_into_memory()
H
hutuxian 已提交
893 894 895 896 897

    def _dynamic_adjust_before_train(self, thread_num):
        if not self.is_user_set_queue_num:
            self.dataset.dynamic_adjust_channel_num(thread_num, True)
        self.dataset.dynamic_adjust_readers_num(thread_num)