查看tf.data.Dataset API，有多种创建数据集的方式。我将概述四种方法 - 但你只需要实现其中一种。
我假设你的csv文件的每一行都是n_features浮点值，后跟一个单独的int值。
创建tf.data.Dataset
使用Dataset.from_generator包装Python生成器
开始的最简单方法是包装本地Python生成器。 这可能会导致性能问题，但对于您的目的可能没有问题。
def read_csv(filename):
with open(filename, 'r') as f:
for line in f.readlines():
record = line.rstrip().split(',')
features = [float(n) for n in record[:-1]]
label = int(record[-1])
yield features, label
def get_dataset():
filename = 'my_train_dataset.csv'
generator = lambda: read_csv(filename)
return tf.data.Dataset.from_generator(
generator, (tf.float32, tf.int32), ((n_features,), ()))
这种方法非常灵活，使您能够在不依赖 TensorFlow 的情况下测试生成器函数 (read_csv)。
支持tensorflow版本1.12+，TensorFlow Datasets是我创建数据集的新选择。它可以自动序列化数据，收集统计信息，并通过info和builder对象向您提供其他元数据。它还可以处理自动下载和提取，从而简化协作。
import tensorflow_datasets as tfds
class MyCsvDatasetBuilder(tfds.core.GeneratorBasedBuilder):
VERSION = tfds.core.Version("0.0.1")
def _info(self):
return tfds.core.DatasetInfo(
builder=self,
description=(
"My dataset"),
features=tfds.features.FeaturesDict({
"features": tfds.features.Tensor(
shape=(FEATURE_SIZE,), dtype=tf.float32),
"label": tfds.features.ClassLabel(
names=CLASS_NAMES),
"index": tfds.features.Tensor(shape=(), dtype=tf.float32)
}),
supervised_keys=("features", "label"),
)
def _split_generators(self, dl_manager):
paths = dict(
train='/path/to/train.csv',
test='/path/to/test.csv',
)
return [
tfds.core.SplitGenerator(
name=tfds.Split.TRAIN,
num_shards=10,
gen_kwargs=dict(path=paths['train'])),
tfds.core.SplitGenerator(
name=tfds.Split.TEST,
num_shards=2,
gen_kwargs=dict(cvs_path=paths['test']))
]
def _generate_examples(self, csv_path):
with open(csv_path, 'r') as f:
for i, line in enumerate(f.readlines()):
record = line.rstrip().split(',')
features = [float(n) for n in record[:-1]]
label = int(record[-1])
yield dict(features=features, label=label, index=i)
使用方法：
builder = MyCsvDatasetBuilder()
builder.download_and_prepare()
datasets = builder.as_dataset(as_supervised=True)
train_ds = datasets['train']
test_ds = datasets['test']
将基于索引的Python函数封装
以上方法的一个缺点是，使用大小为n的混洗缓冲区对生成的数据集进行混洗需要加载n个示例。 这将在您的管道中创建定期暂停（大型n）或导致潜在的差错混洗（小型n）。
def get_record(i):
return features, labels
def get_inputs(batch_size, is_training):
def tf_map_fn(index):
features, labels = tf.py_func(
get_record, (index,), (tf.float32, tf.int32), stateful=False)
features.set_shape((n_features,))
labels.set_shape(())
return features, labels
epoch_size = get_epoch_size()
dataset = tf.data.Dataset.from_tensor_slices((tf.range(epoch_size,))
if is_training:
dataset = dataset.repeat().shuffle(epoch_size)
dataset = dataset.map(tf_map_fn, (tf.float32, tf.int32), num_parallel_calls=8)
dataset = dataset.batch(batch_size)
dataset = dataset.prefetch(1)
features, labels = dataset.make_one_shot_iterator().get_next()
return features, labels
简而言之，我们只创建一个仅包含记录索引的数据集（或任何可以完全加载到内存中的小型记录ID）。然后在这个最小数据集上执行混洗/重复操作，接着通过 tf.data.Dataset.map 和 tf.py_func 将索引映射到实际数据。请参阅下面的 与Estimators一起使用 和 在隔离环境中进行测试 部分以获取用法说明。请注意，这需要按行访问数据，因此您可能需要从 csv 转换为其他格式。
TextLineDataset
您还可以直接使用 tf.data.TextLineDataset 读取 csv 文件。
def get_record_defaults():
zf = tf.zeros(shape=(1,), dtype=tf.float32)
zi = tf.ones(shape=(1,), dtype=tf.int32)
return [zf]*n_features + [zi]
def parse_row(tf_string):
data = tf.decode_csv(
tf.expand_dims(tf_string, axis=0), get_record_defaults())
features = data[:-1]
features = tf.stack(features, axis=-1)
label = data[-1]
features = tf.squeeze(features, axis=0)
label = tf.squeeze(label, axis=0)
return features, label
def get_dataset():
dataset = tf.data.TextLineDataset(['data.csv'])
return dataset.map(parse_row, num_parallel_calls=8)
parse_row 函数有些复杂，因为 tf.decode_csv 需要一个批次(batch)，如果在解析之前对数据集进行分批(batch)处理可以使其变得稍微简单一些。
def parse_batch(tf_string):
data = tf.decode_csv(tf_string, get_record_defaults())
features = data[:-1]
labels = data[-1]
features = tf.stack(features, axis=-1)
return features, labels
def get_batched_dataset(batch_size):
dataset = tf.data.TextLineDataset(['data.csv'])
dataset = dataset.batch(batch_size)
dataset = dataset.map(parse_batch)
return dataset
TFRecordDataset
你可以将csv文件转换为TFRecord文件，并使用TFRecordDataset。这里有一个详细的教程（链接）。
步骤1: 将csv数据转换为TFRecords数据。以下是示例代码（请参见上面的from_generator示例中的read_csv函数）。
with tf.python_io.TFRecordWriter("my_train_dataset.tfrecords") as writer:
for features, labels in read_csv('my_train_dataset.csv'):
example = tf.train.Example()
example.features.feature[
"features"].float_list.value.extend(features)
example.features.feature[
"label"].int64_list.value.append(label)
writer.write(example.SerializeToString())
这只需要运行一次。
步骤2：编写一个数据集，解码这些记录文件。
def parse_function(example_proto):
features = {
'features': tf.FixedLenFeature((n_features,), tf.float32),
'label': tf.FixedLenFeature((), tf.int64)
}
parsed_features = tf.parse_single_example(example_proto, features)
return parsed_features['features'], parsed_features['label']
def get_dataset():
dataset = tf.data.TFRecordDataset(['data.tfrecords'])
dataset = dataset.map(parse_function)
return dataset
使用估算器处理数据集
def get_inputs(batch_size, shuffle_size):
dataset = get_dataset()
dataset = dataset.shuffle(shuffle_size)
dataset = dataset.repeat()
dataset = dataset.batch(batch_size)
dataset = dataset.prefetch(1)
features, label = dataset.make_one_shot_iterator().get_next()
estimator.train(lambda: get_inputs(32, 1000), max_steps=1e7)
隔离测试数据集
我强烈建议你独立地测试你的数据集而不是和估算器一起测试。使用上面的get_inputs，这非常简单。
batch_size = 4
shuffle_size = 100
features, labels = get_inputs(batch_size, shuffle_size)
with tf.Session() as sess:
f_data, l_data = sess.run([features, labels])
print(f_data, l_data) # or some better visualization function
性能
假设您正在使用GPU运行网络，除非csv文件的每一行都很大且您的网络很小，否则您可能不会注意到性能差异。这是因为Estimator实现强制要求在CPU上执行数据加载/预处理，而prefetch意味着在当前批次在GPU上训练时，下一批可以在CPU上准备好。唯一的例外是如果您的数据集具有大量数据记录且洗牌大小很大，则需要在运行任何内容通过GPU之前，花费一些时间来载入若干个样本。
