wehub-resource-sync
1513 lines
53 KiB
Python
Executable File
1513 lines
53 KiB
Python
Executable File
# 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.
|
|
"""This is definition of dataset class, which is high performance IO."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Literal, TypedDict
|
|
|
|
from google.protobuf import text_format
|
|
|
|
import paddle
|
|
from paddle.base import core
|
|
from paddle.base.proto import data_feed_pb2
|
|
|
|
if TYPE_CHECKING:
|
|
from typing import TypeAlias
|
|
|
|
from typing_extensions import NotRequired, Unpack
|
|
|
|
from paddle import Tensor
|
|
from paddle.distributed.fleet import Fleet
|
|
|
|
_InputType: TypeAlias = Literal[0, 1]
|
|
|
|
class _DatasetBaseSettings(TypedDict):
|
|
batch_size: NotRequired[int]
|
|
thread_num: NotRequired[int]
|
|
use_var: NotRequired[list[Tensor]]
|
|
pipe_command: NotRequired[str]
|
|
input_type: NotRequired[_InputType]
|
|
fs_name: NotRequired[str]
|
|
fs_ugi: NotRequired[str]
|
|
download_cmd: NotRequired[str]
|
|
|
|
class _InMemoryDatasetDistributedSettings(TypedDict):
|
|
merge_size: NotRequired[int]
|
|
parse_ins_id: NotRequired[bool]
|
|
parse_content: NotRequired[bool]
|
|
fleet_send_batch_size: NotRequired[int]
|
|
fleet_send_sleep_seconds: NotRequired[int]
|
|
fea_eval: NotRequired[bool]
|
|
candidate_size: NotRequired[int]
|
|
|
|
class _InMemoryDatasetSettings(_DatasetBaseSettings):
|
|
data_feed_type: NotRequired[str]
|
|
queue_num: NotRequired[int]
|
|
|
|
class _InMemoryDatasetFullSettings(
|
|
_InMemoryDatasetDistributedSettings, _InMemoryDatasetSettings
|
|
):
|
|
pass
|
|
|
|
|
|
__all__ = []
|
|
|
|
|
|
class DatasetBase:
|
|
"""Base dataset class."""
|
|
|
|
proto_desc: data_feed_pb2.DataFeedDesc
|
|
dataset: core.Dataset
|
|
thread_num: int
|
|
filelist: list[str]
|
|
use_ps_gpu: bool
|
|
psgpu: core.PSGPU | None
|
|
|
|
def __init__(self) -> None:
|
|
"""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 = 1
|
|
self.filelist = []
|
|
self.use_ps_gpu = False
|
|
self.psgpu = None
|
|
|
|
def init(
|
|
self,
|
|
batch_size: int = 1,
|
|
thread_num: int = 1,
|
|
use_var: list[Tensor] = [],
|
|
pipe_command: str = "cat",
|
|
input_type: _InputType = 0,
|
|
fs_name: str = "",
|
|
fs_ugi: str = "",
|
|
download_cmd: str = "cat",
|
|
) -> None:
|
|
"""
|
|
should be called only once in user's python scripts to initialize settings of dataset instance.
|
|
Normally, it is called by InMemoryDataset or QueueDataset.
|
|
|
|
Args:
|
|
batch_size(int): batch size. It will be effective during training. default is 1.
|
|
thread_num(int): thread num, it is the num of readers. default is 1.
|
|
use_var(list): list of variables. Variables which you will use. default is [].
|
|
pipe_command(str): pipe command of current dataset. A pipe command is a UNIX pipeline command that can be used only. default is "cat"
|
|
input_type(int): the input type of generated input. 0 is for one sample, 1 is for one batch. default is 0.
|
|
fs_name(str): fs name. default is "".
|
|
fs_ugi(str): fs ugi. default is "".
|
|
download_cmd(str): customized download command. default is "cat"
|
|
|
|
"""
|
|
self._set_batch_size(batch_size)
|
|
self._set_thread(thread_num)
|
|
self._set_use_var(use_var)
|
|
self._set_pipe_command(pipe_command)
|
|
self._set_input_type(input_type)
|
|
self._set_hdfs_config(fs_name, fs_ugi)
|
|
self._set_download_cmd(download_cmd)
|
|
|
|
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:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.dataset.DatasetBase()
|
|
>>> 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:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> 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:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> 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: list[str]) -> None:
|
|
"""
|
|
Set file list in current worker. The filelist is indicated by a list of file names (string).
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> dataset.set_filelist(['a.txt', 'b.txt'])
|
|
|
|
Args:
|
|
filelist(list[str]): list of file names of inputs.
|
|
"""
|
|
self.dataset.set_filelist(filelist)
|
|
self.filelist = filelist
|
|
|
|
def _set_input_type(self, input_type):
|
|
self.proto_desc.input_type = input_type
|
|
|
|
def _set_uid_slot(self, uid_slot):
|
|
"""
|
|
Set user slot name.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> dataset._set_uid_slot('6048')
|
|
|
|
Args:
|
|
set_uid_slot(string): user slot name
|
|
"""
|
|
multi_slot = self.proto_desc.multi_slot_desc
|
|
multi_slot.uid_slot = uid_slot
|
|
|
|
def _set_use_var(self, var_list):
|
|
"""
|
|
Set Variables which you will use.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> 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 paddle.framework.use_pir_api():
|
|
slot_var.is_dense = True
|
|
slot_var.shape.extend(var.shape)
|
|
else:
|
|
if var.lod_level == 0:
|
|
slot_var.is_dense = True
|
|
slot_var.shape.extend(var.shape)
|
|
if var.dtype == paddle.float32:
|
|
slot_var.type = "float"
|
|
elif var.dtype == paddle.int64:
|
|
slot_var.type = "uint64"
|
|
else:
|
|
raise ValueError(
|
|
"Currently, paddle.distributed.fleet.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:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> 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 _set_download_cmd(self, download_cmd):
|
|
"""
|
|
Set customized download cmd: download_cmd
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> dataset._set_download_cmd("./read_from_afs")
|
|
|
|
Args:
|
|
download_cmd(str): customized download command
|
|
"""
|
|
self.dataset.set_download_cmd(download_cmd)
|
|
|
|
def _prepare_to_run(self):
|
|
"""
|
|
Set data_feed_desc before load or shuffle,
|
|
user no need to call this function.
|
|
"""
|
|
if self.thread_num > len(self.filelist):
|
|
self.thread_num = len(self.filelist)
|
|
self.dataset.set_thread_num(self.thread_num)
|
|
self.dataset.set_data_feed_desc(self._desc())
|
|
self.dataset.create_readers()
|
|
|
|
def _set_use_ps_gpu(self, use_ps_gpu):
|
|
"""
|
|
set use_ps_gpu flag
|
|
|
|
Args:
|
|
use_ps_gpu: bool
|
|
"""
|
|
self.use_ps_gpu = use_ps_gpu
|
|
# if not defined heterps with paddle, users will not use psgpu
|
|
if not core._is_compiled_with_heterps():
|
|
self.use_ps_gpu = 0
|
|
elif self.use_ps_gpu:
|
|
self.psgpu = core.PSGPU()
|
|
|
|
def _finish_to_run(self):
|
|
self.dataset.destroy_readers()
|
|
|
|
def _desc(self):
|
|
"""
|
|
Returns a protobuf message for this DataFeedDesc
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> print(dataset._desc())
|
|
pipe_command: "cat"
|
|
|
|
Returns:
|
|
A string message
|
|
"""
|
|
return text_format.MessageToString(self.proto_desc)
|
|
|
|
def _dynamic_adjust_before_train(self, thread_num):
|
|
pass
|
|
|
|
def _dynamic_adjust_after_train(self):
|
|
pass
|
|
|
|
def _check_use_var_with_data_generator(
|
|
self, var_list, data_generator_class, test_file
|
|
):
|
|
"""
|
|
Var consistency inspection of use_var_list and data_generator data.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('need to work with real dataset')
|
|
>>> import paddle
|
|
>>> from dataset_generator import CTRDataset
|
|
>>> dataset = paddle.distributed.fleet.DatasetBase()
|
|
>>> generator_class = CTRDataset()
|
|
>>> dataset._check_use_var_with_data_generator([data, label], generator_class, "data/part-00000")
|
|
|
|
Args:
|
|
var_list(list): variable list
|
|
data_generator_class(class): data_generator class
|
|
test_file(str): local test file path
|
|
"""
|
|
|
|
f = open(test_file, "r")
|
|
var_len = len(var_list)
|
|
|
|
while True:
|
|
line = f.readline()
|
|
if line:
|
|
line_iter = data_generator_class.generate_sample(line)
|
|
for user_parsed_line in line_iter():
|
|
data_gen_len = len(user_parsed_line)
|
|
if var_len != data_gen_len:
|
|
raise ValueError(
|
|
f"var length mismatch error: var_list = {var_len} vs data_generator = {data_gen_len}"
|
|
)
|
|
|
|
for i, ele in enumerate(user_parsed_line):
|
|
if len(ele[1]) == 0:
|
|
raise ValueError(
|
|
f"var length error: var {ele[0]}'s length in data_generator is 0"
|
|
)
|
|
|
|
if var_list[i].dtype == paddle.float32 and not all(
|
|
isinstance(ele, float) for ele in ele[1]
|
|
):
|
|
raise TypeError(
|
|
"var dtype mismatch error: var name = {}, var type in var_list = {}, while var in data_generator contains non-float value, which is {} \n"
|
|
"Please check if order of var_list and data_generator are aligned. \n"
|
|
"Please check if var's type in data_generator is correct.".format(
|
|
ele[0], "float", ele[1]
|
|
)
|
|
)
|
|
|
|
if (
|
|
var_list[i].dtype == paddle.int64
|
|
or var_list[i].dtype == paddle.int32
|
|
) and not all(isinstance(ele, int) for ele in ele[1]):
|
|
raise TypeError(
|
|
"var dtype mismatch error: var name = {}, var type in var_list = {}, while var in data_generator contains non-int value, which is {} \n"
|
|
"Please check if order of var_list and data_generator are aligned. \n"
|
|
"Please check if var's type in data_generator is correct.".format(
|
|
ele[0], "int", ele[1]
|
|
)
|
|
)
|
|
|
|
else:
|
|
break
|
|
|
|
f.close()
|
|
|
|
|
|
class InMemoryDataset(DatasetBase):
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
It will load data into memory and shuffle data before training.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
|
|
"""
|
|
|
|
dataset: core.Dataset
|
|
proto_desc: data_feed_pb2.DataFeedDesc
|
|
fleet_send_batch_size: int | None
|
|
is_user_set_queue_num: bool
|
|
queue_num: int | None
|
|
parse_ins_id: bool
|
|
parse_content: bool
|
|
parse_logkey: bool
|
|
merge_by_sid: bool
|
|
enable_pv_merge: bool
|
|
merge_by_lineid: bool
|
|
fleet_send_sleep_seconds: int | None
|
|
batch_size: int
|
|
thread_num: int
|
|
use_var: list[Tensor]
|
|
input_type: _InputType
|
|
fs_name: str
|
|
fs_ugi: str
|
|
pipe_command: str
|
|
download_cmd: str
|
|
data_feed_type: str
|
|
queue_num: int
|
|
merge_size: int
|
|
parse_ins_id: bool
|
|
parse_content: bool
|
|
fleet_send_batch_size: int
|
|
fleet_send_sleep_seconds: int
|
|
fea_eval: bool
|
|
candidate_size: int
|
|
|
|
def __init__(self) -> None:
|
|
"""Init."""
|
|
super().__init__()
|
|
self.proto_desc.name = "MultiSlotInMemoryDataFeed"
|
|
self.fleet_send_batch_size = None
|
|
self.is_user_set_queue_num = False
|
|
self.queue_num = None
|
|
self.parse_ins_id = False
|
|
self.parse_content = False
|
|
self.parse_logkey = False
|
|
self.merge_by_sid = True
|
|
self.enable_pv_merge = False
|
|
self.merge_by_lineid = False
|
|
self.fleet_send_sleep_seconds = None
|
|
|
|
def _init_distributed_settings(
|
|
self, **kwargs: Unpack[_InMemoryDatasetDistributedSettings]
|
|
) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
should be called only once in user's python scripts to initialize distributed-related settings of dataset instance
|
|
Args:
|
|
kwargs: Keyword arguments. Currently, we support following keys in **kwargs:
|
|
|
|
merge_size(int): ins size to merge, if merge_size > 0, set merge by line id,
|
|
instances of same line id will be merged after shuffle,
|
|
you should parse line id in data generator. default is -1.
|
|
parse_ins_id(bool): Set if Dataset need to parse ins_id. default is False.
|
|
parse_content(bool): Set if Dataset need to parse content. default is False.
|
|
fleet_send_batch_size(int): Set fleet send batch size in one rpc, default is 1024
|
|
fleet_send_sleep_seconds(int): Set fleet send sleep time, default is 0
|
|
fea_eval(bool): Set if Dataset need to do feature importance evaluation using slots shuffle.
|
|
default is False.
|
|
candidate_size(int): if fea_eval is set True, set the candidate size used in slots shuffle.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=[],
|
|
... )
|
|
|
|
>>> dataset._init_distributed_settings(
|
|
... parse_ins_id=True,
|
|
... parse_content=True,
|
|
... fea_eval=True,
|
|
... candidate_size=10000,
|
|
... )
|
|
|
|
"""
|
|
merge_size = kwargs.get("merge_size", -1)
|
|
if merge_size > 0:
|
|
self._set_merge_by_lineid(merge_size)
|
|
|
|
parse_ins_id = kwargs.get("parse_ins_id", False)
|
|
self._set_parse_ins_id(parse_ins_id)
|
|
|
|
parse_content = kwargs.get("parse_content", False)
|
|
self._set_parse_content(parse_content)
|
|
|
|
fleet_send_batch_size = kwargs.get("fleet_send_batch_size", None)
|
|
if fleet_send_batch_size:
|
|
self._set_fleet_send_batch_size(fleet_send_batch_size)
|
|
|
|
fleet_send_sleep_seconds = kwargs.get("fleet_send_sleep_seconds", None)
|
|
if fleet_send_sleep_seconds:
|
|
self._set_fleet_send_sleep_seconds(fleet_send_sleep_seconds)
|
|
|
|
fea_eval = kwargs.get("fea_eval", False)
|
|
if fea_eval:
|
|
candidate_size = kwargs.get("candidate_size", 10000)
|
|
self._set_fea_eval(candidate_size, True)
|
|
|
|
def update_settings(
|
|
self, **kwargs: Unpack[_InMemoryDatasetFullSettings]
|
|
) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
should be called in user's python scripts to update settings of dataset instance.
|
|
|
|
Args:
|
|
kwargs: Keyword arguments. Currently, we support following keys in **kwargs,
|
|
including single node settings and advanced distributed related settings:
|
|
batch_size(int): batch size. It will be effective during training. default is 1.
|
|
thread_num(int): thread num, it is the num of readers. default is 1.
|
|
use_var(list): list of variables. Variables which you will use. default is [].
|
|
input_type(int): the input type of generated input. 0 is for one sample, 1 is for one batch. default is 0.
|
|
fs_name(str): fs name. default is "".
|
|
fs_ugi(str): fs ugi. default is "".
|
|
pipe_command(str): pipe command of current dataset. A pipe command is a UNIX pipeline command that can be used only. default is "cat"
|
|
download_cmd(str): customized download command. default is "cat"
|
|
data_feed_type(str): data feed type used in c++ code. default is "MultiSlotInMemoryDataFeed".
|
|
queue_num(int): Dataset output queue num, training threads get data from queues. default is-1, which is set same as thread number in c++.
|
|
|
|
merge_size(int): ins size to merge, if merge_size > 0, set merge by line id,
|
|
instances of same line id will be merged after shuffle,
|
|
you should parse line id in data generator. default is -1.
|
|
parse_ins_id(bool): Set if Dataset need to parse ins_id. default is False.
|
|
parse_content(bool): Set if Dataset need to parse content. default is False.
|
|
fleet_send_batch_size(int): Set fleet send batch size in one rpc, default is 1024
|
|
fleet_send_sleep_seconds(int): Set fleet send sleep time, default is 0
|
|
fea_eval(bool): Set if Dataset need to do feature importance evaluation using slots shuffle.
|
|
default is False.
|
|
candidate_size(int): if fea_eval is set True, set the candidate size used in slots shuffle.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=[],
|
|
... )
|
|
>>> dataset._init_distributed_settings(
|
|
... parse_ins_id=True,
|
|
... parse_content=True,
|
|
... fea_eval=True,
|
|
... candidate_size=10000,
|
|
... )
|
|
>>> dataset.update_settings(batch_size=2)
|
|
|
|
"""
|
|
for key in kwargs:
|
|
if key == "pipe_command":
|
|
self._set_pipe_command(kwargs[key])
|
|
elif key == "batch_size":
|
|
self._set_batch_size(kwargs[key])
|
|
elif key == "thread_num":
|
|
self._set_thread(kwargs[key])
|
|
elif key == "use_var":
|
|
self._set_use_var(kwargs[key])
|
|
elif key == "input_type":
|
|
self._set_input_type(kwargs[key])
|
|
elif key == "fs_name" and "fs_ugi" in kwargs:
|
|
self._set_hdfs_config(kwargs[key], kwargs["fs_ugi"])
|
|
elif key == "download_cmd":
|
|
self._set_download_cmd(kwargs[key])
|
|
elif key == "merge_size" and kwargs.get("merge_size", -1) > 0:
|
|
self._set_merge_by_lineid(kwargs[key])
|
|
elif key == "parse_ins_id":
|
|
self._set_parse_ins_id(kwargs[key])
|
|
elif key == "parse_content":
|
|
self._set_parse_content(kwargs[key])
|
|
elif key == "fleet_send_batch_size":
|
|
self._set_fleet_send_batch_size(kwargs[key])
|
|
elif key == "fleet_send_sleep_seconds":
|
|
self._set_fleet_send_sleep_seconds(kwargs[key])
|
|
elif key == "fea_eval" and kwargs[key]:
|
|
candidate_size = kwargs.get("candidate_size", 10000)
|
|
self._set_fea_eval(candidate_size, True)
|
|
|
|
def init(self, **kwargs: Unpack[_InMemoryDatasetSettings]) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
should be called only once in user's python scripts to initialize settings of dataset instance
|
|
|
|
Args:
|
|
kwargs: Keyword arguments. Currently, we support following keys in **kwargs:
|
|
|
|
batch_size(int): batch size. It will be effective during training. default is 1.
|
|
thread_num(int): thread num, it is the num of readers. default is 1.
|
|
use_var(list): list of variables. Variables which you will use. default is [].
|
|
input_type(int): the input type of generated input. 0 is for one sample, 1 is for one batch. default is 0.
|
|
fs_name(str): fs name. default is "".
|
|
fs_ugi(str): fs ugi. default is "".
|
|
pipe_command(str): pipe command of current dataset. A pipe command is a UNIX pipeline command that can be used only. default is "cat"
|
|
download_cmd(str): customized download command. default is "cat"
|
|
data_feed_type(str): data feed type used in c++ code. default is "MultiSlotInMemoryDataFeed".
|
|
queue_num(int): Dataset output queue num, training threads get data from queues. default is -1, which is set same as thread number in c++.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> import os
|
|
>>> paddle.enable_static()
|
|
|
|
>>> with open("test_queue_dataset_run_a.txt", "w") as f:
|
|
... data = "2 1 2 2 5 4 2 2 7 2 1 3"
|
|
... f.write(data)
|
|
>>> with open("test_queue_dataset_run_b.txt", "w") as f:
|
|
... data = "2 1 2 2 5 4 2 2 7 2 1 3"
|
|
... f.write(data)
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> dataset.set_filelist(
|
|
... [
|
|
... "test_queue_dataset_run_a.txt",
|
|
... "test_queue_dataset_run_b.txt",
|
|
... ]
|
|
... )
|
|
>>> dataset.load_into_memory()
|
|
|
|
>>> place = paddle.CPUPlace()
|
|
>>> exe = paddle.static.Executor(place)
|
|
>>> startup_program = paddle.static.Program()
|
|
>>> main_program = paddle.static.Program()
|
|
>>> exe.run(startup_program)
|
|
|
|
>>> exe.train_from_dataset(main_program, dataset)
|
|
|
|
>>> os.remove("./test_queue_dataset_run_a.txt")
|
|
>>> os.remove("./test_queue_dataset_run_b.txt")
|
|
|
|
"""
|
|
batch_size = kwargs.get("batch_size", 1)
|
|
thread_num = kwargs.get("thread_num", 1)
|
|
use_var = kwargs.get("use_var", [])
|
|
input_type = kwargs.get("input_type", 0)
|
|
fs_name = kwargs.get("fs_name", "")
|
|
fs_ugi = kwargs.get("fs_ugi", "")
|
|
pipe_command = kwargs.get("pipe_command", "cat")
|
|
download_cmd = kwargs.get("download_cmd", "cat")
|
|
|
|
if self.use_ps_gpu:
|
|
data_feed_type = "SlotRecordInMemoryDataFeed"
|
|
else:
|
|
data_feed_type = "MultiSlotInMemoryDataFeed"
|
|
self._set_feed_type(data_feed_type)
|
|
|
|
super().init(
|
|
batch_size=batch_size,
|
|
thread_num=thread_num,
|
|
use_var=use_var,
|
|
pipe_command=pipe_command,
|
|
input_type=input_type,
|
|
fs_name=fs_name,
|
|
fs_ugi=fs_ugi,
|
|
download_cmd=download_cmd,
|
|
)
|
|
|
|
if kwargs.get("queue_num", -1) > 0:
|
|
queue_num = kwargs.get("queue_num", -1)
|
|
self._set_queue_num(queue_num)
|
|
|
|
def _set_feed_type(self, data_feed_type):
|
|
"""
|
|
Set data_feed_desc
|
|
"""
|
|
self.proto_desc.name = data_feed_type
|
|
if self.proto_desc.name == "SlotRecordInMemoryDataFeed":
|
|
self.dataset = core.Dataset("SlotRecordDataset")
|
|
|
|
def _prepare_to_run(self):
|
|
"""
|
|
Set data_feed_desc before load or shuffle,
|
|
user no need to call this function.
|
|
"""
|
|
if self.thread_num <= 0:
|
|
self.thread_num = 1
|
|
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)
|
|
self.dataset.set_parse_ins_id(self.parse_ins_id)
|
|
self.dataset.set_parse_content(self.parse_content)
|
|
self.dataset.set_parse_logkey(self.parse_logkey)
|
|
self.dataset.set_merge_by_sid(self.merge_by_sid)
|
|
self.dataset.set_enable_pv_merge(self.enable_pv_merge)
|
|
self.dataset.set_data_feed_desc(self._desc())
|
|
self.dataset.create_channel()
|
|
self.dataset.create_readers()
|
|
|
|
def _dynamic_adjust_before_train(self, thread_num):
|
|
if not self.is_user_set_queue_num:
|
|
if self.use_ps_gpu:
|
|
self.dataset.dynamic_adjust_channel_num(thread_num, True)
|
|
else:
|
|
self.dataset.dynamic_adjust_channel_num(thread_num, False)
|
|
self.dataset.dynamic_adjust_readers_num(thread_num)
|
|
|
|
def _dynamic_adjust_after_train(self):
|
|
if not self.is_user_set_queue_num:
|
|
if self.use_ps_gpu:
|
|
self.dataset.dynamic_adjust_channel_num(self.thread_num, True)
|
|
else:
|
|
self.dataset.dynamic_adjust_channel_num(self.thread_num, False)
|
|
self.dataset.dynamic_adjust_readers_num(self.thread_num)
|
|
|
|
def _set_queue_num(self, queue_num):
|
|
"""
|
|
Set Dataset output queue num, training threads get data from queues
|
|
|
|
Args:
|
|
queue_num(int): dataset output queue num
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset._set_queue_num(12)
|
|
|
|
"""
|
|
self.is_user_set_queue_num = True
|
|
self.queue_num = queue_num
|
|
|
|
def _set_parse_ins_id(self, parse_ins_id):
|
|
"""
|
|
Set if Dataset need to parse ins id
|
|
|
|
Args:
|
|
parse_ins_id(bool): if parse ins_id or not
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.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:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset._set_parse_content(True)
|
|
|
|
"""
|
|
self.parse_content = parse_content
|
|
|
|
def _set_fleet_send_batch_size(self, fleet_send_batch_size=1024):
|
|
"""
|
|
Set fleet send batch size, default is 1024
|
|
|
|
Args:
|
|
fleet_send_batch_size(int): fleet send batch size
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset._set_fleet_send_batch_size(800)
|
|
|
|
"""
|
|
self.fleet_send_batch_size = fleet_send_batch_size
|
|
|
|
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:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset._set_fleet_send_sleep_seconds(2)
|
|
|
|
"""
|
|
self.fleet_send_sleep_seconds = fleet_send_sleep_seconds
|
|
|
|
def _set_merge_by_lineid(self, merge_size=2):
|
|
"""
|
|
Set merge by line id, instances of same line id will be merged after
|
|
shuffle, you should parse line id in data generator.
|
|
|
|
Args:
|
|
merge_size(int): ins size to merge. default is 2.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset._set_merge_by_lineid()
|
|
|
|
"""
|
|
self.dataset.set_merge_by_lineid(merge_size)
|
|
self.merge_by_lineid = True
|
|
self.parse_ins_id = True
|
|
|
|
def _set_shuffle_by_uid(self, enable_shuffle_uid):
|
|
"""
|
|
Set if Dataset need to shuffle by uid.
|
|
|
|
Args:
|
|
set_shuffle_by_uid(bool): if shuffle according to uid or not
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset._set_shuffle_by_uid(True)
|
|
|
|
"""
|
|
self.dataset.set_shuffle_by_uid(enable_shuffle_uid)
|
|
|
|
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
|
|
)
|
|
|
|
def set_date(self, date: str) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
Set training date for pull sparse parameters, saving and loading model. Only used in psgpu
|
|
|
|
Args:
|
|
date(str): training date(format : YYMMDD). eg.20211111
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> dataset.set_date("20211111")
|
|
|
|
"""
|
|
year = int(date[:4])
|
|
month = int(date[4:6])
|
|
day = int(date[6:])
|
|
if self.use_ps_gpu and core._is_compiled_with_heterps():
|
|
self.psgpu.set_date(year, month, day)
|
|
|
|
def tdm_sample(
|
|
self,
|
|
tree_name: str,
|
|
tree_path: str,
|
|
tdm_layer_counts: list[int],
|
|
start_sample_layer: int,
|
|
with_hierarchy: bool,
|
|
seed: int,
|
|
id_slot: int,
|
|
) -> None:
|
|
self.dataset.tdm_sample(
|
|
tree_name,
|
|
tree_path,
|
|
tdm_layer_counts,
|
|
start_sample_layer,
|
|
with_hierarchy,
|
|
seed,
|
|
id_slot,
|
|
)
|
|
|
|
def load_into_memory(self, is_shuffle: bool = False) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
Load data into memory
|
|
|
|
Args:
|
|
is_shuffle(bool): whether to use local shuffle, default is False
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> filelist = ["a.txt", "b.txt"]
|
|
>>> dataset.set_filelist(filelist)
|
|
>>> dataset.load_into_memory()
|
|
|
|
"""
|
|
self._prepare_to_run()
|
|
if not self.use_ps_gpu:
|
|
self.dataset.load_into_memory()
|
|
elif core._is_compiled_with_heterps():
|
|
self.psgpu.set_dataset(self.dataset)
|
|
self.psgpu.load_into_memory(is_shuffle)
|
|
|
|
def preload_into_memory(self, thread_num: int | None = None) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
Load data into memory in async mode
|
|
|
|
Args:
|
|
thread_num(int): preload thread num
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> filelist = ["a.txt", "b.txt"]
|
|
>>> dataset.set_filelist(filelist)
|
|
>>> dataset.preload_into_memory()
|
|
>>> dataset.wait_preload_done()
|
|
|
|
"""
|
|
self._prepare_to_run()
|
|
if thread_num is None:
|
|
thread_num = self.thread_num
|
|
self.dataset.set_preload_thread_num(thread_num)
|
|
self.dataset.create_preload_readers()
|
|
self.dataset.preload_into_memory()
|
|
|
|
def wait_preload_done(self) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
Wait preload_into_memory done
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> filelist = ["a.txt", "b.txt"]
|
|
>>> dataset.set_filelist(filelist)
|
|
>>> dataset.preload_into_memory()
|
|
>>> dataset.wait_preload_done()
|
|
|
|
"""
|
|
self.dataset.wait_preload_done()
|
|
self.dataset.destroy_preload_readers()
|
|
|
|
def local_shuffle(self) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
Local shuffle
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> 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: Fleet | None = None, thread_num: int = 12
|
|
) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
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:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> filelist = ["a.txt", "b.txt"]
|
|
>>> dataset.set_filelist(filelist)
|
|
>>> dataset.load_into_memory()
|
|
>>> dataset.global_shuffle()
|
|
|
|
Args:
|
|
fleet(Fleet): fleet singleton. Default None.
|
|
thread_num(int): shuffle thread num. Default is 12.
|
|
|
|
"""
|
|
trainer_num = 1
|
|
if fleet is not None:
|
|
fleet._role_maker.barrier_worker()
|
|
trainer_num = fleet.worker_num()
|
|
if self.fleet_send_batch_size is None:
|
|
self.fleet_send_batch_size = 1024
|
|
if self.fleet_send_sleep_seconds is None:
|
|
self.fleet_send_sleep_seconds = 0
|
|
self.dataset.register_client2client_msg_handler()
|
|
self.dataset.set_trainer_num(trainer_num)
|
|
self.dataset.set_fleet_send_batch_size(self.fleet_send_batch_size)
|
|
self.dataset.set_fleet_send_sleep_seconds(self.fleet_send_sleep_seconds)
|
|
if fleet is not None:
|
|
fleet._role_maker.barrier_worker()
|
|
self.dataset.global_shuffle(thread_num)
|
|
if fleet is not None:
|
|
fleet._role_maker.barrier_worker()
|
|
if self.merge_by_lineid:
|
|
self.dataset.merge_by_lineid()
|
|
if fleet is not None:
|
|
fleet._role_maker.barrier_worker()
|
|
|
|
def release_memory(self) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
Release InMemoryDataset memory data, when data will not be used again.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> filelist = ["a.txt", "b.txt"]
|
|
>>> dataset.set_filelist(filelist)
|
|
>>> dataset.load_into_memory()
|
|
>>> dataset.global_shuffle()
|
|
>>> exe = paddle.static.Executor(paddle.CPUPlace())
|
|
>>> startup_program = paddle.static.Program()
|
|
>>> main_program = paddle.static.Program()
|
|
>>> exe.run(startup_program)
|
|
>>> exe.train_from_dataset(main_program, dataset)
|
|
>>> dataset.release_memory()
|
|
|
|
"""
|
|
self.dataset.release_memory()
|
|
|
|
def get_memory_data_size(self, fleet: Fleet | None = None) -> int:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
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|None): Fleet Object.
|
|
|
|
Returns:
|
|
The size of memory data.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
|
|
>>> filelist = ["a.txt", "b.txt"]
|
|
>>> dataset.set_filelist(filelist)
|
|
>>> dataset.load_into_memory()
|
|
>>> print(dataset.get_memory_data_size())
|
|
|
|
"""
|
|
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.all_reduce_worker(
|
|
local_data_size, global_data_size
|
|
)
|
|
return global_data_size[0]
|
|
return local_data_size[0]
|
|
|
|
def get_shuffle_data_size(self, fleet: Fleet | None = None) -> int:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
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|None): Fleet Object.
|
|
|
|
Returns:
|
|
The size of shuffle data.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
|
|
>>> filelist = ["a.txt", "b.txt"]
|
|
>>> dataset.set_filelist(filelist)
|
|
>>> dataset.load_into_memory()
|
|
>>> dataset.global_shuffle()
|
|
>>> print(dataset.get_shuffle_data_size())
|
|
|
|
"""
|
|
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.all_reduce_worker(
|
|
local_data_size, global_data_size
|
|
)
|
|
return global_data_size[0]
|
|
return local_data_size[0]
|
|
|
|
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
|
|
fea_eval(bool): whether enable fea eval mode to enable slots shuffle.
|
|
default is True.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
>>> dataset = paddle.distributed.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: list[str]) -> None:
|
|
"""
|
|
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:
|
|
.. code-block:: pycon
|
|
|
|
>>> # doctest: +SKIP('No files to read')
|
|
>>> import paddle
|
|
>>> paddle.enable_static()
|
|
|
|
>>> dataset = paddle.distributed.InMemoryDataset()
|
|
>>> dataset._init_distributed_settings(fea_eval=True)
|
|
>>> slots = ["slot1", "slot2", "slot3", "slot4"]
|
|
>>> slots_vars = []
|
|
>>> for slot in slots:
|
|
... var = paddle.static.data(
|
|
... name=slot,
|
|
... shape=[None, 1],
|
|
... dtype="int64",
|
|
... lod_level=1,
|
|
... )
|
|
... slots_vars.append(var)
|
|
>>> dataset.init(
|
|
... batch_size=1,
|
|
... thread_num=2,
|
|
... input_type=1,
|
|
... pipe_command="cat",
|
|
... use_var=slots_vars,
|
|
... )
|
|
>>> filelist = ["a.txt", "b.txt"]
|
|
>>> dataset.set_filelist(filelist)
|
|
>>> dataset.load_into_memory()
|
|
>>> dataset.slots_shuffle(['slot1'])
|
|
"""
|
|
if self.fea_eval:
|
|
slots_set = set(slots)
|
|
self.dataset.slots_shuffle(slots_set)
|
|
|
|
|
|
class QueueDataset(DatasetBase):
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
QueueDataset, it will process data streamly.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.QueueDataset()
|
|
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""
|
|
Initialize QueueDataset
|
|
"""
|
|
super().__init__()
|
|
self.proto_desc.name = "MultiSlotDataFeed"
|
|
|
|
def init(self, **kwargs: Unpack[_DatasetBaseSettings]) -> None:
|
|
"""
|
|
:api_attr: Static Graph
|
|
|
|
should be called only once in user's python scripts to initialize settings of dataset instance
|
|
|
|
"""
|
|
super().init(**kwargs)
|
|
|
|
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()
|
|
|
|
|
|
class FileInstantDataset(DatasetBase):
|
|
"""
|
|
FileInstantDataset, it will process data streamly.
|
|
|
|
Examples:
|
|
.. code-block:: pycon
|
|
|
|
>>> import paddle
|
|
>>> dataset = paddle.distributed.fleet.FileInstantDataset()
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""
|
|
Initialize FileInstantDataset
|
|
"""
|
|
super().__init__()
|
|
self.proto_desc.name = "MultiSlotFileInstantDataFeed"
|
|
|
|
def init(self, **kwargs: Unpack[_DatasetBaseSettings]) -> None:
|
|
"""
|
|
should be called only once in user's python scripts to initialize settings of dataset instance
|
|
"""
|
|
super().init(**kwargs)
|