tensorflow|tf.data与tf.feature_column 深度学习|tensorflow

1 tf.data 1.1 概述
? tf.data API 可以轻松处理大量数据、不同的数据格式以及复杂的转换。tf.data API 在 TensorFlow 中引入了两个新的抽象类：

tf.data.Dataset表示一系列元素，其中每个元素包含一个或多个Tensor对象。：
- 创建来源（例如 Dataset.from_tensor_slices()），以通过一个或多个 tf.Tensor 对象构建数据集。
- 应用转换（例如 Dataset.batch()），以通过一个或多个 tf.data.Dataset 对象构建数据集。
- dataset如果用于tf.estimator，必须是字典形式的feature, label
tf.data.Iterator 提供了从数据集中提取元素的主要方法。Iterator.get_next() 返回的操作会在执行时生成 Dataset 的下一个元素，并且此操作通常充当输入管道代码和模型之间的接口。

我们建议使用 TensorFlow 的 Dataset API，它可以解析各种数据。概括来讲，Dataset API 包含下列类：

文章图片

Dataset - 包含创建和转换数据集的方法的基类。可以通过该类从内存中的数据或 Python 生成器初始化数据集。

TextLineDataset - 从文本文件中读取行。

TFRecordDataset - 从 TFRecord 文件中读取记录。

FixedLengthRecordDataset - 从二进制文件中读取具有固定大小的记录。

Iterator - 提供一次访问一个数据集元素的方法。

1.2 使用机制

要启动输入管道，必须定义来源。
过内存中的某些张量构建 Dataset，可以使用 tf.data.Dataset.from_tensor_slices()
如果输入数据以推荐的 TFRecord 格式存储在磁盘上，可以构建tf.data.TFRecordDataset。
一旦有了 Dataset 对象，可以将其转换为新的 Dataset
Dataset.map()（为每个元素应用一个函数）
需要获取 Dataset 中的值
一次访问数据集中的一个元素（通过调用 Dataset.make_one_shot_iterator()）， Iterator.get_next()

1.3 tf.data.Dataset.from_tensor_slices获取数据

工作原理：将输入的张量的第一个维度看做样本的个数，沿其第一个维度将tensor切片，得到的每个切片是一个样本数据。实现了输入张量的自动切片。

一个数据集包含多个元素，每个元素的结构都相同。一个元素包含一个或多个 tf.Tensor 对象，这些对象称为组件。可以通过 Dataset.output_types 和 Dataset.output_shapes 属性检查数据集元素各个组件的推理类型和形状。

import tensorflow as tf import numpy as np## 测试1：输入是一个 tensor，函数将样本个数识别为8，然后对张量切片，每个样本的维度是（100） dataset_tensor = tf.data.Dataset.from_tensor_slices(tf.random_uniform([8, 100])) print("dataset_tensor.output_shapes = ", dataset_tensor.output_shapes)## 测试2：输入是一个 numpy dataset_numpy = tf.data.Dataset.from_tensor_slices(np.random.randn(8, 100)) print("dataset_numpy.output_shapes = ", dataset_numpy.output_shapes)## 测试3：输入是一个 dict:当不同的value-tensor的第一个维度不同时，会报错，无法对各张量统一切片 dataset_dict = tf.data.Dataset.from_tensor_slices( {"a": tf.random_uniform([8, 100]), "b": tf.random_uniform([8, 1000])}) print("dataset_dict.output_shapes = ", dataset_dict.output_shapes)## 测试4：输入是一个 tuple:当不同的 tensor元素的第一个维度不同时，会报错，无法对各张量统一切片 dataset_tuple = tf.data.Dataset.from_tensor_slices( (tf.ones([8, 10]), tf.zeros([8, 100]), tf.random_uniform([8, 15, 100])) ) print("dataset_tuple.output_shapes = ", dataset_tuple.output_shapes) ========================================================================== dataset_tensor.output_shapes =(100,) dataset_numpy.output_shapes =(100,) dataset_dict.output_shapes ={'a': TensorShape([Dimension(100)]), 'b': TensorShape([Dimension(1000)])} dataset_tuple.output_shapes =(TensorShape([Dimension(10)]), TensorShape([Dimension(100)]), TensorShape([Dimension(15), Dimension(100)]))

1.4 创建迭代器
构建了表示输入数据的 Dataset 后，下一步就是创建 Iterator 来访问该数据集中的元素。
1.4.1 单次Iterator 创建单次迭代器，非常的简单，只需要调用 Dataset 对象中的make_one_shot_iterator()方法。返回一个Iterator对象，调用 iterator 的 get_next() 就可以轻松地取出数据了。

import tensorflow as tf dataset = tf.data.Dataset.range(5) iterator = dataset.make_one_shot_iterator()with tf.Session() as sess: while True: try: print(sess.run(iterator.get_next())) except tf.errors.OutOfRangeError: break

单次的迭代器，不支持动态的数据集，它比较单纯，它不支持参数化。

1.4.2 可初始化的 Iterator

下面代码报错:
ValueError: Cannot capture a placeholder (name:Placeholder, type:Placeholder) by value.

import tensorflow as tfdef initialable_test(): numbers = tf.placeholder(tf.int64, shape=[]) dataset = tf.data.Dataset.range(numbers) iterator = dataset.make_one_shot_iterator()with tf.Session() as sess:while True: try: print(sess.run(iterator.get_next(), feed_dict={numbers: 5})) except tf.errors.OutOfRangeError: break

改写之后的代码:

import tensorflow as tfdef initialable_test(): numbers = tf.placeholder(tf.int64, shape=[]) dataset = tf.data.Dataset.range(numbers) # iterator = dataset.make_one_shot_iterator() iterator = dataset.make_initializable_iterator()with tf.Session() as sess:sess.run(iterator.initializer, feed_dict={numbers: 5}) while True: try: print(sess.run(iterator.get_next())) except tf.errors.OutOfRangeError: breaksess.run(iterator.initializer, feed_dict={numbers: 6}) while True: try: print(sess.run(iterator.get_next())) except tf.errors.OutOfRangeError: break

与单次Iterator不同的是:

创建的方式不同，iterator.make_initialnizer()
每次重新初始化的时候，都要调用sess.run(iterator.initializer)

1.4.3 可重新初始化的 Iterator

def reinitialable_iterator_test(): training_data = https://www.it610.com/article/tf.data.Dataset.range(10) validation_data = tf.data.Dataset.range(5)iterator = tf.data.Iterator.from_structure(training_data.output_types, training_data.output_shapes)train_op = iterator.make_initializer(training_data) validation_op = iterator.make_initializer(validation_data)next_element = iterator.get_next()with tf.Session() as sess:for _ in range(3): sess.run(train_op) for _ in range(3): print(sess.run(next_element))sess.run(validation_op) for _ in range(2): print(sess.run(next_element))

1.4.4 可馈送的 Iterator 馈送的 Iterator 一定程度上可以解决重复的代码，同时又将训练集和验证集的操作清晰得分离开来。

def feeding_iterator_test(): train_data = https://www.it610.com/article/tf.data.Dataset.range(100).map( lambda x: x + tf.random_uniform([], 0, 10, tf.int64) ) val_data = tf.data.Dataset.range(5) # 通过一个 string 类型的 handle 实现的。返回的是一个 Tensor handle = tf.placeholder(tf.string, shape=[])iterator = tf.data.Iterator.from_string_handle( handle, train_data.output_types, train_data.output_shapes)next_element = iterator.get_next()train_op = train_data.make_one_shot_iterator() validation_op = val_data.make_initializable_iterator()with tf.Session() as sess: train_iterator_handle = sess.run(train_op.string_handle()) val_iterator_handle = sess.run(validation_op.string_handle())for _ in range(3):for _ in range(2): print(sess.run(next_element, feed_dict={handle: train_iterator_handle}))sess.run(validation_op.initializer)for _ in range(5): print(sess.run(next_element, feed_dict={handle: val_iterator_handle}))

总结:

单次 Iterator ，它最简单，但无法重用，无法处理数据集参数化的要求。
可初始化的 Iterator ，它可以满足 Dataset 重复加载数据，满足了参数化要求。
可重新初始化的 Iterator，它可以对接不同的 Dataset，也就是可以从不同的 Dataset 中读取数据。
可馈送的 Iterator，它可以通过 feeding 的方式，让程序在运行时候选择正确的 Iterator,它和可重新初始化的 Iterator 不同的地方就是它的数据在不同的 Iterator 切换时，可以做到不重头开始读取数据。

1.5 读取输入数据
1.5.1 读取现有数组数据

features：Python 字典，其中：
- 每个键都是特征的名称。
- 每个值都是包含此特征所有值的数组。
label - 包含每个样本的标签值的数组。

features = {'SepalLength': np.array([6.4, 5.0]), 'SepalWidth':np.array([2.8, 2.3]), 'PetalLength': np.array([5.6, 3.3]), 'PetalWidth':np.array([2.2, 1.0])} labels = np.array([2, 1]) # 在函数中调用 def train_input_fn(features, labels, batch_size): dataset = tf.data.Dataset.from_tensor_slices((dict(features), labels)) dataset = dataset.shuffle(1000).repeat().batch(batch_size) return datasetdataset.make_one_shot_iterator().get_next()

1.5.2 读取CSV、文本数据**

filenames = ["1.txt", "2.txt"] # 默认情况下，TextLineDataset 会生成每个文件的每一行。 dataset = tf.data.TextLineDataset(filenames)

1.5.3 读取TFRecords 数据 1.5.3.1 什么是TFRecords文件

TFRecords其实是一种二进制文件，虽然它不如其他格式好理解，但是它能更好的利用内存，更方便复制和移动，并且不需要单独的标签文件。

TFRecords文件包含了tf.train.Example 协议内存块(protocol buffer)(协议内存块包含了字段 Features)。可以获取数据，将数据填入到Example协议内存块(protocol buffer)，将协议内存块序列化为一个字符串，并且通过tf.python_io.TFRecordWriter 写入到TFRecords文件。

文件格式 *.tfrecords

1.5.3.2 Example结构解析 tf.train.Example 协议内存块(protocol buffer)(协议内存块包含了字段 Features)，Features包含了一个Feature字段，Features中包含要写入的数据、并指明数据类型。

example = tf.train.Example(features=tf.train.Features(feature={ "features": tf.train.Feature(bytes_list=tf.train.BytesList(value=https://www.it610.com/article/[features])),"label": tf.train.Feature(int64_list=tf.train.Int64List(value=https://www.it610.com/article/[label])), }))

tf.train.Example(features=None)
- 写入tfrecords文件
- features:tf.train.Features类型的特征实例
- return：example格式协议块
tf.train.Features(feature=None)
- 构建每个样本的信息键值对
- feature:字典数据,key为要保存的名字
- value为tf.train.Feature实例
- return:Features类型
tf.train.Feature(options)
- options：例如
  - bytes_list=tf.train. BytesList(value=https://www.it610.com/article/[Bytes])
  - int64_list=tf.train. Int64List(value=https://www.it610.com/article/[Value])
- 支持存入的类型如下
- tf.train.Int64List(value=https://www.it610.com/article/[Value])
- tf.train.BytesList(value=https://www.it610.com/article/[Bytes])
- tf.train.FloatList(value=https://www.it610.com/article/[value])

1.5.3.3 使用步骤第一步:生成TFRecord Writer

# path：TFRecord文件的存放路径； # option：TFRecordOptions对象，定义TFRecord文件保存的压缩格式； writer = tf.python_io.TFRecordWriter(path, options=None)

第二步:tf.train.Feature生成协议信息
一个协议信息特征是将原始数据编码成特定的格式，内层feature是一个字典值，它是将某个类型列表编码成特定的feature格式，而该字典键用于读取TFRecords文件时索引得到不同的数据，某个类型列表可能包含零个或多个值，列表类型一般有BytesList, FloatList, Int64List

tf.train.BytesList(value=https://www.it610.com/article/[value]) # value转化为字符串（二进制）列表 tf.train.FloatList(value=[value]) # value转化为浮点型列表 tf.train.Int64List(value=[value]) # value转化为整型列表

# 外层:tf.train.Features,内层:tf.train.Feature value是需要保存的值 features_extern = tf.train.Features( { "width": tf.train.Feature(int64_list=tf.train.Int64List(value=https://www.it610.com/article/[width])),"weights": tf.train.Feature(float_list=tf.train.FloatList(value=https://www.it610.com/article/[weights])),"image_raw": tf.train.Feature(bytes_list=tf.train.BytesList(value=https://www.it610.com/article/[image_raw])) } )

第三步:使用tf.train.Example将features编码数据封装成特定的PB协议格式

example = tf.train.Example(features_extern)

第四步，将example数据系列化为字符串

example_str = example.SerializeToString()

第五步，将系列化为字符串的example数据写入协议缓冲区

writer.write(example_str)

1.6 DataSet的转换
当使用Dataset.map()，Dataset.flat_map()，以及Dataset.filter()转换时，它们会对每个element应用一个function

dataset1 = dataset1.map(lambda x: ...)

1.7 训练的数据集大小指定
tf.data API 提供了两种主要方式来处理同一数据的多个epoch。要迭代数据集多个周期，最简单的方法是使用 Dataset.repeat() 转换。
如果 Dataset.repeat()中没有参数转换将无限次地重复输入。

# 需求:要创建一个将其输入重复 10 个周期的数据集 dataset = dataset.map(...) dataset = dataset.repeat(10) dataset = dataset.batch(32)# 随机重排数据 dataset = dataset.map(...) dataset = dataset.shuffle(buffer_size=10000) dataset = dataset.batch(32) dataset = dataset.repeat()

2. 特征处理tf.feature_colum 2.1 特征列

Estimator 的 feature_columns 参数来指定模型的输入。
特征列在输入数据（由input_fn返回）与模型之间架起了桥梁。

文章图片

本文档介绍了该模块中的 9 个函数。如下图所示，除了 bucketized_column 外的函数要么返回一个 Categorical Column 对象，要么返回一个 Dense Column 对象。

文章图片

1.数值列（tf.feature_column.numeric_column）
2.分桶列（tf.feature_column.bucketized_column）
3.分类标识列（tf.feature_column.categorical_column_with_identity）
4.分类词汇列（tf.feature_column.categorical_column_with_vocabulary_list 或者 tf.feature_column.categorical_column_with_vocabulary_file）
5.经过哈希处理的列（tf.feature_column.categorical_column_with_hash_bucket）
6.组合列（tf.feature_column.crossed_column）
7.指标列（tf.feature_column.indicator_column）
8.嵌入列（tf.feature_column.embedding_column）

Numeric column（数值列）

# tf.feature_column 有许多可选参数。如果不指定可选参数，将默认指定该特征列的数值类型为 tf.float32。 numeric_feature_column = tf.feature_column.numeric_column(key="SepalLength")
Bucketized column（分桶列）
以表示房屋建造年份的原始数据为例。我们并非以标量数值列表示年份，而是将年份分成下列四个分桶：

文章图片

日期范围	表示为…
< 1960 年	[1, 0, 0, 0]
>= 1960 年但 < 1980 年	[0, 1, 0, 0]
>= 1980 年但 < 2000 年	[0, 0, 1, 0]
>= 2000 年	[0, 0, 0, 1]

# 首先，将原始输入转换为一个numeric column numeric_feature_column = tf.feature_column.numeric_column("Year")# 然后，按照边界[1960,1980,2000]将numeric column进行bucket bucketized_feature_column = tf.feature_column.bucketized_column( source_column = numeric_feature_column, boundaries = [1960, 1980, 2000])

Categorical identity column（类别标识列）
【tensorflow|tf.data与tf.feature_column】输入的列数据就是为固定的离散值，假设您想要表示整数范围 [0, 4)。在这种情况下，分类标识映射如下所示：

文章图片

identity_feature_column = tf.feature_column.categorical_column_with_identity( key='my_feature_b', num_buckets=4) # Values [0, 4)
Categorical vocabulary column（类别词汇表）
将字符串映射为数值或类别值。Categorical vocabulary column 可以将字符串表示为one_hot格式的向量。

文章图片

vocabulary_feature_column = tf.feature_column.categorical_column_with_vocabulary_list( key=feature_name_from_input_fn, vocabulary_list=["kitchenware", "electronics", "sports"])
Hashed Column（哈希列）

hashed_feature_column = tf.feature_column.categorical_column_with_hash_bucket( key = "some_feature", hash_bucket_size = 100) # The number of categories