如何在TensorFlow中使用parallel_interleave

Question

我正在阅读TensorFlow 基准测试报告中的代码.以下代码是从TFRecord文件创建TensorFlow数据集的部分:

ds = tf.data.TFRecordDataset.list_files(tfrecord_file_names)
ds = ds.apply(interleave_ops.parallel_interleave(tf.data.TFRecordDataset, cycle_length=10))

我试图更改此代码以直接从JPEG图像文件创建数据集:

ds = tf.data.Dataset.from_tensor_slices(jpeg_file_names)
ds = ds.apply(interleave_ops.parallel_interleave(?, cycle_length=10))

我不知道写什么？地点.parallel_interleave()中的map_func是TF_cord文件的tf.data.TFRecordDataset类的__init __(),但我不知道要为JPEG文件写什么.

我们不需要在这里进行任何转换.因为我们将压缩两个数据集,然后再进行转换.代码如下:

counter = tf.data.Dataset.range(batch_size)
ds = tf.data.Dataset.zip((ds, counter))
ds = ds.apply( \
     batching.map_and_batch( \
     map_func=preprocess_fn, \
     batch_size=batch_size, \
     num_parallel_batches=num_splits))

因为我们不需要改造吗？我尝试使用空的map_func,但是有错误"map_func must return aDataset` object".我也尝试使用tf.data.Dataset,但是输出说Dataset是一个不允许放在那里的抽象类.

任何人都可以帮忙吗？非常感谢.

Answer 1

parallel_interleave当您具有将源数据集的每个元素转换为多个元素到目标数据集的转换时非常有用.我不确定为什么他们会在那些基准测试中使用它,当他们可以使用map并行调用时.

以下是我建议parallel_interleave用于从多个目录中读取图像的方法,每个目录包含一个类:

classes = sorted(glob(directory + '/*/')) # final slash selects directories only
num_classes = len(classes)

labels = np.arange(num_classes, dtype=np.int32)

dirs = DS.from_tensor_slices((classes, labels))               # 1
files = dirs.apply(tf.contrib.data.parallel_interleave(
    get_files, cycle_length=num_classes, block_length=4,      # 2
    sloppy=False)) # False is important ! Otherwise it mixes labels
files = files.cache()
imgs = files.map(read_decode, num_parallel_calls=20)\.        # 3
            .apply(tf.contrib.data.shuffle_and_repeat(100))\
            .batch(batch_size)\
            .prefetch(5)

有三个步骤.首先,我们获取目录及其标签列表(#1).

然后,我们将这些映射到文件的数据集.但是如果我们做一个简单的操作.flatmap(),我们将最终得到标签0的所有文件1,然后是标签的所有文件,然后2等等......然后我们需要非常大的shuffle缓冲区来获得有意义的随机播放.

所以,相反,我们应用parallel_interleave(#2).这是get_files():

def get_files(dir_path, label):
    globbed = tf.string_join([dir_path, '*.jpg'])
    files = tf.matching_files(globbed)

    num_files = tf.shape(files)[0] # in the directory
    labels = tf.tile([label], [num_files, ]) # expand label to all files
    return DS.from_tensor_slices((files, labels))

使用parallel_interleave确保list_files每个目录的并行运行,因此当第block_length一个目录从第一个目录中列出时,第二个目录中的第一个block_length文件也将可用(也来自第3个,第4个等).此外,结果数据集将包含每个标签的交错块,例如1 1 1 1 2 2 2 2 3 3 3 3 3 1 1 1 1 ...(对于3个类别block_length=4)

最后,我们从文件列表中读取图像(#3).这是read_and_decode():

def read_decode(path, label):
    img = tf.image.decode_image(tf.read_file(path), channels=3)
    img = tf.image.resize_bilinear(tf.expand_dims(img, axis=0), target_size)
    img = tf.squeeze(img, 0)
    img = preprocess_fct(img) # should work with Tensors !

    label = tf.one_hot(label, num_classes)
    img = tf.Print(img, [path, label], 'Read_decode')
    return (img, label)

此函数采用图像路径及其标签,并为每个:路径的图像张量和标签的one_hot编码返回张量.这也是您可以对图像进行所有转换的地方.在这里,我做了调整大小和基本的预处理.