如何在张量流模型中的tf.data管道中提供.h5文件

Question

我正在尝试使用tf.data优化.h5数据的输入管道。但我遇到了TypeError: expected str, bytes or os.PathLike object, not Tensor。我做了一项研究，但找不到任何关于将一个字符串转换为字符串的信息。

此简化代码是可执行的并返回相同的错误：

batch_size = 1000
conv_size = 3
nb_conv = 32
learning_rate = 0.0001

# define parser function
def parse_function(fname):
    with h5py.File(fname, 'r') as f: #Error comes from here
        X = f['X'].reshape(batch_size, patch_size, patch_size, 1)
        y = f['y'].reshape(batch_size, patch_size, patch_size, 1)
        return X, y

# create a list of files path
flist = []
for dirpath, _, fnames in os.walk('./proc/'):
    for fname in fnames:
        if fname.startswith('{}_{}'.format(patch_size, batch_size)) and fname.endswith('h5'):
            flist.append(fname)

# prefetch data
dataset = tf.data.Dataset.from_tensor_slices((flist))
dataset = dataset.shuffle(len(flist))
dataset = dataset.map(parse_function, num_parallel_calls=4)
dataset = dataset.batch(1)
dataset = dataset.prefetch(3)

# simplest model that I think of
X_ph = tf.placeholder(tf.float32, shape=None)
y_ph = tf.placeholder(tf.float32, shape=None)
W = tf.get_variable('w', shape=[conv_size, conv_size, 1, 1], initializer=tf.contrib.layers.xavier_initializer())
loss = tf.reduce_mean(tf.losses.mean_squared_error(tf.nn.softmax(labels=y_ph, predictions=tf.matmul(X_ph, W))))
train_op = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(loss)

# start session
with tf.Session() as sess:
    sess.run(tf.global_variables_initializer())
    print(sess.run(train_op, feed_dict={X_ph: dataset[0], y_ph: dataset[1]}))

显然，fname是一个字符串的张量，但位置参数只等待一个字符串。我找不到任何关于此的文件。而another post的答案并没有解决这个问题。在我的情况下，我只使用h5，其中一个h5存储一批。

---

更新解决方案：感谢@kvish的评论，加载.h5文件的部分已解决。使用简单的转换层升级代码，占位符已被占用。每个.h5都是一批。我希望以多个批次并行预取（h5py不支持多线程读取，因此我将批处理写入多个文件）。可以复制粘贴和启动：

import h5py
import threading
import numpy as np
import tensorflow as tf

# generate some img data
for i in range(5):
    with h5py.File('./test_{}.h5'.format(i), 'w') as f:
        f.create_dataset('X', shape=(1000, 100, 100), dtype='float32', data=np.random.rand(10**7).reshape(1000, 100, 100))
        f.create_dataset('y', shape=(1000, 100, 100), dtype='float32', data=np.random.rand(10**7).reshape(1000, 100, 100))
        print(threading.get_ident())

# params
num_cores = 3
shuffle_size = 1
batch_size = 1

# read .h5 file
def parse_file(f):
    print(f.decode('utf-8'))
    with h5py.File(f.decode("utf-8"), 'r') as fi:
        X = fi['X'][:].reshape(1000, 100, 100, 1)
        y = fi['y'][:].reshape(1000, 100, 100, 1)
        print(threading.get_ident())  # to see the thread id
        return X, y

# py_func wrapper
def parse_file_tf(filename):
    return tf.py_func(parse_file, [filename], [tf.float32, tf.float32])

# tf.data input pipeline
files = tf.data.Dataset.list_files('./test_*.h5')
dataset = files.map(parse_file_tf, num_parallel_calls=num_core)
dataset = dataset.batch(batch_size).shuffle(shuffle_size).prefetch(3)
it = dataset.make_initializable_iterator()
iter_init_op = it.initializer
X_it, y_it = it.get_next()

# simplest model that I can think of 
with tf.name_scope("Conv1"):
    W = tf.get_variable("W", shape=[3, 3, 1, 1],
                         initializer=tf.contrib.layers.xavier_initializer())
    b = tf.get_variable("b", shape=[1], initializer=tf.contrib.layers.xavier_initializer())
    layer1 = tf.nn.conv2d(X_it, W, strides=[1, 1, 1, 1], padding='SAME') + b
    out = tf.nn.relu(layer1)

loss = tf.reduce_mean(tf.losses.mean_squared_error(labels=y_it, predictions=out))
train_op = tf.train.AdamOptimizer(learning_rate=0.0001).minimize(loss)

# session
sess = tf.Session()
sess.run(tf.global_variables_initializer())
sess.run(iter_init_op)
sess.run([train_op])
sess.close()

---

不知何故会有另一个与这篇文章无关的cudnn问题。

tensorflow-cpu v1.12：工作正常

tensorflow-gpu v1.12：发生运行时问题

回溯（最近一次调用最后一次）：文件“/envs/tf/lib/python3.6/site-packages/tensorflow/python/client/session.py”，第1334行，在_do_call中返回fn（* args）文件“/ envs / tf / lib / python3.6 / site-packages / tensorflow / python / client / session.py“，第1319行，_run_fn选项，feed_dict，fetch_list，target_list，run_metadata）文件”/ envs / tf / lib / python3 .6 / site-packages / tensorflow / python / client / session.py“，第1407行，在_call_tf_sessionrun run_metadata中）tensorflow.python.framework.errors_impl.NotFoundError：没有算法工作！ [[{{node Conv1 / Conv2D}} = Conv2D [T = DT_FLOAT，data_format =“NCHW”，dilations = [1,1,1,1]，padding =“SAME”，strides = [1,1,1， 1]，use_cudnn_on_gpu = true，_device =“/ job：localhost / replica：0 / task：0 / device：GPU：0”]（gradients / Conv1 / Conv2D_grad / Conv2DBackpropFilter-0-TransposeNHWCToNCHW-LayoutOptimizer，W / read）] ] [[{{node mean_squared_error / num_present / broadcast_weights / assert_broadcastable / AssertGuard / Assert / Switch_2 / _37}} = _Recvclient_terminated = false，recv_device =“/ job：localhost / replica：0 / task：0 / device：CPU：0” ，send_device =“/ job：localhost / replica：0 / task：0 / device：GPU：0”，send_device_incarnation = 1，tensor_name =“edge_63_me ... t / Switch_2”，tensor_type = DT_INT32，_device =“/ job： localhost / replica：0 / task：0 / device：CPU：0“]] tensorflow-cpu v1.12：工作正常！

Answer 1

这是一个如何在py_func的帮助下包装函数的示例。请注意，这在TF V2中已弃用。您可以按照文档获取更多详细信息。

def parse_function_wrapper(filename):
   # Assuming your data and labels are float32
   # Your input is parse_function, who arg is filename, and you get X and y as output
   # whose datatypes are indicated by the tuple argument  
   features, labels = tf.py_func(
       parse_function, [filename], (tf.float32, tf.float32)) 
   return features, labels

# Create dataset of filenames.
dataset = tf.data.Dataset.from_tensor_slices(flist)
dataset = dataset.shuffle(len(flist))
dataset = dataset.map(parse_function_wrapper)