Keras的fit_generator()模型方法需要一个生成形状元组（输入，目标）的生成器，其中两个元素都是NumPy数组。 The documentation似乎暗示如果我只是将Dataset iterator包装在一个生成器中，并确保将Tensors转换为NumPy数组，我应该很高兴。但是，这段代码给了我一个错误：
import numpy as np
import os
import keras.backend as K
from keras.layers import Dense, Input
from keras.models import Model
import tensorflow as tf
from tensorflow.contrib.data import Dataset
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
with tf.Session() as sess:
def create_data_generator():
dat1 = np.arange(4).reshape(-1, 1)
ds1 = Dataset.from_tensor_slices(dat1).repeat()
dat2 = np.arange(5, 9).reshape(-1, 1)
ds2 = Dataset.from_tensor_slices(dat2).repeat()
ds = Dataset.zip((ds1, ds2)).batch(4)
iterator = ds.make_one_shot_iterator()
while True:
next_val = iterator.get_next()
yield sess.run(next_val)
datagen = create_data_generator()
input_vals = Input(shape=(1,))
output = Dense(1, activation='relu')(input_vals)
model = Model(inputs=input_vals, outputs=output)
model.compile('rmsprop', 'mean_squared_error')
model.fit_generator(datagen, steps_per_epoch=1, epochs=5,
verbose=2, max_queue_size=2)
这是我得到的错误：
Using TensorFlow backend.
Epoch 1/5
Exception in thread Thread-1:
Traceback (most recent call last):
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 270, in __init__
fetch, allow_tensor=True, allow_operation=True))
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/framework/ops.py", line 2708, in as_graph_element
return self._as_graph_element_locked(obj, allow_tensor, allow_operation)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/framework/ops.py", line 2787, in _as_graph_element_locked
raise ValueError("Tensor %s is not an element of this graph." % obj)
ValueError: Tensor Tensor("IteratorGetNext:0", shape=(?, 1), dtype=int64) is not an element of this graph.
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/home/jsaporta/anaconda3/lib/python3.6/threading.py", line 916, in _bootstrap_inner
self.run()
File "/home/jsaporta/anaconda3/lib/python3.6/threading.py", line 864, in run
self._target(*self._args, **self._kwargs)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/keras/utils/data_utils.py", line 568, in data_generator_task
generator_output = next(self._generator)
File "./datagen_test.py", line 25, in create_data_generator
yield sess.run(next_val)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 895, in run
run_metadata_ptr)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 1109, in _run
self._graph, fetches, feed_dict_tensor, feed_handles=feed_handles)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 413, in __init__
self._fetch_mapper = _FetchMapper.for_fetch(fetches)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 233, in for_fetch
return _ListFetchMapper(fetch)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 340, in __init__
self._mappers = [_FetchMapper.for_fetch(fetch) for fetch in fetches]
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 340, in <listcomp>
self._mappers = [_FetchMapper.for_fetch(fetch) for fetch in fetches]
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 241, in for_fetch
return _ElementFetchMapper(fetches, contraction_fn)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/tensorflow/python/client/session.py", line 277, in __init__
'Tensor. (%s)' % (fetch, str(e)))
ValueError: Fetch argument <tf.Tensor 'IteratorGetNext:0' shape=(?, 1) dtype=int64> cannot be interpreted as a Tensor. (Tensor Tensor("IteratorGetNext:0", shape=(?, 1), dtype=int64) is not an element of this graph.)
Traceback (most recent call last):
File "./datagen_test.py", line 34, in <module>
verbose=2, max_queue_size=2)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/keras/legacy/interfaces.py", line 87, in wrapper
return func(*args, **kwargs)
File "/home/jsaporta/anaconda3/lib/python3.6/site-packages/keras/engine/training.py", line 2011, in fit_generator
generator_output = next(output_generator)
StopIteration
奇怪的是，在我初始化next(datagen)之后直接添加包含datagen的行会导致代码运行正常，没有错误。
为什么我的原始代码不起作用？当我将该行添加到我的代码中时，为什么它开始工作？是否有更有效的方法将TensorFlow的数据集API与Keras一起使用，而不涉及将Tensors转换为NumPy阵列并再次返回？
确实有一种更有效的方法来使用Dataset而不必将张量转换为numpy数组。但是，官方文档上还没有（还是？）。从发行说明中，它是Keras 2.0.7中引入的一项功能。您可能必须安装keras> = 2.0.7才能使用它。
x = np.arange(4).reshape(-1, 1).astype('float32')
ds_x = Dataset.from_tensor_slices(x).repeat().batch(4)
it_x = ds_x.make_one_shot_iterator()
y = np.arange(5, 9).reshape(-1, 1).astype('float32')
ds_y = Dataset.from_tensor_slices(y).repeat().batch(4)
it_y = ds_y.make_one_shot_iterator()
input_vals = Input(tensor=it_x.get_next())
output = Dense(1, activation='relu')(input_vals)
model = Model(inputs=input_vals, outputs=output)
model.compile('rmsprop', 'mse', target_tensors=[it_y.get_next()])
model.fit(steps_per_epoch=1, epochs=5, verbose=2)
几个区别：
tensor参数提供给Input层。 Keras将从此张量读取值，并将其用作适合模型的输入。target_tensors提供Model.compile()论证。float32。在正常使用情况下，Keras会为您完成此转换。但现在你必须自己做。Dataset的构造期间指定。使用steps_per_epoch和epochs来控制何时停止模型拟合。简而言之，如果要从张量读取数据，请使用Input(tensor=...)，model.compile(target_tensors=...)和model.fit(x=None, y=None, ...)。
tf.data.Dataset对象直接传递到keras.Model.fit()，它的行为类似于fit_generator。# Load mnist training data
(x_train, y_train), _ = tf.keras.datasets.mnist.load_data()
training_set = tfdata_generator(x_train, y_train,is_training=True)
model = # your keras model here
model.fit(
training_set.make_one_shot_iterator(),
steps_per_epoch=len(x_train) // 128,
epochs=5,
verbose = 1)
tfdata_generator是一个返回可迭代tf.data.Dataset的函数。def tfdata_generator(images, labels, is_training, batch_size=128):
'''Construct a data generator using `tf.Dataset`. '''
def map_fn(image, label):
'''Preprocess raw data to trainable input. '''
x = tf.reshape(tf.cast(image, tf.float32), (28, 28, 1))
y = tf.one_hot(tf.cast(label, tf.uint8), _NUM_CLASSES)
return x, y
dataset = tf.data.Dataset.from_tensor_slices((images, labels))
if is_training:
dataset = dataset.shuffle(1000) # depends on sample size
dataset = dataset.map(map_fn)
dataset = dataset.batch(batch_size)
dataset = dataset.repeat()
dataset = dataset.prefetch(tf.contrib.data.AUTOTUNE)
return dataset
除了@Yu-Yang的回答，你还可以修改tf.data.Dataset成为fit_generator的生成器如下
from tensorflow.contrib.learn.python.learn.datasets import mnist
data = mnist.load_mnist()
model = # your Keras model
model.fit_generator(generator = tfdata_generator(data.train.images, data.train.labels),
steps_per_epoch=200,
workers = 0 , # This is important
verbose = 1)
def tfdata_generator(images, labels, batch_size=128, shuffle=True,):
def map_func(image, label):
'''A transformation function'''
x_train = tf.reshape(tf.cast(image, tf.float32), image_shape)
y_train = tf.one_hot(tf.cast(label, tf.uint8), num_classes)
return [x_train, y_train]
dataset = tf.data.Dataset.from_tensor_slices((images, labels))
dataset = dataset.map(map_func)
dataset = dataset.shuffle().batch(batch_size).repeat()
iterator = dataset.make_one_shot_iterator()
next_batch = iterator.get_next()
while True:
yield K.get_session().run(next_batch)
@Yu_Yang和@Dat-Nguyen的解决方案都运行良好。通过使用可馈送迭代器并将验证集的句柄作为验证“数据”传递，也可以在训练期间制作@Yu-Yang的解决方案支持验证集。这有点令人费解，但它确实有效。
您还可以将Keras模型转换为Estimator，它们支持数据集：
estimator = tf.keras.estimator.model_to_estimator(keras_model=model,
model_dir=model_dir)
input_name = model.layers[0].input.op.name
def input_fn(dataset):
dataset = dataset.map(lambda X,y: {input_name: X}, y)
return dataset.make_one_shot_iterator().get_next()
train_spec = tf.estimator.TrainSpec(
input_fn=lambda: input_fn(train_set), max_steps=100)
eval_spec = tf.estimator.EvalSpec(
input_fn=lambda: input_fn(test_set))
tf.estimator.train_and_evaluate(estimator, train_spec, eval_spec)
其他答案都很好，但重要的是要注意直接使用大型numpy数组的from_tensor_slices可以快速填满你的记忆，因为，IIRC，这些值被复制到图中，如tf.constants。根据我的经验，这将导致无声的失败，最终训练将开始，但不会出现损失等方面的改善。
更好的方法是使用占位符。例如。这是我为代码创建生成器及其onehot目标的代码：
def create_generator_tf_dataset(self, images, onehots, batch_size):
# Get shapes
img_size = images.shape
img_size = (None, img_size[1], img_size[2], img_size[3])
onehot_size = onehots.shape
onehot_size = (None, onehot_size[1])
# Placeholders
images_tensor = tf.placeholder(tf.float32, shape=img_size)
onehots_tensor = tf.placeholder(tf.float32, shape=onehot_size)
# Dataset
dataset = tf.data.Dataset.from_tensor_slices((images_tensor, onehots_tensor))
# Map function (e.g. augmentation)
if map_fn is not None:
dataset = dataset.map(lambda x, y: (map_fn(x), y), num_parallel_calls=tf.data.experimental.AUTOTUNE)
# Combined shuffle and infinite repeat
dataset = dataset.apply(
tf.data.experimental.shuffle_and_repeat(len(images), None))
dataset = dataset.batch(batch_size)
dataset = dataset.prefetch(1)
# Make the iterator
iterator = dataset.make_initializable_iterator()
init_op = iterator.initializer
next_val = iterator.get_next()
with K.get_session().as_default() as sess:
sess.run(init_op, feed_dict={images_tensor: images, onehots_tensor: onehots})
while True:
inputs, labels = sess.run(next_val)
yield inputs, labels
如果您使用Pandas库创建TensorFlow数据集，这是一个解决方案。请注意，如果没有tf.reshape()，此代码将无效，因为由于某种原因，来自tf.py_func()的张量没有形状信息。所以这不适用于tuple。有人有解决方法吗？
def _get_input_data_for_dataset(file_name):
df_input=pd.read_csv(file_name.decode(),usecols=['Wind_MWh'])
X_data = df_input.as_matrix()
return X_data.astype('float32', copy=False)
X_dataset = tf.data.Dataset.from_tensor_slices(file_names)
X_dataset = X_dataset.flat_map(lambda file_name: tf.data.Dataset.from_tensor_slices(
tf.reshape(tf.py_func(_get_input_data_for_dataset,[file_name], tf.float32),[-1,1])))
X_dataset = X_dataset.batch(5)
X_iter = X_dataset.make_one_shot_iterator()
X_batch = X_iter.get_next()
input_X1 = Input(tensor= X_batch ,name='input_X1')
y1 = Dense(units=64, activation='relu',kernel_initializer=tf.keras.initializers.Constant(1),name='layer_FC1')(input_X1)
我最近的经验中的一个重要观察是使用tf.keras而不是本地keras。对我有用> tf> 1.12。
希望它也可以帮助别人。