python - 使用 tf.contrib.data.parallel_interleave 并行化 tf.from_generator
问题描述
我有一堆 JSON 数组文件(准确地说是 AVRO),每个文件都会产生多个用于训练 Keras 模型的样本。使用来自@GPhilo和@jsimsa的想法,我能够想出这个来并行化我的输入管道。无法弄清楚如何设计generator(n)
来划分处理文件的工作。代码在内部失败,parse_file(f)
因为函数需要字符串文件路径而不是Tensor
,
N = num_cores = 2
files_to_process = ["f1.avro", "f2.avro", "f3.avro"]
shuffle_size = prefetch_buffer = 1000
batch_size = 512
def generator(n):
size = math.ceil(len(files_to_process) / N)
start_index = n * size
end_index = start_index + size
def gen():
# for f in files_to_process[start_index:end_index]:
for f in tf.slice(files_to_process, start_index, size):
yield f
return gen
def dataset(n):
return tf.data.Dataset.from_generator(generator(n), (tf.string,))
def process_file(f):
examples_x, examples_y = parse_file(f)
return examples_x, examples_y
ds = tf.data.Dataset.range(N)
ds = ds.apply(tf.contrib.data.parallel_interleave(dataset, cycle_length=N))
ds = ds.map(process_file, num_parallel_calls=N)
ds = ds.prefetch(prefetch_buffer)
ds = ds.flat_map(lambda *x: tf.data.Dataset.from_tensor_slices(x))
ds = ds.batch(batch_size).shuffle(shuffle_size)
...
myTfKerasModel.fit( ds.make_one_iterator(), NUM_TRAIN_SAMPLES // batch_size )
generator(n)
这里的正确设计方法是什么- 这是使用
parallel_interleave
和设计我的输入管道的优化方法吗?flat_map
解决方案
在我看来,您使用发电机不必要地使您的生活复杂化。这就是我实现输入管道的方式:
def parse_file_tf(filename):
return tf.py_func(parse_file, [filename], [tf.float32, tf.float32])
# version with map
files = tf.data.Dataset.from_tensor_slices(files_to_process)
dataset = files.map(parse_file_tf, num_parallel_calls=N)
dataset = dataset.flat_map(lambda *x: tf.data.Dataset.from_tensor_slices(x))
dataset = dataset.batch(batch_size).shuffle(shuffle_size).prefetch(2)
it = dataset.make_one_shot_iterator()
为了测试它,我将一个假人定义parse_file
为:
i=0
def parse_file(f):
global i
i += 1
return np.asarray([i]*i, dtype=np.float32), np.asarray([i]*i, dtype=np.float32) # mimicks variable-length examples_x, examples_y
我将其输入到一个基本循环中,该循环显示了迭代器返回的内容:
sess = tf.Session()
try:
while True:
x, y = it.get_next()
vx, vy = sess.run([x,y])
print(vx)
print(vy)
except tf.errors.OutOfRangeError:
pass
sess.close()
运行上面的代码打印:
[2. 3. 2. 1. 3. 3.]
[2. 3. 2. 1. 3. 3.]
管道的解释
本质上,我将并行化问题留给map
,在那里我可以传递它应该运行的线程数。不需要生成器迭代范围和那些额外的复杂性。
我之所以选择 mapparallel_interleave
是因为后者要求您为它返回的每个项目生成一个Dataset
实例,在您的情况下这并没有真正意义,因为您在运行时已经将所有值加载到内存中parse_file
。
如果您缓慢生成值(例如,通过应用到文件名列表)parallel_interleave
是有意义的,但如果您的数据集适合内存,则使用.tf.data.TFRecordDataset
map
关于tf.py_func
限制,它们不会影响您训练的网络,只会影响输入管道。理想情况下,您的训练和网络的最终使用会有不同的管道。您只需要注意后者期间的限制,而对于培训(除非您对分布式培训和/或跨机器移动培训做了非常具体的事情),您是相当安全的。
带发电机的版本
如果您的 JSON 文件非常大并且它们的内容不适合内存,您可以使用生成器,但与您开始使用的方法略有不同。这个想法是,生成器遍历 JSON 文件并一次生成yield
一条记录。然后,生成器必须是您的parse_file
功能。例如,假设您有以下parse_file
生成器:
i = 3
def parse_file(filename):
global i
i += 1
ctr = 0
while ctr < i:
yield ctr, ctr
在这种情况下,管道将如下所示:
def wrap_generator(filename):
return tf.data.Dataset.from_generator(parse_file(filename), [tf.int32, tf.int32])
files = tf.data.Dataset.from_tensor_slices(files_to_process)
dataset = files.apply(tf.contrib.data.parallel_interleave(wrap_generator, cycle_length=N))
dataset = dataset.flat_map(lambda *x: tf.data.Dataset.from_tensor_slices(x))
dataset = dataset.shuffle(shuffle_size).batch(batch_size).prefetch(2)
it = dataset.make_one_shot_iterator()
请注意,这里我们需要使用parallel_interleave
,因为我们将生成器转换为Dataset
从中提取值的实例。其余的保持不变。
将其馈送到与上述相同的示例循环打印:
[6. 5. 4. 4. 6. 5. 6. 6. 5. 4. 6. 4. 5. 5. 6.]
[6. 5. 4. 4. 6. 5. 6. 6. 5. 4. 6. 4. 5. 5. 6.]
推荐阅读
- java - 如何在 11 Java JDK 版本中解决这个 JavaFX 问题?
- erlang - 混合 ecto.create 时出现 argon2 elixir 编译问题
- web - 404 在backbone.jsI 中实例化Web Worker 时未找到
- maven - 当涉及到 scala.js 和 IntelliJ IDE 时,sbt 是唯一的选择吗?
- r - 使用 fviz_dend from factoextra 和 ggplotly 时在悬停时显示点标签
- html - Shopify - 自定义购物车属性未显示在订单 API 的 note_attributes 中
- ssms - SSMS 扩展特定数据库
- javascript - 如何在不改变函数结构的情况下保持上下文?
- python - 在 Python 中将点云拆分为特定大小的单元格
- javascript - 从下一页单击时隐藏 div