我有一个非常简单的输入管道,from_generator非常适合......
dataset = tf.data.Dataset.from_generator(complex_img_label_generator,
(tf.int32, tf.string))
dataset = dataset.batch(64)
iter = dataset.make_one_shot_iterator()
imgs, labels = iter.get_next()
其中complex_img_label_generator动态生成图像,并返回表示一个numpy的阵列(H, W, 3)图像和一个简单的string标签.处理不是我可以表示从文件和tf.image操作中读取的内容.
我的问题是关于如何平衡发电机？我如何让N个这些生成器在自己的线程中运行.
一个想法是使用dataset.map与num_parallel_calls处理线程; 但是地图在张量上运行......另一个想法是创建多个生成器,每个生成器都有自己的,prefetch并以某种方式加入它们,但我看不出我如何加入N个生成器流？
我可以遵循任何规范的例子吗？
我正在将TensorFlow代码从旧队列接口更改为新的Dataset API.使用旧接口,我可以指定队列的num_threads参数tf.train.shuffle_batch.但是,控制数据集API中线程数量的唯一方法似乎是map使用num_parallel_calls参数的函数.但是,我正在使用该flat_map函数,它没有这样的参数.
问题:有没有办法控制flat_map函数的线程/进程数？或者是否可以map结合使用flat_map并仍然指定并行呼叫的数量？
请注意,并行运行多个线程至关重要,因为我打算在数据进入队列之前在CPU上运行大量预处理.
GitHub上有两个(这里和这里)相关的帖子,但我不认为他们回答了这个问题.
这是我用例的最小代码示例:
with tf.Graph().as_default():
data = tf.ones(shape=(10, 512), dtype=tf.float32, name="data")
input_tensors = (data,)
def pre_processing_func(data_):
# normally I would do data-augmentation here
results = (tf.expand_dims(data_, axis=0),)
return tf.data.Dataset.from_tensor_slices(results)
dataset_source = tf.data.Dataset.from_tensor_slices(input_tensors)
dataset = dataset_source.flat_map(pre_processing_func)
# do something with 'dataset'