InMemoryDataset¶

class paddle.distributed. InMemoryDataset [source]

Api_attr: Static Graph

It will load data into memory and shuffle data before training.

Examples

           import paddle
paddle.enable_static()
dataset = paddle.distributed.InMemoryDataset()

          

update_settings ( **kwargs )
update_settings¶

Api_attr: Static Graph

should be called in user’s python scripts to update setings of dataset instance.

Parameters

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

             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)

            

init ( **kwargs )
init¶

Api_attr: Static Graph

should be called only once in user’s python scripts to initialize setings of dataset instance

Parameters

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

             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")

            

set_date ( date ) set_date¶

Api_attr: Static Graph

Set training date for pull sparse parameters, saving and loading model. Only used in psgpu

Parameters: date (str) – training date(format : YYMMDD). eg.20211111

Examples

             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")

            

load_into_memory ( is_shuffle=False ) load_into_memory¶

Api_attr: Static Graph

Load data into memory

Parameters: is_shuffle (bool) – whether to use local shuffle, default is False

Examples

             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()

            

preload_into_memory ( thread_num=None ) preload_into_memory¶

Api_attr: Static Graph

Load data into memory in async mode

Parameters: thread_num (int) – preload thread num

Examples

             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()

            

wait_preload_done ( ) wait_preload_done¶

Api_attr: Static Graph

Wait preload_into_memory done

Examples

             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()

            

local_shuffle ( ) local_shuffle¶

Api_attr: Static Graph

Local shuffle

Examples

             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()

            

global_shuffle ( fleet=None, thread_num=12 ) global_shuffle¶

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

             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()

            

Parameters

fleet (Fleet) – fleet singleton. Default None.
thread_num (int) – shuffle thread num. Default is 12.

release_memory ( ) release_memory¶

Api_attr: Static Graph

Release InMemoryDataset memory data, when data will not be used again.

Examples

             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()

            

get_memory_data_size ( fleet=None ) get_memory_data_size¶

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

Parameters: fleet (Fleet) – Fleet Object.
Returns: The size of memory data.

Examples

             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()

            

get_shuffle_data_size ( fleet=None ) get_shuffle_data_size¶

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.

Parameters: fleet (Fleet) – Fleet Object.
Returns: The size of shuffle data.

Examples

             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()

            

slots_shuffle ( slots ) slots_shuffle¶

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).

Parameters: slots (list[string]) – the set of slots(string) to do slots shuffle.

Examples

             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'])

            

set_filelist ( filelist ) set_filelist¶

Set file list in current worker. The filelist is indicated by a list of file names (string).

Examples

             import paddle
dataset = paddle.distributed.fleet.DatasetBase()
dataset.set_filelist(['a.txt', 'b.txt'])

            

Parameters: filelist (list[str]) – list of file names of inputs.