QueueyDataset

class paddle.distributed. QueueDataset [源代码]

QueueyDataset是流式处理数据使用Dataset类。与InmemoryDataset继承自同一父类，用于单机训练，不支持分布式大规模参数服务器相关配置和shuffle。此类由paddle.distributed.QueueDataset直接创建。

代码示例:

  1. import paddle
  2. dataset = paddle.distributed.QueueDataset()

init ( **kwargs )

注意：

1. 该API只在非 Dygraph 模式下生效

对QueueDataset的实例进行配置初始化。

参数：

  • kwargs - 可选的关键字参数，由调用者提供， 目前支持以下关键字配置。

  • batch_size (int) - batch size的大小. 默认值为1。

  • thread_num (int) - 用于训练的线程数, 默认值为1。

  • use_var (list) - 用于输入的variable列表，默认值为[]。

  • input_type (int) - 输入到模型训练样本的类型. 0 代表一条样本, 1 代表一个batch。 默认值为0。

  • fs_name (str) - hdfs名称. 默认值为””。

  • fs_ugi (str) - hdfs的ugi. 默认值为””。

  • pipe_command (str) - 在当前的 dataset 中设置的pipe命令用于数据的预处理。pipe命令只能使用UNIX的pipe命令，默认为”cat”。

  • download_cmd (str) - 数据下载pipe命令。 pipe命令只能使用UNIX的pipe命令, 默认为”cat”。

返回：None。

代码示例

  1. import paddle
  2. import os
  4. paddle.enable_static()
  6. with open("test_queue_dataset_run_a.txt", "w") as f:
  7.     data = "2 1 2 2 5 4 2 2 7 2 1 3n"
  8.     data += "2 6 2 2 1 4 2 2 4 2 2 3n"
  9.     data += "2 5 2 2 9 9 2 2 7 2 1 3n"
  10.     data += "2 7 2 2 1 9 2 3 7 2 5 3n"
  11.     f.write(data)
  12. with open("test_queue_dataset_run_b.txt", "w") as f:
  13.     data = "2 1 2 2 5 4 2 2 7 2 1 3n"
  14.     data += "2 6 2 2 1 4 2 2 4 2 2 3n"
  15.     data += "2 5 2 2 9 9 2 2 7 2 1 3n"
  16.     data += "2 7 2 2 1 9 2 3 7 2 5 3n"
  17.     f.write(data)
  19. slots = ["slot1", "slot2", "slot3", "slot4"]
  20. slots_vars = []
  21. for slot in slots:
  22.     var = paddle.static.data(
  23.         name=slot, shape=[None, 1], dtype="int64", lod_level=1)
  24.     slots_vars.append(var)
  26. dataset = paddle.distributed.QueueDataset()
  27. dataset.init(
  28.     batch_size=1,
  29.     thread_num=2,
  30.     input_type=1,
  31.     pipe_command="cat",
  32.     use_var=slots_vars)
  33. dataset.set_filelist(
  34.     ["test_queue_dataset_run_a.txt", "test_queue_dataset_run_b.txt"])
  36. paddle.enable_static()
  38. place = paddle.CPUPlace()
  39. exe = paddle.static.Executor(place)
  40. startup_program = paddle.static.Program()
  41. main_program = paddle.static.Program()
  42. exe.run(startup_program)
  44. exe.train_from_dataset(main_program, dataset)
  46. os.remove("./test_queue_dataset_run_a.txt")
  47. os.remove("./test_queue_dataset_run_b.txt")

set_filelist ( filelist )

在当前的worker中设置文件列表。

代码示例:

  1. import paddle
  2. import os
  4. paddle.enable_static()
  6. with open("test_queue_dataset_run_a.txt", "w") as f:
  7.     data = "2 1 2 2 5 4 2 2 7 2 1 3n"
  8.     data += "2 6 2 2 1 4 2 2 4 2 2 3n"
  9.     data += "2 5 2 2 9 9 2 2 7 2 1 3n"
  10.     data += "2 7 2 2 1 9 2 3 7 2 5 3n"
  11.     f.write(data)
  12. with open("test_queue_dataset_run_b.txt", "w") as f:
  13.     data = "2 1 2 2 5 4 2 2 7 2 1 3n"
  14.     data += "2 6 2 2 1 4 2 2 4 2 2 3n"
  15.     data += "2 5 2 2 9 9 2 2 7 2 1 3n"
  16.     data += "2 7 2 2 1 9 2 3 7 2 5 3n"
  17.     f.write(data)
  18. dataset = paddle.distributed.QueueDataset()
  19. slots = ["slot1", "slot2", "slot3", "slot4"]
  20. slots_vars = []
  21. for slot in slots:
  22.     var = paddle.static.data(
  23.         name=slot, shape=[None, 1], dtype="int64", lod_level=1)
  24.     slots_vars.append(var)
  25. dataset.init(
  26.     batch_size=1,
  27.     thread_num=2,
  28.     input_type=1,
  29.     pipe_command="cat",
  30.     use_var=slots_vars)
  31. filelist = ["a.txt", "b.txt"]
  32. dataset.set_filelist(filelist)
  33. os.remove("./test_queue_dataset_run_a.txt")
  34. os.remove("./test_queue_dataset_run_b.txt")

参数：

  • filelist (list[string]) - 文件列表