You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
453 lines
14 KiB
453 lines
14 KiB
# 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.
|
|
|
|
from paddle.fluid.proto import data_feed_pb2
|
|
from google.protobuf import text_format
|
|
from . import core
|
|
__all__ = ['DatasetFactory', 'InMemoryDataset', 'QueueDataset']
|
|
|
|
|
|
class DatasetFactory(object):
|
|
"""
|
|
DatasetFactory is a factory which create dataset by its name,
|
|
you can create "QueueDataset" or "InMemoryDataset",
|
|
the default is "QueueDataset".
|
|
|
|
Example:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset("InMemoryDataset")
|
|
|
|
"""
|
|
|
|
def __init__(self):
|
|
""" Init. """
|
|
pass
|
|
|
|
def create_dataset(self, datafeed_class="QueueDataset"):
|
|
"""
|
|
Create "QueueDataset" or "InMemoryDataset",
|
|
the default is "QueueDataset".
|
|
|
|
Args:
|
|
datafeed_class(str): datafeed class name, QueueDataset or InMemoryDataset.
|
|
Default is QueueDataset.
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset()
|
|
|
|
"""
|
|
try:
|
|
dataset = globals()[datafeed_class]()
|
|
return dataset
|
|
except:
|
|
raise ValueError("datafeed class %s does not exist" %
|
|
datafeed_class)
|
|
|
|
|
|
class DatasetBase(object):
|
|
""" Base dataset class. """
|
|
|
|
def __init__(self):
|
|
""" Init. """
|
|
# 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"
|
|
self.dataset = core.Dataset("MultiSlotDataset")
|
|
self.thread_num = 0
|
|
|
|
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
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset()
|
|
dataset.set_pipe_command("python my_script.py")
|
|
|
|
Args:
|
|
pipe_command(str): pipe command
|
|
|
|
"""
|
|
self.proto_desc.pipe_command = pipe_command
|
|
|
|
def set_batch_size(self, batch_size):
|
|
"""
|
|
Set batch size. Will be effective during training
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset()
|
|
dataset.set_batch_size(128)
|
|
|
|
Args:
|
|
batch_size(int): batch size
|
|
|
|
"""
|
|
self.proto_desc.batch_size = batch_size
|
|
|
|
def set_thread(self, thread_num):
|
|
"""
|
|
Set thread num, it is the num of readers.
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset()
|
|
dataset.set_thread(12)
|
|
|
|
Args:
|
|
thread_num(int): thread num
|
|
"""
|
|
self.dataset.set_thread_num(thread_num)
|
|
self.thread_num = thread_num
|
|
|
|
def set_filelist(self, filelist):
|
|
"""
|
|
Set file list in current worker.
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset()
|
|
dataset.set_filelist(['a.txt', 'b.txt'])
|
|
|
|
Args:
|
|
filelist(list): file list
|
|
"""
|
|
self.dataset.set_filelist(filelist)
|
|
|
|
def set_use_var(self, var_list):
|
|
"""
|
|
Set Variables which you will use.
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset()
|
|
dataset.set_use_var([data, label])
|
|
|
|
Args:
|
|
var_list(list): variable list
|
|
"""
|
|
multi_slot = self.proto_desc.multi_slot_desc
|
|
for var in var_list:
|
|
slot_var = multi_slot.slots.add()
|
|
slot_var.is_used = True
|
|
slot_var.name = var.name
|
|
if var.lod_level == 0:
|
|
slot_var.is_dense = True
|
|
slot_var.shape.extend(var.shape)
|
|
if var.dtype == core.VarDesc.VarType.FP32:
|
|
slot_var.type = "float"
|
|
elif var.dtype == core.VarDesc.VarType.INT64:
|
|
slot_var.type = "uint64"
|
|
else:
|
|
raise ValueError(
|
|
"Currently, fluid.dataset only supports dtype=float32 and dtype=int64"
|
|
)
|
|
|
|
def set_hdfs_config(self, fs_name, fs_ugi):
|
|
"""
|
|
Set hdfs config: fs name ad ugi
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset()
|
|
dataset.set_hdfs_config("my_fs_name", "my_fs_ugi")
|
|
|
|
Args:
|
|
fs_name(str): fs name
|
|
fs_ugi(str): fs ugi
|
|
"""
|
|
self.dataset.set_hdfs_config(fs_name, fs_ugi)
|
|
|
|
def _prepare_to_run(self):
|
|
"""
|
|
Set data_feed_desc before load or shuffle,
|
|
user no need to call this function.
|
|
"""
|
|
self.dataset.set_data_feed_desc(self.desc())
|
|
|
|
def desc(self):
|
|
"""
|
|
Returns a protobuf message for this DataFeedDesc
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset()
|
|
print(dataset.desc())
|
|
|
|
Returns:
|
|
A string message
|
|
"""
|
|
return text_format.MessageToString(self.proto_desc)
|
|
|
|
|
|
class InMemoryDataset(DatasetBase):
|
|
"""
|
|
InMemoryDataset, it will load data into memory
|
|
and shuffle data before training.
|
|
This class should be created by DatasetFactory
|
|
|
|
Example:
|
|
dataset = paddle.fluid.DatasetFactory().create_dataset("InMemoryDataset")
|
|
"""
|
|
|
|
def __init__(self):
|
|
""" Init. """
|
|
super(InMemoryDataset, self).__init__()
|
|
self.proto_desc.name = "MultiSlotInMemoryDataFeed"
|
|
|
|
def load_into_memory(self):
|
|
"""
|
|
Load data into memory
|
|
|
|
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()
|
|
"""
|
|
self._prepare_to_run()
|
|
self.dataset.load_into_memory()
|
|
|
|
def local_shuffle(self):
|
|
"""
|
|
Local shuffle
|
|
|
|
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()
|
|
"""
|
|
self.dataset.local_shuffle()
|
|
|
|
def global_shuffle(self, fleet=None):
|
|
"""
|
|
Global shuffle.
|
|
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.
|
|
|
|
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)
|
|
|
|
Args:
|
|
fleet(Fleet): fleet singleton. Default None.
|
|
|
|
"""
|
|
trainer_num = 1
|
|
fleet_send_batch_size = 80000
|
|
if fleet is not None:
|
|
fleet._role_maker._barrier_worker()
|
|
trainer_num = fleet.worker_num()
|
|
self.dataset.register_client2client_msg_handler()
|
|
self.dataset.set_trainer_num(trainer_num)
|
|
self.dataset.set_fleet_send_batch_size(fleet_send_batch_size)
|
|
if fleet is not None:
|
|
fleet._role_maker._barrier_worker()
|
|
self.dataset.global_shuffle()
|
|
if fleet is not None:
|
|
fleet._role_maker._barrier_worker()
|
|
|
|
def release_memory(self):
|
|
"""
|
|
Release InMemoryDataset memory data, when data will not be used again.
|
|
|
|
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()
|
|
|
|
"""
|
|
self.dataset.release_memory()
|
|
|
|
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.
|
|
|
|
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)
|
|
|
|
"""
|
|
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
|
|
fleet._role_maker._node_type_comm.Allreduce(local_data_size,
|
|
global_data_size)
|
|
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.
|
|
|
|
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)
|
|
|
|
"""
|
|
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
|
|
fleet._role_maker._node_type_comm.Allreduce(local_data_size,
|
|
global_data_size)
|
|
return global_data_size[0]
|
|
return local_data_size[0]
|
|
|
|
|
|
class QueueDataset(DatasetBase):
|
|
"""
|
|
QueueDataset, it will process data streamly.
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset("QueueDataset")
|
|
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""
|
|
Initialize QueueDataset
|
|
This class should be created by DatasetFactory
|
|
"""
|
|
super(QueueDataset, self).__init__()
|
|
self.proto_desc.name = "MultiSlotDataFeed"
|
|
|
|
def local_shuffle(self):
|
|
"""
|
|
Local shuffle data.
|
|
|
|
Local shuffle is not supported in QueueDataset
|
|
NotImplementedError will be raised
|
|
|
|
Examples:
|
|
.. code-block:: python
|
|
|
|
import paddle.fluid as fluid
|
|
dataset = fluid.DatasetFactory().create_dataset("QueueDataset")
|
|
dataset.local_shuffle()
|
|
|
|
"""
|
|
raise NotImplementedError(
|
|
"QueueDataset does not support local shuffle, "
|
|
"please use InMemoryDataset for local_shuffle")
|
|
|
|
def global_shuffle(self, fleet=None):
|
|
"""
|
|
Global shuffle data.
|
|
|
|
Global shuffle is not supported in QueueDataset
|
|
NotImplementedError will be raised
|
|
|
|
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)
|
|
|
|
"""
|
|
raise NotImplementedError(
|
|
"QueueDataset does not support global shuffle, "
|
|
"please use InMemoryDataset for global_shuffle")
|